Я пытаюсь создать поток данных (пакетный), который считывает файл в час из облачного хранилища Google, анализирует его и записывает запись в таблицу BigQuery. Файл представляет собой .json, в котором каждая строка содержит сложный json.

Я создал простой конвейер:

 (p 
| "Read file" >> beam.io.ReadFromText(cusom_options.file_name)
| "Parse line json" >> beam.Map(parse)
| "Write in BigQuery" >> beam.io.gcp.bigquery.WriteToBigQuery(
        table=cusom_options.table))
  

Функция синтаксического анализа выглядит следующим образом:

 def parse(input_elem):
   log = json.loads(input_elem)
   result = {}

   ... # Get some fields from input and add to "result"

   return result
  

Конвейер отлично работает с размером файла 100 Мб и 70 Тыс. строк (примерно 5 минут на задание). Но когда файл увеличивается, поток данных занимает больше времени (15 минут, 200-300 Мб) или не завершается и завершается ошибкой (более 1,5 Гб и 350 Тыс. строк).

Я провел некоторый тест, когда я создал образец json в функции parse, но не использовал input_elem, поток данных работает нормально и создает строку для каждой записи за 7-8 минут.

Я не знаю, в чем проблема с конвейером, у кого-нибудь есть похожая проблема?

Подробная информация

• Я использую beam.io.gcp.bigquery.WriteToBigQuery для автоматического определения схемы из таблицы BQ, но если я использую beam.io.WriteToBigQuery, возникает та же проблема.
• Имя таблицы предварительно вычисляется вне потока данных и является входным
• Параметры, используемые в потоке данных —experiment=use_beam_bq_sink, —subnetwork=MY_SUBNET, —region=MY_REGION

Мы, наконец, решили проблему. Параллельно с работой с потоком данных в приложении были созданы некоторые сети VPC, и правила брандмауэра были неправильно настроены.

Этот случай похож на тот, который описан в документации (сеть VPC, используемая для вашей работы, может отсутствовать). Правила существовали, но не были правильно настроены

Спасибо!