大型DataFrame关联更新：Union All后批量赋值高效方案求助

针对你提到的700万+1200万条记录的大数据场景，绝对不能用for循环逐行处理，以下是基于PySpark（分布式处理，适合超大数据量）和Pandas（单机内存足够时适用）的高效方案，核心都是用矢量化/分布式关联操作替代逐行循环：

一、PySpark 方案（推荐用于超大数据量） #

假设两个DataFrame的关联列为link_col，目标更新字段为family_number、emp_contract、emp_cert：

1. 先从emp_df提取关联列和目标字段并去重，得到参考映射表：

ref_df = emp_df.select("link_col", "family_number", "emp_contract", "emp_cert").distinct()

2. 对合并后的combined_df做左关联，用coalesce函数优先取参考表的字段值（覆盖原表空值或不正确的值）：

from pyspark.sql.functions import coalesce

updated_df = combined_df.join(ref_df, on="link_col", how="left") \
    .select(
        # 保留原表其他字段
        combined_df["*"],
        # 用参考表的值更新目标字段，原表有值则保留，无则替换
        coalesce(combined_df.family_number, ref_df.family_number).alias("family_number"),
        coalesce(combined_df.emp_contract, ref_df.emp_contract).alias("emp_contract"),
        coalesce(combined_df.emp_cert, ref_df.emp_cert).alias("emp_cert")
    )

或者用窗口函数按关联列分组，取组内第一个非空的目标字段值（适合同关联列有多条记录的场景）：

from pyspark.sql.window import Window
from pyspark.sql.functions import first, lit

# 按关联列分组，无需排序则用lit(1)占位
window_spec = Window.partitionBy("link_col").orderBy(lit(1))

updated_df = combined_df.withColumn(
    "family_number",
    first("family_number", ignorenulls=True).over(window_spec)
).withColumn(
    "emp_contract",
    first("emp_contract", ignorenulls=True).over(window_spec)
).withColumn(
    "emp_cert",
    first("emp_cert", ignorenulls=True).over(window_spec)
)

二、Pandas 方案（单机内存足够时适用） #

同样基于关联映射的思路，用矢量化操作替代循环：

1. 提取emp_df的关联列和目标字段并去重：

ref_df = emp_df[["link_col", "family_number", "emp_contract", "emp_cert"]].drop_duplicates(subset="link_col")

2. 方法一：合并后用combine_first更新字段

# 左关联参考表
merged = combined_df.merge(ref_df, on="link_col", suffixes=('', '_ref'), how='left')
# 用参考表的值覆盖原表空值
merged['family_number'] = merged['family_number_ref'].combine_first(merged['family_number'])
merged['emp_contract'] = merged['emp_contract_ref'].combine_first(merged['emp_contract'])
merged['emp_cert'] = merged['emp_cert_ref'].combine_first(merged['emp_cert'])
# 删除临时辅助列
updated_df = merged.drop(['family_number_ref', 'emp_contract_ref', 'emp_cert_ref'], axis=1)

3. 方法二：用字典映射批量更新

# 构建关联列到目标字段的映射字典
family_map = ref_df.set_index('link_col')['family_number'].to_dict()
contract_map = ref_df.set_index('link_col')['emp_contract'].to_dict()
cert_map = ref_df.set_index('link_col')['emp_cert'].to_dict()

# 批量更新字段，原表有值则保留，无则用映射值填充
updated_df = combined_df.copy()
updated_df['family_number'] = updated_df['link_col'].map(family_map).fillna(updated_df['family_number'])
updated_df['emp_contract'] = updated_df['link_col'].map(contract_map).fillna(updated_df['emp_contract'])
updated_df['emp_cert'] = updated_df['link_col'].map(cert_map).fillna(updated_df['emp_cert'])

以上方案都是基于底层优化的批量操作，效率比for循环高几个数量级，完全能处理千万级别的数据量。

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