Почему мой конвейер потока данных Apache Beam не записывает данные в BigQuery? ⇐ Python

[Anonymous]

Я работаю над конвейером Apache Beam, который обрабатывает данные и записывает их в BigQuery. Конвейер отлично работает при использовании DirectRunner, но когда я переключаюсь на DataflowRunner, он завершается без ошибок и предупреждений, но не вставляет никаких строк в BigQuery. Кроме того, я вижу большие оставшиеся файлы во временном каталоге моего сегмента Cloud Storage (

Код: Выделить всё

gs://my-bucket/temp/bq_load/...

), а в целевой таблице данные не отображаются.
Вот структура конвейера:

Код: Выделить всё

worker_options.sdk_container_image = '...'

with beam.Pipeline(options=pipeline_options) as p:
processed_data = (
p
| "ReadFiles" >> beam.Create(FILE_LIST)
| "ProcessFiles" >> beam.ParDo(ProcessAvroFileDoFn())
| "WriteToBigQuery" >> beam.io.WriteToBigQuery(
table=f"{PROJECT_ID}:{DATASET_ID}.{TABLE_ID}",
schema=BQ_SCHEMA,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
)
)

Основные наблюдения:

• Конвейер успешно работает с DirectRunner, записывая данные в BigQuery без каких-либо проблем.
  При использовании DataflowRunner конвейер завершается без ошибок и предупреждений, но: строки не записываются в BigQuery, а большие временные файлы остаются в корзине (например, bq_load/...).
• Обрабатываемые данные допустим NDJSON.
• Схема BigQuery соответствует структуре данных.

Что я пробовал:

• Проверяя оставшиеся временные файлы, я загрузил временный файл и убедился, что он содержит допустимые строки NDJSON. Загрузка этого файла в BigQuery вручную с помощью команды bq load работает нормально.
• Тестирование с другими наборами данных:
  Я пробовал много разных входных данных. , но проблема остается.
• Проверка журналов потока данных:
  Я просмотрел журналы в консоли мониторинга потока данных, но не обнаружил ошибок. или предупреждения.
• Другая учетная запись службы: учетная запись службы с недостаточными разрешениями на поток данных выдает ошибку. Поэтому маловероятно, что проблема заключается в разрешениях для работников.

Я видел еще одну ветку об этом (можно не заставляю Apache записывать выходные данные в bigquery при использовании DataflowRunner), но там ничего не решено.

Подробнее здесь: https://stackoverflow.com/questions/793 ... o-bigquery