如何在Spark中对嵌套JSON反规范化并保存为Avro

作为有Java背景刚接触Spark的开发者，你需要把嵌套JSON转成点分隔键的扁平结构再存成Avro，下面我给你一步步拆解实现思路和代码，帮你快速适应Spark的函数式风格。

1. 先搞定依赖准备 #

首先要确保你的项目引入Spark SQL和Avro的依赖，以Maven为例：

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.3.0</version> <!-- 替换成你实际使用的Spark版本 -->
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-avro_2.12</artifactId>
        <version>3.3.0</version>
    </dependency>
</dependencies>

2. 完整代码实现 #

下面是可直接运行的Java代码，包含读取JSON、递归扁平化、保存Avro的全流程：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import java.util.ArrayList;
import java.util.List;

public class JsonFlattenToAvro {
    public static void main(String[] args) {
        // 初始化SparkSession - Spark SQL的核心入口
        SparkSession spark = SparkSession.builder()
                .appName("JsonFlattenAvro")
                .master("local[*]") // 本地调试用，生产环境请移除该行
                .getOrCreate();

        // 1. 读取嵌套JSON数据（这里用示例字符串模拟，实际可替换为文件路径）
        String jsonSample = "{ \"player\": { \"username\": \"user1\", \"characteristics\": { \"race\": \"Human\", \"class\": \"Warlock\", \"subclass\": \"Dawnblade\", \"power\": 300, \"playercountry\": \"USA\" }, \"arsenal\": { \"kinetic\": { \"name\": \"Sweet Business\", \"type\": \"Auto Rifle\", \"power\": 300, \"element\": \"Kinetic\" }, \"energy\": { \"name\": \"MIDA Mini-Tool\", \"type\": \"Submachine Gun\", \"power\": 300, \"element\": \"Solar\" }, \"power\": { \"name\": \"Play of the Game\", \"type\": \"Grenade Launcher\", \"power\": 300, \"element\": \"Arc\" } }, \"armor\": { \"head\": \"Eye of Another World\", \"arms\": \"Philomath Gloves\", \"chest\": \"Philomath Robes\", \"leg\": \"Philomath Boots\", \"classitem\": \"Philomath Bond\" }, \"location\": { \"map\": \"Titan\", \"waypoint\": \"The Rig\" } } }";
        Dataset<Row> nestedDF = spark.read().json(spark.createDataset(List.of(jsonSample), spark.sqlContext().conf().getDefaultStringType()));

        // 2. 调用自定义方法扁平化DataFrame
        Dataset<Row> flattenedDF = flattenDataFrame(nestedDF);

        // 3. 查看结果（调试用，生产环境可注释）
        flattenedDF.show(false);
        flattenedDF.printSchema();

        // 4. 保存为Avro文件
        flattenedDF.write()
                .format("avro")
                .mode("overwrite") // 可选模式：append/ignore/errorIfExists
                .save("./output/flattened_player.avro");

        spark.stop();
    }

    // 核心：递归扁平化任意层级的嵌套DataFrame
    private static Dataset<Row> flattenDataFrame(Dataset<Row> df) {
        List<String> flatColumns = new ArrayList<>();
        List<String> nestedColumns = new ArrayList<>();

        // 遍历所有列，区分普通列和嵌套结构列（StructType）
        df.schema().fields().forEach(field -> {
            if (field.dataType() instanceof StructType) {
                nestedColumns.add(field.name());
            } else {
                flatColumns.add(field.name());
            }
        });

        // 生成扁平化列表达式：用点连接嵌套层级，比如 player.username
        List<String> expandedColumns = new ArrayList<>(flatColumns);
        for (String nestedCol : nestedColumns) {
            StructType nestedSchema = (StructType) df.schema().apply(nestedCol).dataType();
            for (String field : nestedSchema.fieldNames()) {
                String flatColName = nestedCol + "." + field;
                expandedColumns.add(nestedCol + "." + field + " AS `" + flatColName + "`");
            }
        }

        // 选择所有扁平化后的列
        Dataset<Row> tempFlattenedDF = df.selectExpr(expandedColumns.toArray(new String[0]));

        // 递归处理剩余的嵌套列（如果还有更深层级）
        boolean hasNested = tempFlattenedDF.schema().fields().stream()
                .anyMatch(field -> field.dataType() instanceof StructType);

        return hasNested ? flattenDataFrame(tempFlattenedDF) : tempFlattenedDF;
    }
}

3. 关键逻辑解释 #

• SparkSession初始化：这是Spark SQL的入口，本地调试用master("local[*]")，生产环境交给集群管理即可。
• 递归扁平化方法：
  • 先区分当前DataFrame的普通列和嵌套列（StructType类型）。
  • 对嵌套列生成点分隔的列名，并用AS重命名，确保列名符合你的需求。
  • 递归调用直到没有嵌套列，完美适配任意层级的JSON结构。
• Avro保存：Spark原生支持Avro读写，只需指定format("avro")，注意保存模式的选择（比如overwrite会覆盖已有文件）。

4. 适应Spark函数式风格的小技巧 #

作为Java开发者，刚接触Spark可能会有点不适应，给你几个小建议：

• 多用Java 8+的Lambda表达式（比如forEach、stream().filter()），和Spark的API风格更契合。
• 理解Spark的惰性求值：所有转换操作（比如selectExpr）不会立即执行，只有遇到行动操作（比如show()、write()）才会触发计算。
• 重点熟悉Dataset、Row、StructType这些核心类的方法，它们是Spark SQL操作的基础。

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