如何将HDFS现有文本数据转Avro？及Avro Schema演进适配旧数据

我来帮你解决这两个HDFS与Avro结合的常见问题，都是实际工作中经常遇到的场景：

这里提供两种最常用的实现方式，你可以根据自己的技术栈选择：

方式1：用Spark快速转换（推荐） #

Spark的Avro集成非常友好，代码简洁易维护，适合大多数场景：

1. 确保你的Spark环境支持Avro：Spark 2.4及以上版本自带avro模块，运行时可以通过--packages org.apache.spark:spark-avro_2.12:3.5.0（根据你的Spark版本调整）引入依赖。
2. 读取HDFS上的文本文件，假设每行是分隔符分隔的记录（比如逗号、制表符）。
3. 定义Avro Schema（JSON格式），匹配文本数据的字段。
4. 将文本数据映射到Avro Schema，然后写入HDFS。

示例PySpark代码：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("TextToAvroConverter").getOrCreate()

# 读取HDFS文本文件（这里以逗号分隔为例，可根据实际调整分隔符）
text_data = spark.read.csv("hdfs://your/text/data/path", header=False, sep=",")

# 定义Avro Schema，替换成你实际的字段名和类型
avro_schema = """
{
  "type": "record",
  "name": "OriginalTextRecord",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "order_date", "type": "string"},
    {"name": "amount", "type": "string"}
  ]
}
"""

# 给文本数据的列命名，匹配Avro Schema的字段
named_text_df = text_data.toDF("user_id", "order_date", "amount")

# 转换并写入Avro格式到HDFS
named_text_df.write \
    .format("avro") \
    .option("avroSchema", avro_schema) \
    .save("hdfs://your/avro/output/path")

方式2：用MapReduce自定义转换 #

如果你的环境更偏向传统MapReduce，可以借助Avro的MapReduce API实现：

1. 编写Mapper类，读取每行文本，分割字段后创建Avro的GenericRecord实例。
2. 配置Job时指定输出格式为AvroOutputFormat，并传入Avro Schema。
3. 打包Jar包提交到Hadoop集群运行。

示例Java Mapper代码：

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class TextAvroMapper extends Mapper<Object, Text, NullWritable, GenericRecord> {
    private Schema targetSchema;

    @Override
    protected void setup(Context context) throws IOException {
        // 从Job配置中读取Avro Schema字符串
        String schemaStr = context.getConfiguration().get("avro.target.schema");
        targetSchema = new Schema.Parser().parse(schemaStr);
    }

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        // 分割文本行（这里用逗号，根据实际调整）
        String[] fields = value.toString().split(",");
        GenericRecord record = new GenericData.Record(targetSchema);
        // 给Avro记录赋值，顺序要和Schema一致
        record.put("user_id", fields[0]);
        record.put("order_date", fields[1]);
        record.put("amount", fields[2]);
        // 输出记录
        context.write(NullWritable.get(), record);
    }
}

Job配置的核心代码片段：

Job job = Job.getInstance(config, "TextToAvroJob");
job.setJarByClass(TextAvroMapper.class);
job.setMapperClass(TextAvroMapper.class);
// 输出键值类型：NullWritable + GenericRecord
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(GenericRecord.class);
// 设置Avro输出格式和Schema
job.setOutputFormatClass(AvroOutputFormat.class);
AvroOutputFormat.setSchema(job, targetSchema);
// 指定输入输出路径
FileInputFormat.addInputPath(job, new Path("hdfs://your/text/data/path"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://your/avro/output/path"));

这个场景的核心是利用Avro的Schema演进特性，同时兼容未迁移的Text格式旧数据，给你两种可行方案：

方案1：批量迁移旧数据到Avro（推荐长期方案） #

彻底统一数据格式，避免后续维护两种格式的麻烦：

1. 先定义包含新增列的Avro Schema，注意给新增列设置default属性（这是Schema演进的关键，确保旧数据转换时能自动填充默认值）。比如：

{
  "type": "record",
  "name": "UpdatedRecord",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "order_status", "type": "string", "default": "unprocessed"}, // 新增列，带默认值
    {"name": "order_date", "type": "string"},
    {"name": "amount", "type": "string"}
  ]
}

2. 按照问题1的方法，将原有Text数据转换为Avro格式，转换时新增列会自动用默认值填充。
3. 后续所有新数据都用这个最新的Avro Schema写入，读取时统一使用最新Schema，Avro会自动兼容新旧Avro数据（旧Avro数据的新增列会用默认值填充）。

方案2：不迁移旧数据，在读取层兼容两种格式 #

如果旧数据量极大，暂时无法批量迁移，可以在读取时合并两种格式的数据：

用Hive实现兼容 #

创建两个外部表分别指向Text旧数据和Avro新数据，再通过视图合并：

-- 指向原有Text数据的外部表
CREATE EXTERNAL TABLE old_text_orders (
    user_id string,
    order_date string,
    amount string
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 'hdfs://your/text/data/path';

-- 指向Avro新数据的外部表（包含新增列）
CREATE EXTERNAL TABLE new_avro_orders (
    user_id string,
    order_status string,
    order_date string,
    amount string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
TBLPROPERTIES ('avro.schema.url'='hdfs://your/updated/schema.avsc');

-- 创建合并视图，给Text表的新增列填充默认值
CREATE VIEW merged_orders AS
SELECT user_id, 'unprocessed' as order_status, order_date, amount FROM old_text_orders
UNION ALL
SELECT user_id, order_status, order_date, amount FROM new_avro_orders;

用Spark实现兼容 #

分别读取Text和Avro数据，给Text数据添加新增列的默认值后合并：

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

spark = SparkSession.builder.appName("MergeTextAvro").getOrCreate()

# 读取旧Text数据，添加新增列并填充默认值
text_df = spark.read.csv("hdfs://your/text/data/path", header=False, sep=",").toDF("user_id", "order_date", "amount")
text_with_new_col = text_df.withColumn("order_status", lit("unprocessed"))

# 读取新Avro数据
avro_df = spark.read.format("avro").load("hdfs://your/avro/data/path")

# 合并两个DataFrame（确保列名顺序一致）
merged_df = text_with_new_col.select("user_id", "order_status", "order_date", "amount") \
                             .unionByName(avro_df)

# 后续可以将merged_df用于分析或写入统一格式

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