在庞大的数据集上计算令牌计数器答案

【问题标题】：Computing token counters on huge dataset在庞大的数据集上计算令牌计数器
【发布时间】：2011-04-23 20:38:07
【问题描述】：

我需要检查大量文本（> 2 Tb，维基百科完整转储）并为每个看到的令牌保留两个计数器（每个计数器根据当前事件递增）。我需要对这些计数器进行的唯一操作是增加。在第二阶段，我应该根据这些计数器计算两个浮点数并存储它们。

它应该执行以下步骤：

检查大量文本并根据当前事件为找到的每个单词增加两个计数器。
检查所有标记，为每个标记，根据这些计数器计算两个额外的浮点数。
允许查询（获取任何给定令牌的值）。

要求和其他细节：

它必须扩展到 O(10^8) 个令牌。
需要非常快的查询最终结果！
查看文本时，只会增加两个计数器。这是一次性处理，因此在处理过程中不会有任何查询。仅值更新。
无需动态/可更新架构。

我一直在尝试 CouchDB 和 MongoDB，但效果不佳。

您认为解决这个问题的最佳方法是什么？

谢谢！

编辑 1： 有人建议我尝试 Patricia trie 并测试所有键是否都适合内存（我怀疑它们不适合）。一个带有额外操作符的自定义 Patricia trie 用于在一步中增加每个键的值可能是一种可能的解决方案。

编辑 2： 阐明了我所说的“巨大”的含义：> 2 Tb 的文本。更多说明。

编辑 3： 唯一令牌估计。正如 Mike Dunlavey 所建议的，我尝试对独特的令牌进行快速估计。在数据集的前 830Mb 中，唯一令牌线性增长到 52134。除非在处理更多数据后唯一令牌的数量增长较慢（这很可能），否则应该有 O(10^8) 个唯一令牌。

编辑 4： 首选 Java 和 Python 解决方案，但任何其他语言也可以。

编辑 5： 通常标记只包含可打印的 ASCII 字符，但它们可以包含任何可打印的 Unicode 字符。我将尝试相同的过程，同时保持小写和大写不变；并且仅适用于小写。

【问题讨论】：

您的数据目前是如何存储的？它已经在数据库中了吗？一个大文件？许多不是那么大的文件？您对令牌了解多少——您建议 O(10^12) 字节中的 O(10^9) 个令牌，有多少个唯一令牌？（组）令牌有哪些共同特征？
已更新。这是一个完整的维基百科转储。那是在 7z 中压缩的单个 XML，我将其提取为流并动态处理。我目前正在估计唯一令牌编号。您可以在标记中期望自然语言中存在的任何特征，以及一些其他标记，例如标记和标点符号。
我会尽力在周末完成这项工作。我会接受答案并在检查后分配赏金。
@Luca Martinetti：不，但它与语言模型部分相关。

标签： performance nosql large-data-volumes key-value-store

【解决方案1】：

您必须使用数据库而不是读取文本文件吗？

一个简单的 C 类型编译语言可以在读取文件所需时间的一小部分时间内运行一个简单的解析器，因此它应该基本上是“I/O 绑定”的。这将是一个类似于 unix wc 的程序，字数统计。

听起来数学很琐碎，甚至不应该引起注意。

编辑：好的，我不明白你想建立一个唯一标记的字典，并计算每个标记。在这种情况下，一个基于 trie 或哈希的字典就足够了。其存储大小将取决于令牌的典型长度以及有多少不同的令牌。这可能类似于 unix sort | uniq 成语。

【讨论】：

我不关心它是数据库、纯文本文件还是任何其他方法，只要它完成任务即可。但是，我看不到与 wc 的链接。
@smmv: wc 是一个简单的程序，可以读取文件并计算令牌。如果您愿意，获取它的源代码应该不会太难。在我看来，您的程序会很相似，但从头开始编写也不难。如果您想变得花哨，可以尝试异步重叠 I/O（双缓冲）。这会将它从主要是 I/O 绑定变为完全 I/O 绑定。
wc 只计算单词、字符或字节。我需要为 each 令牌保留计数器（取决于其类型）。对于 > 2 Tb 的文本，所有标记都无法按原样放入内存中。所以我怀疑 wc 方法是否相关。（对不起，如果我没有明确说明每个令牌都有计数器，而不是整体）。
@smmv：对。我误解了。谢谢。然后，如果我站在你的立场上，我会首先尝试在数据集上运行sort | uniq，看看字典实际上有多大。你说它是多 TB 的，但字典中唯一标记的数量一开始应该迅速增加，然后逐渐减少，最后增长非常缓慢，所以你可以只使用数据集的一小部分来估计字典大小。
谢谢，我正在估计，我会报告。

【解决方案2】：

一种策略，而不是一种解决方案；

没有一个进程对输入数据的通读转义，即我看不到如何并行化初始操作，除非文件位于并行 I/O 系统上，即使那样我认为这可能很困难并行处理 7z 文件。

