在 Dataflow Python 中从 PubSub 读取 AVRO 消息

Question

我需要从另一个 GCP 项目的 PubSub 主题中读取 AVRO 消息。我之前实现了 Python 数据流管道，它从 PubSub 读取 JSON 消息并写入 BigQuery。但我是处理 AVRO 消息的新手。我试图查找 AVRO 的 Python 文档，它指向此链接 https://avro.apache.org/docs/current/gettingstartedpython.html

在这个链接中有一些从文件读取和写入文件的例子，但我认为这些函数对于从 PubSub 读取没有用。我正在使用以下转换从输出为字节串的 PubSub 中读取。

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

我需要一种读取这些字节的方法（AVRO 格式）

Answer 1

这是一个您可以使用的示例代码

1. 从 Pub/Sub 读取消息

from fastavro import parse_schema, schemaless_reader

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

1. 使用 Fastavro 包通过类定义定义架构和读取器

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

1. 现在映射字节元素并指定架构

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

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

这些行应该有PCollection，格式为Avro。