Блог

У меня есть поток данных заданий, который использует bigquery для чтения некоторых данных из pubsub для потоковой передачи (или gcs для пакетной передачи), а затем вставляет их в bigquery.

Для пакетной обработки я использую следующий шаг:

 (processed_rows | "Write Full to Bigquery" >> apache_beam.io.WriteToBigQuery(
    table_fn,
    schema=schema_fn,
    additional_bq_parameters=additional_bq_parameters_fn,
    write_disposition=apache_beam.io.BigQueryDisposition.WRITE_APPEND,
    create_disposition=apache_beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
    method=apache_beam.io.WriteToBigQuery.Method.FILE_LOADS
    )
 )
 

Для потоковой передачи у меня есть следующий:

 (processed_rows | "Write to Bigquery Streaming" >> apache_beam.io.WriteToBigQuery(
    table_fn,
    schema=schema_fn,
    additional_bq_parameters=additional_bq_parameters_fn,
    write_disposition=apache_beam.io.BigQueryDisposition.WRITE_APPEND,
    create_disposition=apache_beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
    method=apache_beam.io.WriteToBigQuery.Method.STREAMING_INSERTS)
 )
 

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

 def additional_bq_parameters_fn(table_reference_name):
    table_name = table_reference_name.split('.')[-1]
    bq_param = MAP_TABLE_BQ_PARAM[table_name]
    return bq_param
 

До сих пор он работал для пакетного задания, но не для потокового, где я получаю следующую ошибку:

Аргумент объекта _MessageClass после ** должен быть отображением, а не функцией

Когда я углубился в код луча apache, я нашел это в файлах BigQueryBatchFileLoads:

 if callable(self._additional_bq_parameters):
  additional_parameters = self._additional_bq_parameters(destination)
 

Однако я не могу найти ничего эквивалентного в потоковой части.

Итак, мой вопрос: я что-то пропустил или в коде apache beam есть ошибка (или неполная функция)?

Если я правильно понял, то я открою билет в apache beam Jira.