但是，您可以尝试实现一个进程，该进程读取输入数据并在文件系统中写入数据块，最好是到足够多的不同磁盘上，这样您接下来要启动的进程不会全部排队等待相同的读/写头。

一旦第一个块被写入，你就在另一个核心上启动一个进程（你有多核不是吗？甚至可能是一个集群或工作站网络？）开始消化那个块。此过程将部分结果写入文件。

一旦第二个块被写入，你就在另一个核心上启动一个进程......

...你明白了

处理完整个输入后，您就可以设计任务来合并处理每个块的任务的输出结果。您可以在某种级联中执行此操作（例如，如果您有 32 个块和 16 个处理器，您可能每个合并 2 个块，然后其中 8 个合并了 2 个合并块，依此类推）。

我最好的猜测是，您应该可以使用平面文件，但不确定数据库的额外功能是否值得额外成本（在性能和编程复杂性方面）。我想您可能希望将最终结果写入数据库以支持查询。

编辑：好吧，如果您的所有查询都是“给我获取令牌 XXX 的计数器”的形式，那么您可以通过单个排序的文本文件进行二进制搜索。我并不是建议您这样做，但它可能会为您指明解决方案的方向。暂时忘记标记可能以任何字符开头（这只是字母表的问题），您可以有 26 个文件，一个用于以 A 开头的标记，一个用于以 B 开头的标记，依此类推。

或者您可以在主文件中构建一个索引，其中包含 A（从文件开头偏移 0）B（从开始偏移 12456）等条目。

我个人会尝试使用按首字母排序的文本文件的方法，直到找到可行的解决方案，然后确定它是否足够快。但是我可以访问具有大量磁盘和大量 RAM 的大型集群，您的平台可能会决定另一种可能更复杂的方法。

【讨论】：

谢谢。我只有一个只有 1Tb 磁盘的四核（即将推出第二个 1.5Tb）。最终，我正在使用一些体面的机器建立一个集群，但是当我得到那个集群时，这将超过我的截止日期;-)我已经考虑过这个策略......处理数据集不是所以有问题，而是以有效的方式存储结果（在处理期间和查询之后）。我很可能会保留部分结果并在以后合并它们，但我仍然没有得到完整的图片。
@smmv：告诉我们更多关于您想要支持的查询类型：成熟的 SQL（或类似的）或“查找令牌 XXX”。这将对处理数据的数据结构选择产生重大影响。
所有查询都将是“给我令牌 X 的计数器”。
@smmv：好吧，如果您需要指向 RDBMS 解决方案的 SQL 的全部功能。不是吗？或者，查询语言和数据结构的其他组合？
“给我令牌 X 的计数器”听起来不像“SQL 的全部力量”

【解决方案3】：

据我了解，您只想计算令牌。第一个解决方案可能只是在内存中使用哈希映射。 52-100k 个标记（英语单词的优势长度是 ca 5.1）+ 每个标记 4 个字节用于保持计数并不是那么多数据。您可以轻松地将地图存储在开发人员机器的内存中。

第二种解决方案是使用 apache lucene 来存储新令牌——除非你没有 1M 条目，否则你不需要对索引进行分区——以及我将存储在数据库中的计数器值，例如 sqllite （因为更新 lucene 索引不是最好的主意）。

为了加快这一过程——对于这两种解决方案——我只需将您的数据集拆分为 k*100 数据集，然后在不同的机器上（或并行）分别运行它们，然后合并它们的结果。您计算的结果，您可以毫无问题地求和。

您的用例是 apache hadoop 教程中的经典示例，但我认为部署它会过度设计。

【讨论】：

【解决方案4】：

如果你有很多内存，你可以使用普通的 redis 来存储计数器（我猜 10^8 个唯一令牌和两个计数器每个大约需要 12GB）。

如果你没有那么多内存，你仍然可以使用 redis，但需要一点散列策略和 vm_enabled 以使其适合内存：

您可以将令牌除以第一个和第二个字母（aa、ab、ac...zz）作为哈希名称，将实际单词 + 令牌标识符作为哈希键，将计数作为值。它看起来像这样：

hash ab
- absence_c1 5
- absence_c2 2
- abandon_c1 2
- abandon_c1 10
hash st
- stack_c1 10
- stack_c2 14

但是在这种方法中，因为 redis 不能在哈希上“增加”，你会得到以前的值，然后他们增加并设置它回来，这种方式（伪代码）：

var last = redis("hget st stack_c1")
var actual = last + 1
redis("hset st stack_c1 actual")

使用此哈希模式并启用 vm 后，redis 将保持较低的内存使用率，同时仍然足够快。我能够存储 200 万个令牌，每个令牌 15 个字符，使用更少的 100MB 内存和几乎 4G 的磁盘。

【讨论】：

谢谢。我继续用这个方法。我注意到还有“hincrby”，所以我可以在这些哈希表中增加一个值。我将使用第一个和最后一个字符（而不是两个第一个）来进行更稀疏的分布，并且我修改了标记器以进行一些简化，以减少唯一标记的数量。让我们看看这是否有效！

【解决方案5】：

