如何将Avro Schema转换为PyArrow Schema并用于Beam写Parquet？

步骤1：读取并解析Avro Schema文件 #

先读取.avsc文件内容，将其解析为标准Avro Schema对象，推荐用fastavro（比官方avro库效率更高）：

import json
from fastavro import parse_schema

# 读取avsc文件内容
with open("your_schema.avsc", "r") as f:
    avro_schema_dict = json.load(f)

# 解析为Avro Schema对象
avro_schema = parse_schema(avro_schema_dict)

步骤2：转换为PyArrow Schema #

借助pyarrow.avro模块的工具方法，直接完成Avro到PyArrow Schema的转换：

import pyarrow.avro as pa_avro

# 转换得到PyArrow Schema
pyarrow_schema = pa_avro.from_avro_schema(avro_schema)

步骤3：在Beam写入Parquet时使用该Schema #

将转换后的PyArrow Schema直接传入WriteToParquet的schema参数即可：

import apache_beam as beam
from apache_beam.io.parquetio import WriteToParquet

with beam.Pipeline() as p:
    (p
     | "加载数据" >> beam.Create(your_dataset)  # 替换为实际的数据加载逻辑
     | "写入Parquet" >> WriteToParquet(
         filename="output.parquet",
         schema=pyarrow_schema
     ))

额外说明 #

• 提前安装依赖：pip install pyarrow fastavro apache-beam
• 若使用官方avro库解析Schema，只需把fastavro.parse_schema替换为avro.schema.parse(json.dumps(avro_schema_dict))
• 复杂嵌套类型、联合类型的转换，pyarrow.avro能覆盖绝大多数标准场景，特殊自定义类型需单独处理

内容的提问来源于stack exchange，提问作者Cosmin Chauciuc