2

我有一个包含小消息的大型文件（4-5 GB 压缩），我希望按消息类型将其解析为大约 6,000 个文件。消息很小；根据类型，从 5 到 50 个字节不等。

每条消息都以一个固定大小的类型字段（一个 6 字节的密钥）开始。如果我读取“000001”类型的消息，我想将其有效负载附加到 000001.dat 等。输入文件包含消息的混合；我想要 N 个同质输出文件，其中每个输出文件仅包含给定类型的消息。

将这些消息写入这么多单个文件的有效快速方法是什么？我想使用尽可能多的内存和处理能力来尽快完成它。我可以将压缩或未压缩的文件写入磁盘。

我正在考虑使用带有消息类型键和输出流值的哈希图，但我确信有更好的方法来做到这一点。

谢谢！

4

6 回答 6

4

您可能不需要哈希映射。你可以只是...

  1. 阅读消息
  2. 以追加模式打开新文件
  3. 将消息写入新文件
  4. 关闭新文件

不确定这是否会更快，因为你会做很多打开和关闭。

于 2009-12-30T20:28:52.040 回答
4

类 Unix 系统通常会限制在任何给定时间打开的文件句柄的数量。例如，在我的 Linux 上，它当前为 1024，尽管我可以在合理范围内更改它。但是这些限制是有充分理由的，因为打开的文件是系统的负担。

您还没有回答我关于您的输入中是否多次出现同一个键的问题，这意味着可能需要将几批单独的数据连接到每个文件中。如果不是这种情况，佩斯的回答将是你能做的最好的，因为所有这些工作都需要完成，围绕如此简单的一系列事件建立一个庞大的管理部门是没有意义的。

但是，如果您的输入中有多个消息用于同一个键，那么保持大量文件打开会很有效。不过，我建议不要尝试同时打开所有 6000 个。相反，我会选择 500 份，先到先得。即，您为前 500 个（左右）不同的消息键打开文件，然后仔细阅读整个输入文件，寻找要添加到这 500 个中的内容，然后在输入 EOF 时将它们全部关闭。您还需要保留一个HashSet已处理的键，因为然后您会继续重新读取输入文件，处理下一批您在第一轮中没有捕获的 500 个键。

理由：打开和关闭文件（通常）是一项昂贵的操作；如果您能提供帮助，您不想多次打开和关闭数千个文件。因此，您尽可能多地保持打开状态，所有这些最终都会在您的输入中一次填充。另一方面，通过单个输入文件顺序流式传输非常有效，即使您必须通过输入文件进行 12 次传递，与打开/关闭 6000 个其他文件所需的时间相比，这样做的时间几乎可以忽略不计文件。

伪代码：

processedSet = [ ]
keysWaiting = true
MAXFILE = 500
handlesMap = [ ]
while (keysWaiting) {
  keysWaiting = false
  open/rewind input file
  while (not EOF(input file)) {
    read message
    if (handlesMap.containsKey(messageKey)) {
       write data to handlesMap.get(messageKey)
    } else if (processedSet.contains(messageKey) {
       continue // already processed
    } else if (handlesMap.size < MAXFILE) {
       handlesMap.put(messageKey, new FileOutputStream(messageKey + ".dat")
       processedSet.add(messageKey)
       write data to handlesMap.get(messageKey)
    else
       keysWaiting = true
    endif
  }
  for all handlesMap.values() {
     close file handle
  }
  handlesMap.clear
}
于 2009-12-30T20:49:16.533 回答
2

我推荐某种智能池：保持打开最大/最常用的文件以提高性能并关闭其余文件以节省资源。

如果主文件主要由记录类型 1-5 组成，请在需要时保持这些文件打开。其他的可以根据需要打开和关闭，这样您就不会饿死系统的资源。

于 2009-12-30T20:35:04.273 回答
1

我将对你的问题做一些假设：

  • 每条消息都以消息类型开头，作为一个固定大小的字段
  • 您有一个异构输入文件，其中包含混合消息；您需要 N 个同质输出文件，其中每个输出文件仅包含给定类型的消息。

想到的方法是基于函子的：您创建消息类型到处理该特定消息的对象的映射。您的 main() 是一个调度循环，它读取固定的消息头，从映射中找到适当的函子，然后调用它。

您可能无法同时打开 6,000 个文件（每种消息类型一个）；大多数操作系统都有大约 1,024 个同时打开文件的限制（尽管使用 Linux，您可以更改控制它的内核参数）。所以这意味着您将反复打开和关闭文件。

可能最好的方法是在每个仿函数上设置一个固定计数的缓冲区，以便它在 10 条消息之后打开、写入和关闭。如果您的消息最大为 50 字节，那么在任何给定时间将保留在内存中的 500 字节 (10 x 50) x 6,000。

我可能会编写我的仿函数来保存固定大小的字节数组，并创建一个通用仿函数类，一次将 N 个字节读取到该数组中：

public class MessageProcessor
{
    int _msgSize;                   // the number of bytes to read per message
    byte[] _buf = new byte[1024];   // bigger than I said, but it's only 6 Mb total
    int _curSize;                   // when this approaches _buf.length, write
于 2009-12-30T20:49:54.677 回答
0

由于您正在对许多文件进行许多小型写入，因此您希望尽量减少写入次数，特别是考虑到最简单的设计几乎可以保证每次新写入都会涉及新文件的打开/关闭。

相反，为什么不将每个键映射到缓冲区呢？最后，将每个缓冲区写入磁盘。或者，如果您担心会占用太多内存，您可以构建缓冲区以写入每 1K、5K 或任何行。例如

public class HashLogger {

          private HashMap<String,MessageBuffer> logs;

          public void write(String messageKey, String message)
          {
              if (!logs.contains(messageKey)) { logs.put(messageKey, new MessageBuffer(messageKey); }
              logs.get(messageKey).write(message);
          }

         public void flush()
         {
             for (MessageBuffer buffer: logs.values())
             {
                buffer.flush();
             }
            // ...flush all the buffers when you're done...
         }

    private class MessageBuffer {
             private MessageBuffer(String name){ ... }
             void flush(){ .. something here to write to a file specified by name ... }
             void write(String message){
             //... something here to add to internal buffer, or StringBuilder, or whatever... 
             //... you could also have something here that flushes if the internal builder gets larger than N lines ...
     }
}

您甚至可以创建单独的 Log4j 记录器，可以将其配置为使用缓冲日志记录，如果更现代的日志记录框架（如 slf4j）也不支持这一点，我会感到惊讶。

于 2009-12-30T20:38:04.093 回答
0

系统中打开的文件通常是有限制的，无论如何，以或多或少的随机顺序访问数千个小文件都会使您的系统陷入非常严重的困境。

考虑将大文件分解为单个消息的文件（或某种内存表，如果你有内存的话），并按消息类型对其进行排序。完成后，将消息写入相应的文件。

于 2009-12-30T20:53:50.103 回答