Python SDK Потоковый бегунок потока данных - PullRequest
Новая
       7

Python SDK Потоковый бегунок потока данных

0 голосов
/ 31 октября 2018

У меня есть поток данных, написанный на Python, который я пытаюсь запустить на GCP. Поток данных продолжает завершаться с:

Рабочий процесс не выполнен. Причины: неизвестный код сообщения.

Основной код в моем конвейере данных: 

schema = 'Member_ID:INTEGER,First_Name:STRING,Last_Name:STRING,Gender:STRING,Age:INTEGER,Height:STRING,weight:INTEGER,Hours_Sleep:INTEGER,Calories_Consumed:INTEGER,Calories_Burned:INTEGER,Evt_Date:DATE,Height_Inches:INTEGER,Min_Sleep_Hours:INTEGER,Max_Sleep_Hours:INTEGER,Enough_Sleep:BOOL'

# read, transform and local source data
p = beam.Pipeline(options=options)

# Read from PubSub into a PCollection.
events = (p | 'Read PubSub' >> beam.io.ReadFromPubSub (topic='projects/prefab-envoy-220213/topics/health_event')
            | 'Parse CSV' >> beam.ParDo(getCSVFields())
            | 'Convert Types' >> beam.ParDo(ConvDataTypes())
            | 'Convert Height' >> beam.ParDo(ConvHeight())
            | 'Join CDC Sleep' >> beam.ParDo(CDCSleepJoin(), cdcsleep)
            | 'Create Row' >> beam.ParDo(CreateRow()) 
            | 'Write to BQ' >> beam.io.Write(beam.io.BigQuerySink(
                    'prefab-envoy-220213:nhcdata.nhcevents', schema=schema,
                    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))
)

results = p.run()
results.wait_until_finish()

Если я удалю 

            | 'Write to BQ' >> beam.io.Write(beam.io.BigQuerySink(
                    'prefab-envoy-220213:nhcdata.nhcevents', schema=schema,
                    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))

тогда поток данных начинается нормально.

1 Ответ

0 голосов
/ 31 октября 2018

beam.io.Write(beam.io.BigQuerySink()) - это собственный приемник потока данных, который работает только для пакетных заданий. Для потоковых конвейеров вы должны использовать beam.io.WriteToBigQuery(). Использование: 

beam.io.WriteToBigQuery(table ='table_name',dataset='dataset',project='project_id')

Хороший пример можно найти в этом ответе .

...