Spark DataFrame中同一transaction_id的transaction_label对比问题

我们可以通过分组聚合+条件判断的方式在Spark原生环境中完成验证，无需转换为Pandas，从根源避免Driver内存不足的问题。

实现思路 #

1. 按transaction_id分组，分别提取module_name='mcc'和module_name='regex'对应的transaction_label
2. 对regex类型的标签截取前5个字符，与mcc类型的标签做匹配校验
3. 输出验证结果，同时保留原始数据的关键标识信息

代码实现 #

方式1：使用Spark SQL #

WITH trans_grouped AS (
    SELECT 
        transaction_id,
        -- 提取mcc对应的标签
        MAX(CASE WHEN module_name = 'mcc' THEN transaction_label END) AS mcc_label,
        -- 提取regex标签并截取前5位
        MAX(CASE WHEN module_name = 'regex' THEN SUBSTRING(transaction_label, 1, 5) END) AS regex_prefix
    FROM all_trans
    GROUP BY transaction_id
)
SELECT 
    transaction_id,
    mcc_label,
    regex_prefix,
    -- 生成验证结果
    CASE 
        WHEN mcc_label = regex_prefix THEN '匹配'
        WHEN mcc_label IS NULL AND regex_prefix IS NULL THEN '无标签'
        WHEN mcc_label IS NULL THEN '缺少mcc标签'
        WHEN regex_prefix IS NULL THEN '缺少regex标签'
        ELSE '不匹配'
    END AS validation_result
FROM trans_grouped

方式2：使用Spark DataFrame API #

from pyspark.sql import functions as F

result_df = (
    all_trans
    .groupBy("transaction_id")
    .agg(
        F.max(F.when(F.col("module_name") == "mcc", F.col("transaction_label"))).alias("mcc_label"),
        F.max(F.when(F.col("module_name") == "regex", F.substring(F.col("transaction_label"), 1, 5))).alias("regex_prefix")
    )
    .withColumn(
        "validation_result",
        F.when(F.col("mcc_label") == F.col("regex_prefix"), "匹配")
        .when(F.col("mcc_label").isNull() & F.col("regex_prefix").isNull(), "无标签")
        .when(F.col("mcc_label").isNull(), "缺少mcc标签")
        .when(F.col("regex_prefix").isNull(), "缺少regex标签")
        .otherwise("不匹配")
    )
)

result_df.show()

结果说明 #

• 同时存在mcc和regex标签的transaction_id，直接对比mcc_label与regex标签的前5位字符
• 仅存在单一标签的transaction_id，会标注对应缺失信息
• 无任何标签的transaction_id，统一标注为"无标签"

内容的提问来源于stack exchange，提问作者Stanislav Jirak