编写函数比较两个DataFrame对应字段值，返回布尔判断结果

我针对你给出的Spark DataFrame结构来实现方案，如果是用pandas的话也补充了简化版本，你可以按需选用：

任务1：比较两个DataFrame并返回True/False #

如果要判断两个DataFrame完全相等（结构、字段名、所有数据内容都一致），可以用以下函数：

def compare_dataframes(df_a, df_b):
    # 先校验结构是否一致
    if df_a.schema != df_b.schema:
        return False
    # 再校验行数是否相同
    if df_a.count() != df_b.count():
        return False
    # 最后校验数据内容无差异
    return df_a.subtract(df_b).count() == 0

这个函数会逐层排查：先对比schema，再对比行数，最后用subtract找出差异行，无差异则返回True。

如果只需要对比特定字段的内容，比如仅客户相关字段，可以用这个扩展版：

def compare_specific_fields(df_a, df_b, fields_a, fields_b):
    if len(fields_a) != len(fields_b):
        return False
    # 重命名字段对齐后再对比
    df_a_aligned = df_a.select([col(a).alias(b) for a, b in zip(fields_a, fields_b)])
    return compare_dataframes(df_a_aligned, df_b.select(fields_b))

调用示例：compare_specific_fields(df1, df2, ["Customer_ID", "Customer_Name"], ["CustomerID", "CustomerName"])

任务2：创建客户字段匹配的函数 #

这里分两种常见场景，你可以根据需求选择：

场景1：检查df2中是否存在至少一行匹配df1的客户信息 #

def has_matching_customer(df1, df2):
    # 通过内连接统计匹配行数
    matched_rows = df2.join(
        df1,
        (df2.CustomerID == df1.Customer_ID) & (df2.CustomerName == df1.Customer_Name),
        how="inner"
    ).count()
    return matched_rows > 0

只要有任意一行匹配，函数就返回True，否则返回False。

场景2：给df2每行添加匹配标记（True/False） #

def add_customer_match_flag(df1, df2):
    # 提取df1的客户信息作为查找表
    df1_lookup = df1.select(
        col("Customer_ID").alias("match_id"),
        col("Customer_Name").alias("match_name")
    ).distinct()
    
    # 左连接后标记是否匹配
    return df2.join(
        df1_lookup,
        (df2.CustomerID == df1_lookup.match_id) & (df2.CustomerName == df1_lookup.match_name),
        how="left"
    ).withColumn(
        "is_customer_match",
        col("match_id").isNotNull()
    ).drop("match_id", "match_name")

调用后df2会新增is_customer_match列，每行对应是否匹配到df1的客户。

如果是用pandas实现任务2的场景1，会更简洁：

def has_matching_customer_pandas(df1, df2):
    # 把客户字段组合成元组集合，检查交集
    df1_customer_pairs = set(zip(df1["Customer_ID"], df1["Customer_Name"]))
    df2_customer_pairs = set(zip(df2["CustomerID"], df2["CustomerName"]))
    return len(df1_customer_pairs & df2_customer_pairs) > 0

内容的提问来源于stack exchange，提问作者S M