Почему мой конвейер потока данных 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 соответствует структуре данных.

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

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

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

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