python - 使用 apache beam/google 云数据流读取多行 JSON

Question

我正在尝试读取管道中的 JSON 文件(多行)，但 beam.io.ReadFromText(somefile.json 一次读取一行。

我正在尝试将文件内容读取为 JSON，以便我可以在每个类别上应用 map 来下载相关产品文件。

这就是我的 JSON 文件 (productindex.json) 的样子:

{
  "productcategories" : {
    "category1" : {
      "productfile" : "http://products.somestore.com/category1/products.json"
    },
    "category2" : {
      "productfile" : "http://products.somestore.com/category2/products.json"
    },
    "category3" : {
      "productfile" : "http://products.somestore.com/category3/products.json"
    },
    "category4" : {
      "productfile" : "http://products.somestore.com/category4/products.json"
    }
}

这就是我的管道开始的样子:

with beam.Pipeline(options=pipeline_options) as p:
    rows = (
        p | beam.io.ReadFromText(
            "http://products.somestore.com/allproducts/productindex.json")
    )

我正在使用apache-beam[gcp]模块。

如何实现这一目标？

Answer 1

Apache Beam/Cloud Dataflow 不直接支持读取多行 Json 数据。

主要原因是这很难并行完成。 Beam 如何知道每条记录的结束位置？这对于单个读者来说很容易，但对于并行读者来说非常复杂。

我推荐的最佳解决方案是在由 Beam/Dataflow 处理之前将 Json 数据转换为换行符分隔的 Json (NDJSON)。这可能就像更改上游任务写入的输出格式一样简单，或者可能需要预处理。