Read AVRO messages from PubSub in Dataflow Python

Question

I have a requirement to read AVRO messages from another GCP project's PubSub topic. I have earlier implemented Python Dataflow pipelines which read JSON messages from PubSub and write to BigQuery. But I am new to handling AVRO messages. I tried to lookup Python documentation for AVRO and it points me to this link https://avro.apache.org/docs/current/gettingstartedpython.html

In this link there are examples that read from files and write to files, but I don't think these functions will be useful to read from PubSub. I am using the below transform to read from PubSub where the output is a bytestring.

"Read from PubSub" >> beam.io.ReadFromPubSub(topic=TOPIC).with_output_types(bytes)

I need a way to read these bytes(AVRO format)

Jayadeep Jayaraman Jayadeep Jayaraman · Accepted Answer · 2020-04-15T04:33:59

Here is a sample code that you can use

Read Messages from Pub/Sub

from fastavro import parse_schema, schemaless_reader

messages = (p
            | beam.io.ReadFromPubSub(
                subscription=known_args.input_subscription)
            .with_output_types(bytes))

Use Fastavro package to define the schema and a reader via a Class definition

class AvroReader:
    def __init__(self, schema):
        self.schema = schema

    def deserialize(self, record):
        bytes_reader = io.BytesIO(record)
        dict_record = schemaless_reader(bytes_reader, self.schema)
        return dict_record

Now map over the byte elements and specify the schema

schema = avro.schema.parse(open("avro.avsc", "rb").read())
avroReader = AvroReader(schema)

lines = messages | "decode" >> beam.Map(lambda input: avroReader.deserialize(input))

The lines should have PCollection in the form of Avro.

Read AVRO messages from PubSub in Dataflow Python

1 Answers