Pubsubio to Bigquery using Dataflow

Question

I am getting the error belwo while inserting a message from pubsubio to BigQuery.

How can I insert a record from pubsub to BQ. Can we convert pcollection into a list, or is there another alternate?

AttributeError: 'PCollection' object has no attribute 'split'

Here is my code:

def create_record(columns):
    #import re
    col_value=record_ids.split('|')
    col_name=columns.split(",")
    for i in range(length(col_name)):
        schmea_dict[col_name[i]]=col_value[i]
    return schmea_dict

schema = 'tungsten_opcode:STRING,tungsten_seqno:INTEGER
columns="tungsten_opcode,tungsten_seqno"
lines = p | 'Read PubSub' >> beam.io.ReadStringsFromPubSub(INPUT_TOPIC) | 
    beam.WindowInto(window.FixedWindows(15))
record_ids = lines | 'Split' >> 
    (beam.FlatMap(split_fn).with_output_types(unicode))
records = record_ids | 'CreateRecords' >> beam.Map(create_record(columns))
records | 'BqInsert' >> beam.io.WriteToBigQuery(
    OUTPUT,
    schema=schema,
    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)

def create_record(columns): #import re col_value=record_ids.split('|') col_name=columns.split(",") for i in range(length(col_name)): schmea_dict[col_name[i]]=col_value[i] return schmea_dict — Surya
– Surya, Commented Jun 22, 2018 at 14:58
Is there any alternative to load pubsub message to BQ using dataflow — Surya
– Surya, Commented Jun 22, 2018 at 15:00

Ryan M · Accepted Answer · 2018-06-26 00:24:08Z

2

Needs to be done as a transform, you cant directly access data in a pcollection.

Write a DoFn class to perform the split transform on the record with schema as a side input, and create the dict with column/records eg.

class CreateRecord(beam.DoFn):
  def process(self, element, schema):
    cols = element.split(',')
    header = map(lambda x: x.split(':')[0], schema.split(','))
    return [dict(zip(header, cols))]

Apply transform like:

schema = 'tungsten_opcode:STRING,tungsten_seqno:INTEGER'
records = record_ids | 'CreateRecords' >> beam.ParDo(CreateRecord(), SCHEMA)

edited Jun 26, 2018 at 0:24

answered Jun 25, 2018 at 4:02

Ryan M

3421 silver badge7 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Pubsubio to Bigquery using Dataflow

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related