如何用Java在Apache Spark中关联DataFrame填充空列

当然可以用Java实现这个需求，同时也有不少更简便的替代方案，下面我结合你提供的示例数据详细说明：

一、Java实现方案 #

Java里处理DataFrame场景，最常用的是Apache Spark（适合大数据量）或者结合Apache Commons CSV做集合操作（小数据量）。这里以Spark为例，因为它天生擅长处理DataFrame的关联和更新：

假设你的两个DataFrame共同匹配列是BILL_ID和BILL_NBR_TYPE_CD，需要填充的目标列是BILL_NBR（对应示例中的字段），DataFrame2中该列存在空值。

步骤&代码示例： #

1. 先加载两个CSV文件为Spark DataFrame：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

// 初始化Spark会话
SparkSession spark = SparkSession.builder()
        .appName("FillNullFromAnotherDF")
        .master("local[*]") // 本地模式，生产环境可去掉
        .getOrCreate();

// 加载源数据DataFrame1（你的File1.csv）
Dataset<Row> df1 = spark.read()
        .option("header", true)
        .csv("File1.csv");

// 加载需要更新的目标DataFrame2
Dataset<Row> df2 = spark.read()
        .option("header", true)
        .csv("File2.csv");

2. 关联两个DataFrame，完成空值填充：

// 左关联保留DataFrame2的所有行，匹配共同列后填充空值
Dataset<Row> updatedDF = df2.join(df1, 
        df2.col("BILL_ID").equalTo(df1.col("BILL_ID"))
        .and(df2.col("BILL_NBR_TYPE_CD").equalTo(df1.col("BILL_NBR_TYPE_CD"))),
        "left")
        .select(
                // 保留DataFrame2的原有字段，仅填充目标列的空值
                df2.col("BILL_ID"),
                df2.col("BILL_NBR_TYPE_CD"),
                when(df2.col("BILL_NBR").isNull(), df1.col("BILL_NBR"))
                        .otherwise(df2.col("BILL_NBR")).alias("BILL_NBR"),
                df2.col("VERSION"),
                df2.col("PRIM_SW")
        );

3. 输出更新后的结果：

updatedDF.write()
        .option("header", true)
        .mode("overwrite")
        .csv("Updated_File2.csv");

如果数据量很小，也可以不用Spark：用Apache Commons CSV读取两个文件到List<Map<String, String>>，遍历DataFrame2的每条数据，根据BILL_ID和BILL_NBR_TYPE_CD去DataFrame1中查找匹配项，然后替换空值即可。

二、更简便的替代方案：Python Pandas #

如果不限制编程语言，Python的Pandas库处理这类数据填充会简洁很多，几行代码就能搞定：

代码示例： #

import pandas as pd

# 加载两个CSV文件
df1 = pd.read_csv("File1.csv")
df2 = pd.read_csv("File2.csv")

# 基于共同列合并，自动用df1的非空值填充df2的空值
df_updated = df2.set_index(["BILL_ID", "BILL_NBR_TYPE_CD"]) \
                .combine_first(df1.set_index(["BILL_ID", "BILL_NBR_TYPE_CD"])) \
                .reset_index()

# 保存更新后的文件
df_updated.to_csv("Updated_File2.csv", index=False)

combine_first()方法会自动匹配相同索引（这里是两个共同列）的行，用df1的非空值覆盖df2的空值，完全符合你的需求。

三、关键注意事项 #

• 确保两个DataFrame的共同列数据类型完全一致，比如BILL_ID如果在df1是字符串，df2也必须是字符串，否则会出现匹配失败的情况
• 如果DataFrame1中存在同一共同列组合的多条数据，建议先去重（比如取最新版本、第一条数据），避免填充结果出现歧义
• 大数据量场景优先选择Spark这类分布式框架，小数据量用Pandas或普通Java集合就足够高效

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