Apache Spark 2.3.0 如何使用Java实现两个array<string>的合并

Spark 2.3.x版本未内置concat_array/flatten高阶函数，无法直接使用2.4+版本的内置方法实现数组合并，可通过以下两种方式实现需求：

方案1：自定义UDF实现（推荐，适配DataFrame/SQL API） #

该方案无需转换数据结构，性能损耗更低，适配Spark SQL的常规使用场景：

• 第一步：编写自定义UDF逻辑

import org.apache.spark.sql.api.java.UDF1;
import java.util.List;
import java.util.ArrayList;

// 入参为嵌套字符串数组，返回扁平后的字符串数组
UDF1 flattenArrayUDF = new UDF1<List<List<String>>, List<String>>() {
    @Override
    public List<String> call(List<List<String>> nestedArray) throws Exception {
        List<String> result = new ArrayList<>();
        if (nestedArray == null) {
            return result;
        }
        for (List<String> subArray : nestedArray) {
            if (subArray != null) {
                result.addAll(subArray);
            }
        }
        return result;
    }
};

• 第二步：注册UDF到SparkSession

import org.apache.spark.sql.types.DataTypes;

// 指定UDF名称、实现类、返回值类型
spark.udf().register("flatten_array", flattenArrayUDF, DataTypes.createArrayType(DataTypes.StringType));

• 第三步：调用UDF转换数据
  假设存储嵌套数组的列名为nested_arr，直接调用即可得到合并后的数组列：

import static org.apache.spark.sql.functions.expr;

Dataset<Row> resultDF = df.withColumn("merged_arr", expr("flatten_array(nested_arr)"));

方案2：RDD转换实现 #

如果不想使用UDF，可以转换为RDD处理后再转回DataFrame：

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.util.ArrayList;
import java.util.List;

// 处理RDD数据，拍平嵌套数组
JavaRDD<Row> processedRDD = df.javaRDD().map(row -> {
    // 获取嵌套数组列
    List<List<String>> nestedArr = row.getList(row.fieldIndex("nested_arr"));
    List<String> mergedArr = new ArrayList<>();
    for (List<String> subArr : nestedArr) {
        if (subArr != null) mergedArr.addAll(subArr);
    }
    // 按需保留原有列，最后追加合并后的数组列
    return RowFactory.create(row.getAs("other_column1"), row.getAs("other_column2"), mergedArr);
});

// 构造新DataFrame的Schema
List<StructField> fields = new ArrayList<>();
fields.add(DataTypes.createStructField("other_column1", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("other_column2", DataTypes.IntegerType, true));
fields.add(DataTypes.createStructField("merged_arr", DataTypes.createArrayType(DataTypes.StringType), true));
StructType newSchema = DataTypes.createStructType(fields);

// 转换回DataFrame
Dataset<Row> resultDF = spark.createDataFrame(processedRDD, newSchema);

注意事项 #

两种实现都做了空值兼容，嵌套数组、子数组为null时不会抛出空指针异常，可直接用于生产环境。

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