如何用Python基于AVDL文件生成AVRO格式的合成测试数据？

步骤1：将AVDL转换为AVSC（Avro Schema JSON） #

Python的Avro库大多不直接支持AVDL（Avro IDL）解析，所以先把AVDL转成JSON格式的Schema（AVSC）。

方法一：用官方Avro工具（Java）

下载Avro Tools的jar包后，执行命令：

java -jar avro-tools-1.11.0.jar idl2schemata your_schema.avdl ./output_schemas/

这会在指定目录生成对应的.avsc文件。

方法二：用Python第三方库解析

安装avro-idl-parser：

pip install avro-idl-parser

然后用代码解析并保存为AVSC：

import json
from avro_idl_parser import parse_idl

with open("your_schema.avdl", "r") as f:
    idl_content = f.read()

# 解析AVDL得到Schema字典
schema_dict = parse_idl(idl_content)

# 保存为AVSC文件
with open("output_schema.avsc", "w") as f:
    json.dump(schema_dict, f, indent=2)

步骤2：生成符合Schema的合成数据 #

用faker生成贴近真实场景的假数据，结合Avro Schema的结构构造样本。先安装依赖：

pip install avro-python3 faker

编写生成数据的函数（可根据你的Schema字段自定义逻辑）：

import random
from faker import Faker
from avro.schema import parse
import json

# 加载AVSC Schema
with open("output_schema.avsc", "r") as f:
    schema = parse(json.dumps(json.load(f)))

fake = Faker()

def generate_fake_data(field_schema):
    """根据Avro字段Schema递归生成假数据"""
    schema_type = field_schema.type

    if schema_type == "record":
        record = {}
        for field in field_schema.fields:
            record[field.name] = generate_fake_data(field.type)
            # 针对特定字段做针对性生成，比如email字段
            if field.name.lower() == "email":
                record[field.name] = fake.email()
            elif field.name.lower() == "address":
                record[field.name] = fake.address().replace("\n", ", ")
        return record
    elif schema_type == "string":
        return fake.text(max_nb_chars=50)
    elif schema_type in ["int", "long"]:
        return random.randint(1, 100000)
    elif schema_type in ["float", "double"]:
        return round(random.uniform(0.0, 1000.0), 2)
    elif schema_type == "boolean":
        return random.choice([True, False])
    elif schema_type == "array":
        return [generate_fake_data(field_schema.items) for _ in range(random.randint(1, 5))]
    elif schema_type == "map":
        return {fake.word(): generate_fake_data(field_schema.values) for _ in range(random.randint(1, 3))}
    elif schema_type == "enum":
        return random.choice(field_schema.symbols)
    elif schema_type == "union":
        # 随机选择Union中的一种类型生成数据
        selected_schema = random.choice(field_schema.schemas)
        return generate_fake_data(selected_schema)
    elif schema_type == "null":
        return None
    else:
        return fake.text(max_nb_chars=20)

# 生成10条样本数据
sample_records = [generate_fake_data(schema) for _ in range(10)]

步骤3：写入AVRO文件 #

用avro-python3的DataFileWriter将数据写入文件：

from avro.datafile import DataFileWriter
from avro.io import DatumWriter

with open("sample_output.avro", "wb") as f:
    writer = DataFileWriter(f, DatumWriter(), schema)
    for record in sample_records:
        writer.append(record)
    writer.close()

替代方案：用fastavro提升速度

如果处理大量数据，推荐用fastavro（速度更快）：

pip install fastavro

写入代码更简洁：

import fastavro

with open("sample_output_fast.avro", "wb") as f:
    fastavro.writer(f, schema, sample_records)

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