如何基于现有Avro Schema生成样本数据？（Kafka性能测试场景）

以下是几个实用且易维护的解决方案，覆盖不同技术栈和复杂度需求：

1. 基于Avro GenericRecord + 随机数据生成库（自定义程度高，通用） #

利用Avro的Generic API结合成熟的随机数据工具，针对Schema字段类型生成对应样本，适合需要定制部分字段规则的场景。

Java示例代码： #

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.lang3.RandomStringUtils;
import java.io.File;
import java.util.concurrent.ThreadLocalRandom;

public class AvroDataGenerator {
    public static void main(String[] args) throws Exception {
        // 加载Avro Schema文件
        Schema schema = new Schema.Parser().parse(new File("your-large-schema.avsc"));
        // 生成单条样本数据
        GenericRecord testRecord = generateRandomRecord(schema);
        // 后续可序列化后发送到Kafka
    }

    private static Object generateRandomValue(Schema.Field field) {
        Schema.Type type = resolveActualType(field.schema());
        switch (type) {
            case STRING:
                return RandomStringUtils.randomAlphanumeric(8, 20);
            case INT:
                return ThreadLocalRandom.current().nextInt(0, 10000);
            case LONG:
                return ThreadLocalRandom.current().nextLong(0, 1000000);
            case BOOLEAN:
                return ThreadLocalRandom.current().nextBoolean();
            case RECORD:
                // 递归生成嵌套Record
                GenericRecord nestedRecord = new GenericData.Record(field.schema());
                for (Schema.Field nestedField : field.schema().getFields()) {
                    nestedRecord.put(nestedField.name(), generateRandomValue(nestedField));
                }
                return nestedRecord;
            case ARRAY:
                // 生成随机长度的数组（1-5个元素）
                int arraySize = ThreadLocalRandom.current().nextInt(1, 6);
                GenericData.Array<Object> array = new GenericData.Array<>(arraySize, field.schema());
                Schema elementSchema = field.schema().getElementType();
                for (int i = 0; i < arraySize; i++) {
                    array.add(generateRandomValue(new Schema.Field("temp", elementSchema)));
                }
                return array;
            // 根据Schema需求扩展其他类型（如MAP、ENUM等）
            default:
                return null;
        }
    }

    // 处理联合类型（如["null", "string"]），优先返回非null类型
    private static Schema.Type resolveActualType(Schema schema) {
        if (schema.getType() == Schema.Type.UNION) {
            for (Schema s : schema.getTypes()) {
                if (s.getType() != Schema.Type.NULL) {
                    return s.getType();
                }
            }
            return Schema.Type.NULL;
        }
        return schema.getType();
    }

    private static GenericRecord generateRandomRecord(Schema schema) {
        GenericRecord record = new GenericData.Record(schema);
        for (Schema.Field field : schema.getFields()) {
            record.put(field.name(), generateRandomValue(field));
        }
        return record;
    }
}

Python示例（用fastavro + Faker）： #

适合快速编写脚本生成数据，无需编译：

import fastavro
import json
from faker import Faker
import random

fake = Faker()

def generate_random_value(field_schema):
    # 处理联合类型
    if isinstance(field_schema, list):
        field_schema = random.choice([s for s in field_schema if s != 'null'])
    
    if isinstance(field_schema, str):
        if field_schema == 'string':
            return fake.name()
        elif field_schema == 'int':
            return random.randint(0, 10000)
        elif field_schema == 'boolean':
            return random.choice([True, False])
    elif isinstance(field_schema, dict):
        if field_schema['type'] == 'record':
            return generate_record(field_schema)
        elif field_schema['type'] == 'array':
            return [generate_random_value(field_schema['items']) for _ in range(random.randint(1,5))]
    return None

def generate_record(schema):
    record = {}
    for field in schema['fields']:
        record[field['name']] = generate_random_value(field['type'])
    return record

# 加载Schema
with open('your-schema.avsc', 'r') as f:
    schema = fastavro.parse_schema(json.load(f))

# 生成100条样本并打印（可序列化后发送到Kafka）
for _ in range(100):
    print(generate_record(schema))

2. 使用开源Avro数据生成库（零自定义开发） #

直接借助成熟的开源库自动生成数据，无需手动处理字段类型映射，适合快速生成通用样本：

Java推荐：avro-data-generator #

引入Maven依赖后，几行代码即可生成数据：

<dependency>
    <groupId>com.github.vincentrussell</groupId>
    <artifactId>avro-data-generator</artifactId>
    <version>1.0.23</version>
</dependency>

import com.github.vincentrussell.avro.data.generator.AvroDataGenerator;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import java.io.File;

public class AvroLibGenerator {
    public static void main(String[] args) throws Exception {
        Schema schema = new Schema.Parser().parse(new File("your-schema.avsc"));
        AvroDataGenerator generator = new AvroDataGenerator();
        // 生成单条随机数据
        GenericRecord record = generator.generate(schema);
        // 支持自定义规则（如指定字段的取值范围），可查看库文档配置
    }
}

3. 批量生成Avro文件后导入Kafka（适合大规模测试） #

如果需要海量测试数据，可以先批量生成Avro文件，再用Kafka工具批量发送：

1. 用上述方法生成一批Avro记录，写入本地Avro文件
2. 使用Kafka官方工具发送：

kafka-console-producer --bootstrap-server your-kafka-broker:9092 --topic test-topic \
  --property value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer \
  --property schema.registry.url=http://your-schema-registry:8081 \
  --property parse.key=false < your-avro-file.avro

优化建议 #

• 缓存Schema解析结果：大型Schema解析开销大，解析一次后复用
• 定制字段规则：对需要特定格式的字段（如邮箱、手机号），用Faker等工具生成符合要求的数据
• 多线程生成：生成海量数据时，用多线程并行生成提升效率
• 复用生成逻辑：将数据生成封装为通用工具类，适配不同Schema只需传入文件路径

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