如何使用Java更新Spark Dataset中列的数值？

问题原因 #

你用foreach修改列值无效的核心原因：

1. Spark的Dataset是不可变的，任何修改都需要生成新的Dataset，而非直接修改原数据
2. foreach是行动算子，它的逻辑仅在执行器端的局部变量上操作，你修改的date只是循环内的临时变量，根本不会写入到数据集中

解决方案 #

方案1：使用Spark内置日期函数（推荐） #

利用Spark SQL提供的to_date和date_format函数，无需手动解析日期，性能更优：

import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;

// 将字符串类型的UPLOADED_ON转换为日期类型，指定输入格式为yyyy-MM-dd
Column dateColumn = functions.to_date(rows.col("UPLOADED_ON"), "yyyy-MM-dd");
// 将日期格式化为dd-MM-yy的字符串格式
Column formattedDate = functions.date_format(dateColumn, "dd-MM-yy");
// 替换原UPLOADED_ON列，生成新的Dataset
Dataset<Row> updatedRows = rows.withColumn("UPLOADED_ON", formattedDate);

方案2：自定义UDF（处理复杂逻辑时使用） #

如果内置函数满足不了需求，可以自定义UDF来处理日期格式化：

注意：推荐用Java 8+的DateTimeFormatter（线程安全）替代SimpleDateFormat

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.types.DataTypes;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

// 定义自定义UDF：输入字符串日期，返回格式化后的字符串
UDF1<String, String> dateFormatUdf = (inputDate) -> {
    if (inputDate == null) {
        return null;
    }
    DateTimeFormatter inputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
    DateTimeFormatter outputFormatter = DateTimeFormatter.ofPattern("dd-MM-yy");
    try {
        LocalDate date = LocalDate.parse(inputDate, inputFormatter);
        return date.format(outputFormatter);
    } catch (Exception e) {
        // 解析失败时返回原字符串，也可以返回null或自定义值
        return inputDate;
    }
};

// 注册UDF到SparkSession
sparkSession.udf().register("customDateFormat", dateFormatUdf, DataTypes.StringType);

// 使用UDF替换原列
Dataset<Row> updatedRows = rows.withColumn(
    "UPLOADED_ON",
    functions.callUDF("customDateFormat", rows.col("UPLOADED_ON"))
);

关键注意点 #

• Spark的Dataset/DataFrame是不可变结构，所有修改操作都会生成新的实例，务必接收返回的新Dataset
• 优先使用Spark内置函数，它们经过分布式优化，性能远高于自定义UDF
• 如果必须用SimpleDateFormat，要注意它不是线程安全的，在UDF中需要用ThreadLocal包装，避免多线程环境下的异常

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