17370845950

本文介绍在 java 版 apache spark 中，如何在 `dataset.select()` 阶段直接添加带固定值的新列（常量列），避免后期手动修改 csv 文件，确保导出数据结构完整、高效且符合生产规范。

在 Spark SQL 的 Java API 中，向 Dataset 添加常量列最简洁、标准的方式是使用 org.apache.spark.sql.functions.lit() 函数。该函数返回一个 Column 对象，表示一个字面量（literal）值，可与其他列一同参与 select() 操作，完全融入 Spark 的 Catalyst 优化器流程，性能无损且语义清晰。

以下是在您原有代码中插入常量列的正确写法：

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

private static void buildReport(FileSystem fileSystem, Dataset joinedDs, String reportName) throws IOException {
    Path report = new Path(reportName);
    joinedDs.filter(aFilter)
            .select(
                    joinedDs.col("AGREEMENT_ID"),
                    lit("REPORT_V1").as("REPORT_VERSION"),  // ✅ 常量列：值为字符串 "REPORT_V1"，别名为 "REPORT_VERSION"
                    joinedDs.col("ERROR_MESSAGE")
            )
            .write()
            .format("csv")
            .option("header", "true")   // 注意：布尔值应传字符串 "true"/"false"
            .option("sep", ",")
            .csv(reportName);

    fileSystem.copyToLocalFile(report, new Path(reportName + ".csv"));
}

? 关键说明与注意事项：

• lit() 支持任意 JVM 基本类型（String, Integer, Boolean, Double 等）及 null（自动推断为 NullType）；
• 必须配合 .as("columnName") 显式指定列名，否则生成的列名为默认字面量表达式（如 REPORT_V1），但无显式别名可能导致 CSV 头混乱或下游解析异常；
• lit() 是惰性求值的——它不触发任何计算，仅构建逻辑计划，因此零额外开销；
• 若需动态常量（如当前时间戳），请改用 current_timestamp().as("ETL_TIME") 等内置函数，而非 lit(new Timestamp(...))（后者会固化为作业提交时刻的静态值）；
• 所有 lit() 列均与原始数据行一一对应，行为等同于 SQL 中的 SELECT ..., 'abc' AS col_name, ...。

通过此方式，您可在数据写出前统一增强 Schema，既保持 ETL 流程的声明式风格，又确保 CSV 输出开箱即用，无需额外文本处理步骤。