高级解决方案：

解析输入，将“[token] +X +Y”行输出到 N 个输出文件中的 1 个（这些“分片”输出文件中的每一个都足够小，可以在内存中进行处理。）
[对于每个文件]将其读入内存，输出带有“[token] [count1] [count2] ...”行的排序文件
在查询时，对正确的文件进行二分搜索

详情：这是第 1 步的 Python 伪代码）

NUM_SHARDS = 1000  # big enough to make each file fit in memory  
output_files = [open("file" + str(n), "w") for n in xrange(NUM_SHARDS)]
for token in input_stream:
   shard_id = hash(token) % NUM_SHARDS
   output_files[shard_id].write(token + " +0 +1\n")
   # TODO: output the correct +X and +Y as needed

这是第 2 步的 Python 伪代码）

input_files = [open("file" + str(n)) for n in xrange(NUM_SHARDS)]
for file in input_files:
   counts = {}   # Key: token   Value: { "count1": 0, "count2": 1 }

   # read the file, and populate 'counts'
   for line in file:
      (token, count1, count2) = line.split(" ")
      # make sure we have a value for this token
      counts.setdefault(token, { "count1": 0, "count2": 0 })
      counts[token]["count1"] += int(count1)
      counts[token]["count2"] += int(count2)
      # TODO: compute those floats, and stuff those inside 'counts' also

   # now write 'counts' out to a file (in sorted order)
   output_file = open(file.name + ".index", "w")
   for token, token_counts in sorted(counts.items()):
      output_file.write(token + " " + token_counts["counts1"] + " " + token_counts["counts2"] + "\n")
      # TODO: also write out those floats in the same line

这里是步骤 3) 的一些 Python 代码：

# assume 'token' contains the token you want to find
shard_id = hash(token) % NUM_SHARDS
filename = "file" + str(shard_id) + ".index"
binary_search(token, open(filename), 0, os.path.getsize(filename))

# print out the line in 'file' whose first token is 'token'
# begin/end always point to the start of a line
def binary_search(token, file, begin, end):
    # If we're close, just do brute force
    if end - begin < 10000:
            file.seek(begin)
            while file.tell() < end:
                    line = file.readline()
                    cur_token = line.strip().split(" ")[0]
                    if cur_token == token:
                            print line
                            return True
            return False  # not found

    # If we're not close, pivot based on a line near the middle
    file.seek((begin + end) / 2)
    partial_line = file.readline()  # ignore the first fractional line
    line = file.readline()

    cur_token = line.strip().split(" ")[0]
    if cur_token == token:
            print line
            return True
    elif cur_token < token:
            return binary_search(token, file, file.tell(), end)
    else:  # cur_token > token
            return binary_search(token, file, begin, file.tell() - len(line))

【讨论】：

【解决方案6】：

好吧，如果 MongoDB 和 CouchDB 不适合你，那么你基本上有一个问题：没有足够的能力。

让我们看看洗衣清单：

它必须扩展到 O(10^8) 个令牌。

你有多少内存？您正在谈论数亿个令牌并且您正在谈论流式传输 7zip 文件。如果你想快速发出“增量”，你需要能够将整个数据结构保存在内存中，否则整个事情会非常缓慢。

需要非常快的查询最终结果！

有多快？微秒、毫秒、数百毫秒？如果你想在一台有 8GB RAM 的机器上查询 500M 条记录，那你就大错特错了。数据不适合，无论您使用什么数据库。

数据集 > 2Tb

好的，让我们假设您的计算机平均可以保持大约 50MB/秒的持续吞吐量并且您的 proc 实际上可以以这种速度解压缩数据。以这样的速度，您说的是 11 多个小时的处理时间只是为了流式传输数据（您希望在周末完成这项工作吗？）

11 小时 50MB/s 的吞吐量可不是小菜一碟，这是一个真正的驱动器。而且，如果您尝试在发生这种情况（或操作系统交换）时将任何内容写入磁盘，那么性能会迅速下降。

从数据库的角度看，MongoDB 可以同时处理前端更新和后端查询。但它需要每分钟左右刷新到磁盘，这将显着延长 11 小时的运行时间。

除非您可以处理内存中的整个数据库和内存中的整个流，否则总运行时间会越来越差。

我的观点...

很简单，你需要更多的力量。

如果您不是在 24GB 以上的 RAM 上运行此操作，那么您所做的一切都会感觉很慢。如果您没有 24GB 以上的 RAM，那么您的最终数据集将不会是“闪电般快速”，充其量是“200 ms-quick”。您可以索引 500M 行并期望找到一个条目，除非您可以在 RAM 中保留索引。

如果您不使用出色的 HDD 运行此操作，那么该操作会显得很慢。我的意思是，您说的是数小时的高吞吐量持续读取（可能还有写入）。

我知道你需要帮助，我知道你在这个问题上付出了很多，但是很难解决以下问题：

我一直在尝试 CouchDB 和 MongoDB，但效果不佳。

当你听起来还没有真正找到合适的工具来解决问题时。

【讨论】：