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> yourdataset = sparkSession.read()....;

// 1. 创建一个名为 "UPLOADED_ON_NEW" 的新列，其值为 "Any-value"
//    如果新列名与旧列名相同，则会直接替换
Dataset<Row> updatedDataset = yourdataset.withColumn("UPLOADED_ON_NEW", lit("Any-value"));

// 2. 如果需要，删除原始的 "UPLOADED_ON" 列
updatedDataset = updatedDataset.drop("UPLOADED_ON");

// 现在 updatedDataset 包含了名为 "UPLOADED_ON_NEW" 的新列，而没有原始的 "UPLOADED_ON" 列
updatedDataset.show();

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.api.java.UDF1; // 导入UDF1接口
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.text.ParseException; // 导入ParseException

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

// 注册一个UDF，用于将日期字符串从 "yyyy-MM-dd" 格式转换为 "dd-MM-yy" 格式
sparkSession.udf().register(
    "formatDateYYYYMMDDtoDDMMYY", // UDF的名称
    (UDF1<String, 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的返回类型
);

System.out.println("UDF 'formatDateYYYYMMDDtoDDMMYY' registered successfully.");

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

// 假设 yourdataset 已经加载，并且 UDF 已经注册
// Dataset<Row> yourdataset = sparkSession.read()....;

// 使用注册的UDF来转换 "UPLOADED_ON" 列，并将结果存入 "UPLOADED_ON_NEW" 列
Dataset<Row> transformedDataset = yourdataset.withColumn(
    "UPLOADED_ON_NEW",
    callUDF(
        "formatDateYYYYMMDDtoDDMMYY", // UDF的名称
        col("UPLOADED_ON") // 传入UDF的列
    )
);

// 如果需要替换原始列，可以删除旧列并重命名新列
transformedDataset = transformedDataset.drop("UPLOADED_ON")
                                       .withColumnRenamed("UPLOADED_ON_NEW", "UPLOADED_ON");

transformedDataset.show();

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// 假设 sparkSession 已经初始化， yourdataset 已经加载，并且 UDF 已经注册

// 1. 将 Dataset 注册为一个临时视图，以便在SQL查询中使用
yourdataset.createOrReplaceTempView("MY_DATASET");

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

// 如果需要，可以进一步处理，例如删除旧列并重命名新列
sqlTransformedDataset = sqlTransformedDataset.drop("UPLOADED_ON")
                                             .withColumnRenamed("UPLOADED_ON_NEW", "UPLOADED_ON");

sqlTransformedDataset.show();