Чтение многострочного JSON с использованием потока данных Apache Beam / Google Cloud

Я пытаюсь прочитать файл 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].

Как мне этого добиться?


python apache-beam google-cloud-platform google-cloud-dataflow
person Asdfg    schedule 19.02.2019    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Apache Beam / Cloud Dataflow не поддерживает чтение многострочных данных Json напрямую.

Основная причина в том, что это очень сложно делать параллельно. Как Beam узнает, где заканчивается каждая запись? Это легко для одного читателя, но очень сложно для параллельных читателей.

Лучшее решение, которое я могу порекомендовать, - преобразовать ваши данные Json в Json с разделителями новой строки (NDJSON) перед обработкой с помощью Beam / Dataflow. Это может быть так же просто, как изменение формата вывода, записанного восходящей задачей, или может потребоваться предварительная обработка.

person John Hanley    schedule 19.02.2019