Spark Dataset 列值更新：Java 实现与UDF应用指南

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.lit; // 导入lit函数

// 假设 yourdataset 已经是一个 Dataset<Row>
Dataset<Row> updatedDataset = yourdataset.withColumn("UPLOADED_ON_NEW", lit("Any-value"));

// 假设 originalDataset 是原始数据集
// 1. 创建一个新列，包含转换后的值（这里用 lit 举例，实际可替换为复杂逻辑）
Dataset<Row> tempDataset = originalDataset.withColumn("UPLOADED_ON_TEMP", lit("new_formatted_date"));

// 2. 移除原始的 UPLOADED_ON 列
Dataset<Row> withoutOldColumn = tempDataset.drop("UPLOADED_ON");

// 3. 将新列重命名为 UPLOADED_ON
Dataset<Row> finalDataset = withoutOldColumn.withColumnRenamed("UPLOADED_ON_TEMP", "UPLOADED_ON");

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

// 假设 sparkSession 已经初始化
SparkSession sparkSession = SparkSession.builder().appName("UDFExample").master("local[*]").getOrCreate();

// 注册日期格式转换UDF
sparkSession.udf().register(
    "formatDateYYYYMMDDtoDDMMYY", // UDF的名称
    (String dateIn) -> { // UDF的实现逻辑，这里使用Lambda表达式
        if (dateIn == null || dateIn.isEmpty()) {
            return null;
        }
        try {
            DateFormat inputFormatter = new SimpleDateFormat("yyyy-MM-dd");
            Date date = inputFormatter.parse(dateIn);
            DateFormat outputFormatter = new SimpleDateFormat("dd-MM-yy");
            return outputFormatter.format(date);
        } catch (ParseException e) {
            // 处理解析异常，例如返回原始值或null
            System.err.println("Error parsing date: " + dateIn + " - " + e.getMessage());
            return null; // 或者 dateIn;
        }
    },
    DataTypes.StringType // UDF的返回类型
);

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.callUDF;
import static org.apache.spark.sql.functions.col;

// 假设 yourdataset 已经加载
// Dataset<Row> yourdataset = sparkSession.read().format(...).load(...);

// 应用UDF创建新列
Dataset<Row> updatedDatasetWithUDF = yourdataset.withColumn(
    "UPLOADED_ON_NEW", // 新列的名称
    callUDF(
        "formatDateYYYYMMDDtoDDMMYY", // UDF的名称
        col("UPLOADED_ON") // 传入UDF的列
    )
);

// 如果需要替换原列，可以进一步操作：
Dataset<Row> finalDataset = updatedDatasetWithUDF
    .drop("UPLOADED_ON") // 移除旧列
    .withColumnRenamed("UPLOADED_ON_NEW", "UPLOADED_ON"); // 重命名新列

// 假设 yourdataset 已经加载
// Dataset<Row> yourdataset = sparkSession.read().format(...).load(...);

// 将Dataset注册为临时视图
yourdataset.createOrReplaceTempView("MY_DATASET");

// 在Spark SQL查询中使用UDF
Dataset<Row> resultDatasetFromSql = sparkSession.sql(
    "SELECT *, formatDateYYYYMMDDtoDDMMYY(UPLOADED_ON) AS UPLOADED_ON_NEW FROM MY_DATASET"
);

// 同样，如果需要替换原列，可以在SQL中直接完成：
// 例如：SELECT col1, col2, formatDateYYYYMMDDtoDDMMYY(UPLOADED_ON) AS UPLOADED_ON FROM MY_DATASET