Почему мой Python BigQuery Dataflow не вставляет записи в базу данных?

Question

Я использую Python (2.7) и работаю в среде DataFlow Google, и, разумеется, Google еще не полностью очистил все, и документации пока недостаточно. Тем не менее, часть для записи из Dataflow в BigQuery приведена здесь BigQuery Sink.

Согласно документации, с тем чтобы определить схему, вам нужно ввести строку:

schema = 'field_1:STRING, field_2:STRING, field_3:STRING, created_at:TIMESTAMP, updated_at:TIMESTAMP, field_4:STRING, field_5:STRING' 

Имя таблицы, ID проекта и набор данных ID, как это: «example_project_id: example_dataset_id.example_table_name»

Теперь все это работает. См. Код ниже, но из того, что я вижу, он успешно создает таблицу и поля. Примечание. Идентификатор проекта задается как часть аргументов функции.

bq_data | beam.io.Write(
    "Write to BQ", beam.io.BigQuerySink(
     'example_dataset_id.{}'.format(bq_table_name), 
     schema=schema, 
     write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND, 
     create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED 
    ) 
) 

Теперь, похоже, что я могу получить все вставки с помощью этого:

bq_data = pipeline | beam.Create(
    [{ 
     'field_1': 'ExampleIdentifier', 
     'field_2': 'ExampleValue', 
     'field_3': 'ExampleFieldValue', 
     'created_at': '2016-12-26T05:50:39Z', 
     'updated_at': '2016-12-26T05:50:39Z', 
     'field_4': 'ExampleDataIdentifier', 
     'field_5: 'ExampleData' 
    }] 
) 

Но по какой-то причине, когда упаковка значения в PCollection, он говорит, что он вставляет в BigQuery, но когда Я запрашиваю таблицу, она ничего не показывает.

Почему он не вставляет? Я не вижу никаких ошибок, но ничего не вставляет в BigQuery.

Это то, что выглядит данные, как, содержащийся в PCollection, я близко к 1,100 строк для вставки:

{'field_1': 'ExampleIdentifier', 'field_2': 'ExampleValue', 'field_3': 'ExampleFieldValue', 'created_at': '2016-12-29 12:10:32', 'updated_at': '2016-12-29 12:10:32', 'field_4': 'ExampleDataIdentifier', 'field_5': 'ExampleData'} 

Примечание: Я проверил в форматирования даты и форматирование даты выше допускается для вставки BigQuery.

Answer 1

Я попробовал пример с вашей точной схемой и вводами, и это сработало для меня. Я должен был выполнить следующие исправления.

(1) Кажется, что вы не указываете проект в своих аргументах. Вы можете указать это в определении конвейера, так как вы не видите ошибку для этого. (2) В коде, указанном выше, есть опечатка. 'field_5: 'ExampleData' должен быть 'field_5': 'ExampleData' Но я предполагаю, что это только опечатка в этом вопросе не в исходном конвейере, так как вы не получаете ошибку для этого.

Вы используете последнюю версию Dataflow? Вы можете попробовать создать новую виртуальную среду и запустить «pip install google-cloud-dataflow» для установки последней версии.

Можно ли поделиться своим полным pipleine для меня, чтобы попробовать?

Трудно отлаживать это дистанционно, так как вы используете «DirectPipelineRunner». Можно ли попробовать использовать тот же конвейер, используя «DataflowPipelineRunner» (обратите внимание, что для этого вам понадобится проект GCP с включенным биллингом)? Я смогу просмотреть журналы, если вы можете запустить это с помощью «DataflowPipelineRunner» и предоставить идентификатор задания.