Java Spark中展开嵌套数组为新列且保留原表其余字段的方法

原始数据Schema #

|-- ID: integer (nullable = true)
 |-- Name: integer (nullable = true)
 |-- response: struct (nullable = true)
 |    |-- status: string (nullable = true)
 |    |-- indicator: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- _VALUE: string (nullable = true)
 |    |    |    |-- _number: long (nullable = true)
 |    |-- result: string (nullable = true)

问题描述 #

此前的解决方案仅保留了ID和展开后的indicator数组，丢失了Status、Name、Result等字段。由于实际场景中存在大量字段及response列内的嵌套元素，无法手动逐个选择字段，需要在保留原表所有内容的前提下，将数组展开为新列。

尝试过的方法包括：将仅含ID和数组的表存为临时表后关联原表（已删除嵌套数组）；选择response列中除indicator外的元素重构结构后关联，但均未成功。测试代码片段如下：

Dataset<Row> result = File.select("response.*").columns().filter(row -> row.equals(col("indicator"))).map(s -> col(s));

Dataset<Row> result = File.select("response.*").filter(c -> !c.equals(col("indicator")));

File.drop(col("response")).withColumn("response", struct(result));

File.drop("response").join(result, File.col(ID).equalTo(result.col(ID)), "outer");

期望最终Schema #

|-- ID: integer (nullable = true)
|-- Name: integer (nullable = true)
|-- response: struct (nullable = true)
|    |-- status: string (nullable = true)
|    |-- result: string (nullable = true)
|-- response_indicator_number: integer (nullable = true) <-- 对应response.indicator._number元素

解决方案 #

核心思路 #

1. 用explode展开嵌套数组，同时保留所有原始字段
2. 从展开的数组元素中提取目标字段作为新列
3. 动态重构response结构（移除indicator字段），避免手动枚举字段
4. 清理临时生成的中间列

完整Java代码 #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import static org.apache.spark.sql.functions.*;
import java.util.Arrays;

public class ExpandNestedArray {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("ExpandNestedArray")
                .master("local[*]")
                .getOrCreate();

        // 替换为你的数据读取逻辑，比如从文件/数据库读取
        Dataset<Row> originalDF = spark.read().format("json").load("path/to/your/data");

        // 1. 展开indicator数组，保留所有原字段
        Dataset<Row> explodedDF = originalDF
                .withColumn("exploded_indicator", explode(col("response.indicator")));

        // 2. 提取_number作为新列
        Dataset<Row> withNewColDF = explodedDF
                .withColumn("response_indicator_number", col("exploded_indicator._number"));

        // 3. 动态获取response下除indicator外的所有字段，避免手动枚举
        StructType responseStruct = (StructType) originalDF.schema().apply("response").dataType();
        String[] responseFieldNames = Arrays.stream(responseStruct.fields())
                .map(StructField::name)
                .filter(name -> !name.equals("indicator"))
                .toArray(String[]::new);

        // 构建重构response所需的列
        org.apache.spark.sql.Column[] responseCols = Arrays.stream(responseFieldNames)
                .map(field -> col("response." + field))
                .toArray(org.apache.spark.sql.Column[]::new);

        // 4. 重构response列并清理临时列
        Dataset<Row> finalDF = withNewColDF
                .withColumn("response", struct(responseCols))
                .drop("exploded_indicator");

        // 打印最终Schema验证
        finalDF.printSchema();
        // 查看数据示例
        finalDF.show();
    }
}

代码说明 #

• explode(col("response.indicator"))：将数组展开为多行，同时保留所有原始字段，不会丢失任何数据
• 动态获取response字段：通过Schema解析自动获取除indicator外的所有字段，适配字段较多的场景
• struct(responseCols)：重构response结构，只保留需要的字段
• drop("exploded_indicator")：清理临时生成的展开列，得到干净的最终表结构

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