如何在使用 MapReduce API 映射到云存储之前过滤数据存储数据？

Posted 2023-03-24

【中文标题】如何在使用 MapReduce API 映射到云存储之前过滤数据存储数据？

【英文标题】：How to filter datastore data before mapping to cloud storage using the MapReduce API?

【发布时间】：2012-08-07 15:42:53

【问题描述】：

关于代码实验室here，我们如何过滤 mapreduce 作业中的数据存储数据，而不是获取特定实体类型的所有对象？

在下面的映射器管道定义中，唯一的一个输入读取器参数是要处理的实体类型，我在 InputReader 类中看不到其他过滤器类型的参数可以提供帮助。

output = yield mapreduce_pipeline.MapperPipeline(
  "Datastore Mapper %s" % entity_type,
  "main.datastore_map",
  "mapreduce.input_readers.DatastoreInputReader",
  output_writer_spec="mapreduce.output_writers.FileOutputWriter",
  params=
      "input_reader":
          "entity_kind": entity_type,
          ,
      "output_writer":
          "filesystem": "gs",
          "gs_bucket_name": GS_BUCKET,
          "output_sharding":"none",
          
      ,
      shards=100)

由于 Google BigQuery 在非规范化数据模型上表现得更好，因此能够从多个数据存储实体类型 (JOIN) 构建一个表会很好，但我也看不出该怎么做？

【参考方案1】：

根据您的应用程序，您可以通过传递一个过滤器参数来解决此问题，该参数是“应用于查询的可选过滤器列表。每个过滤器都是一个元组：(<property_name_as_str>, <query_operation_as_str>, <value>。”

所以，在您的输入阅读器参数中：

"input_reader":
          "entity_kind": entity_type,
          "filters": [("datastore_property", "=", 12345),
                      ("another_datastore_property", ">", 200)]

【讨论】：

谢谢迈克尔。 1/ 您需要在 map_reduce 上进行svn update，因为根据 svn 日志，该功能已于 8 月 1 日推出。

2/ 现在似乎有问题，因为您的元组列表将转换为引发 BadReaderParamsError 异常的列表列表 Filter should be a tuple（即@987654325 @ 变为 [["datastore_property", "=", 12345],["another_datastore_property", ">", 200]])。仍在寻找原因。

好像this issue其实已经被举报了