google-app-engine - App Engine 数据管道谈话 - 对于扇入物化 View ，为什么需要工作索引？

Question

我想了解在 google i/o 上的数据管道演讲: http://www.youtube.com/watch?v=zSDC_TU7rtc

如果我只是要通过输入序列标记进行批处理，我不明白为什么扇入工作索引是必要的。

乐观排队的任务不能获取所有未应用的标记，尽可能多地处理它们(重复获取一批，比如 10 个，然后事务性地更新物化 View 实体)，如果任务重新排队在完成所有标记之前超时？

工作索引是否与查询所有未应用标记的效率有关？即，查询“work_index = 的标记”比查询“applied = False 的标记”更好？如果是，那是为什么？

作为引用，引导我进行数据管道谈话的问题+答案如下: app engine datastore: model for progressively updated terrain height map

Answer 1

一些事情:

• 我的方法假定有多个工作线程(请参阅此处的 ShardedForkJoinQueue:http://code.google.com/p/pubsubhubbub/source/browse/trunk/hub/fork_join_queue.py)，其中任务的入站速率超过单个线程可以完成的工作量。考虑到这一点，您将如何使用简单的“applied = False”来跨 N 个线程拆分工作？可能将模型上的另一个字段随机分配给 worker 的 shard_number；那么您的查询将针对“shard_number=N AND applied=False”(需要另一个复合索引)。好的，应该可以。

• 但是你怎么知道你需要多少工作分片/线程？使用上述方法，您需要静态配置它们，以便您的 shard_number 参数介于 1 和 N 之间。您一次只能让一个线程查询每个 shard_number，否则就会发生争用。我希望系统在运行时计算出分片/线程数。我的方法将批处理一起工作成合理大小的 block (如 10 个项目)，然后将一个连续任务排入队列以处理其余部分。使用查询游标我知道每个延续都不会与最后一个线程重叠，所以没有争用。这为我提供了在同一个分片的工作项上并行工作的动态线程数。

• 现在假设您的队列备份。您如何确保首先处理最旧的工作项？换句话说:你如何防止饥饿？您可以将模型上的另一个字段分配给插入时间——称之为 add_time。现在您的查询将是“shard_number=N AND applied=False ORDER BY add_time DESC”。这适用于低吞吐量队列。

• 如果您的工作项写入率大幅上升怎么办？您将使用大致相同的 add_time 编写很多行。这需要您的实体有一个 Bigtable 行前缀，例如“shard_number=1|applied=False|add_time=2010-06-24T9:15:22”。这意味着每个工作项插入都命中相同 Bigtable 平板电脑服务器，该服务器当前是降序索引词法头的所有者。因此，从根本上说，对于每个工作分片的数据存储写入，您仅限于一台机器的吞吐量。

• 使用我的方法，您唯一的 Bigtable 索引行以递增工作序列号的哈希值作为前缀。每次递增序列号时，此 work_index 值分散在 Bigtable 的词法行空间中。因此，每个顺序工作项排队可能会进入不同的平板电脑服务器(给定足够的数据)，将我的队列负载分散到一台机器之外。使用这种方法，写入速率应该有效地仅受集群中物理 Bigtable 机器数量的限制。