Google Cloud Datastore：导出与特定查询匹配的实体

Question

背景

我需要向大约 100 万台设备发送大量通知，我正在使用 Google Cloud Functions 构建它。

在当前设置中，我将每个设备令牌作为一个 PubSub 消息排队：

• 在 DataStore 中存储待处理通知，用于跟踪重试和成功状态

• 尝试发送通知

• 如果重试次数足够多且未通过，则将通知标记为成功或失败

此过程由人工上传带有所有令牌的 CSV 文件手动启动。原则上内置的重试应该足够了，但我想确保如果云功能本身或 APNs/FCM 出现问题，我可以以与上传相同的格式返回所有失败令牌的 CSV 文件，以便当/如果用户认为这是一个好主意时，他们只能重试失败的那个。

我将通知作为作业的一部分运行，用于与通知的状态一起查询。为此，我设置了一个复合索引，job_id并对status所有匹配的通知运行查询，并希望将其作为文件流式传输给用户或将其存储在 Google Cloud Storage 中，以便用户可以从那里。

问题

假设接近总通知数量失败并且我想在一个文件中获取所有令牌，我的第一个实现只是迭代所有匹配的条目并构建结果。问题是，以这种方式检索它们时，每 100_000 个条目大约需要 1 分钟。对于接近所有通知的事情，我将超过 Cloud Functions 的最大超时时间。每个实体总共大约 300 个字节，这使得整个导出大约 300MB。我可能可以通过添加一个更大的索引将其减少到大约一半/三分之二的大小，让我只对我想要的字段进行投影。

我能想到的唯一替代方法是将通知分片以将整个组分成 100 个分片，创建 100 个文件，每个文件包含 10k 条通知，然后将它们全部下载并在用户尝试下载文件时将它们拼接在一起。

我发布这个问题的原因是，这感觉像是一个相对简单的问题，而且这个解决方案感觉比我预期的要复杂一些，所以我想我可能会遗漏一些东西。

问题

• 我是否缺少一种显而易见的、更简单的方法来实现我想要的？

• 分片是否只是进行此类事情的预期方式，我应该接受这种复杂性吗？

代码

为了清楚起见，这是我正在运行的代码片段，我只是迭代它返回的响应以生成输出。

def get_failures(job_id):
        query = client.query(kind = Notification.kind)
        query.add_filter('job_id', '=', str(job_id))
        query.add_filter('status', '=', "failure")
                return query.fetch()

Answer 1

这个问题的强大解决方案是使用 Google Dataflow。我目前使用它来完成此操作，在 Google Cloud Storage 中生成 csv 文件，其中包含与给定数据存储查询匹配的所有约 500k 条记录。不过，设置它可能有点复杂。在开始之前，我使用了 10 分钟超时而不是 30 秒超时的 Google 任务队列。我不确定您是否可以纯粹在云函数中执行此操作，或者您是否需要创建一个简单的应用引擎项目来充当这些任务的请求处理程序