每小时高效地将数据从 Datastore 导入 BigQuery - Python答案

【问题标题】：Import Data Efficiently from Datastore to BigQuery every Hour - Python每小时高效地将数据从 Datastore 导入 BigQuery - Python
【发布时间】：2016-08-17 23:22:55
【问题描述】：

目前，我使用 Google 的两步法备份数据存储，然后将其导入 BigQuery。我还使用管道查看了代码。这两种方法效率不高，成本高，因为每次都要导入所有数据。我只需要添加上次导入添加的记录。

正确的做法是什么？是否有关于如何在 python 中执行此操作的工作示例？

【问题讨论】：

标签： python google-app-engine google-bigquery google-cloud-datastore

【解决方案1】：

您可以查看Streaming inserts。我现在实际上正在考虑在 Java 中做同样的事情。

如果您想每小时执行一次，您可以在每次将新实体放入 Datastore 时将插入添加到拉取队列（作为序列化实体或键/ID）。然后，您可以使用 cron 作业每小时处理一次队列。

【讨论】：

这是一个有趣的方向@tx802。我过去做过流式插入，但没有使用拉队列对插入进行分组，这是合理的。但是另一个选项呢 - 使用游标从数据存储中读取并使用它从上一个每小时 cron 调用中获取的最后一条记录开始。游标可以保存在 Memcache 或 Datastore 中。这可能会节省拉队列的需要。
在我的脑海中，我想知道光标是否有可能在一小时后过时。即，根据您用于迭代实体的查询，您是否有可能在光标当前指向的点“之前”创建新实体？（不确定我是否解释得很好。）

【解决方案2】：

没有完整的工作示例（据我所知），但我相信以下过程可以帮助您：

1- 您需要向您的实体添加“上次更改”，然后对其进行更新。

2- 您可以每小时运行一次 MapReduce 作业，您的映射器可以有一个过滤器来检查上次更新的时间，并且只选择在过去一小时内更新的实体

3- 手动添加需要添加到备份中的内容。

正如我所说，这是相当高的水平，但实际答案将需要一堆代码。老实说，我认为它不适合 Stack Overflow 的格式。

【讨论】：

如果在映射器中完成过滤，那么我仍然需要读取所有记录，这仍然是一个问题。根据 DatastoreInputReader 的文档：“此阅读器不进行过滤：您需要在映射器中进行任何所需的过滤。”。我在这里错过了什么吗？
哦，您根本不想阅读。我认为这是您想跳过的副本。是的，在这种情况下，tx 的解决方案可能更合适
我想防止额外读取已处理的记录，因为当您拥有大量数据存储实体时，它会显着增加价格。无论如何，我发现您的答案很有用，因为我确实在 DatastoreInputReader 源代码中看到了对过滤器实现的参考，但没有适当的正式文档，我无法触及它。如果有人能澄清这一点，我会很高兴。还是谢谢。
Mapreduce 文档已过时。如果您像这样指定映射器参数： params = {"input_reader" : { "entity_kind" : "my_entities.Blah", "filters" : filters } }, 其中过滤器是元组列表 filters.append(("disabled ", "=", False)) 那么它只会将匹配过滤器的实体发送给映射器。