基于PySpark DataFrame实现含通配符映射匹配的技术方案问询

针对带*通配符的映射规则匹配问题，常规Join无法处理通配符的任意匹配逻辑，我们可以通过交叉连接+过滤+优先级排序的方式实现需求，具体步骤如下：

1. 修正示例代码（修复语法错误） #

原示例代码存在两处语法问题，修正后可正常运行：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, row_number
from pyspark.sql.window import Window

# 创建Spark会话
spark = SparkSession.builder.appName("wildcard_mapping").getOrCreate()

# 映射表数据
map_data = [('a', 'b', 'c', 'good'), ('a', 'a', '*', 'very good'), 
            ('b', 'd', 'c', 'bad'), ('a', 'b', 'a', 'very good'),
            ('c', 'c', '*', 'very bad'), ('a', 'b', 'b', 'bad')]
mapping_columns = ["col1", "col2", 'col3', 'result']
mapping_table = spark.createDataFrame(map_data, mapping_columns)

# 业务数据
data = [('a', 'b', 'c'), ('a', 'a', 'b' ), 
        ('c', 'c', 'a' ), ('c', 'c', 'b' ),
        ('a', 'b', 'b'), ('a', 'a', 'd')]
df_columns = ["col1", "col2", 'col3']
df = spark.createDataFrame(data, df_columns)

2. 核心实现逻辑 #

步骤1：重命名映射表字段（避免冲突） #

先给映射表的字段添加前缀，防止交叉连接后字段名重复：

mapping_table_renamed = mapping_table.withColumnRenamed("col1", "map_col1")\
                                     .withColumnRenamed("col2", "map_col2")\
                                     .withColumnRenamed("col3", "map_col3")

步骤2：交叉连接并过滤匹配规则 #

将业务表和映射表做交叉连接，然后筛选出符合通配符规则的匹配行：

# 交叉连接
cross_df = df.crossJoin(mapping_table_renamed)

# 过滤匹配条件：字段相等 或 映射字段为*
matched_df = cross_df.filter(
    ((col("col1") == col("map_col1")) | (col("map_col1") == "*")) &
    ((col("col2") == col("map_col2")) | (col("map_col2") == "*")) &
    ((col("col3") == col("map_col3")) | (col("map_col3") == "*"))
)

步骤3：计算规则优先级 #

精确匹配的字段数量越多，规则优先级越高（通配符匹配的字段越少，规则越精准）：

priority_df = matched_df.withColumn(
    "exact_match_count",
    when(col("col1") == col("map_col1"), 1).otherwise(0) +
    when(col("col2") == col("map_col2"), 1).otherwise(0) +
    when(col("col3") == col("map_col3"), 1).otherwise(0)
)

步骤4：筛选最优匹配结果 #

对每条业务数据，保留优先级最高的映射结果；若存在多个同优先级规则，默认取第一条：

# 按业务行分组，按优先级降序排序
window_spec = Window.partitionBy("col1", "col2", "col3").orderBy(col("exact_match_count").desc())

final_df = priority_df.withColumn("row_rank", row_number().over(window_spec))\
                      .filter(col("row_rank") == 1)\
                      .drop("map_col1", "map_col2", "map_col3", "exact_match_count", "row_rank")

# 查看最终结果
final_df.show()

3. 边界情况处理（无匹配规则的行） #

如果需要给没有匹配到任何规则的业务行设置默认值，可以通过左连接实现：

# 先获取匹配结果
matched_result = priority_df.withColumn("row_rank", row_number().over(window_spec))\
                            .filter(col("row_rank") == 1)\
                            .drop("map_col1", "map_col2", "map_col3", "exact_match_count", "row_rank")

# 左连接到原业务表，填充默认值
final_df_with_default = df.join(matched_result, on=["col1", "col2", "col3"], how="left")\
                          .fillna({"result": "no_match"})

final_df_with_default.show()

运行结果示例 #

最终输出会为每条业务数据匹配到最精准的映射结果，比如：

• ('c', 'c', 'a')会匹配('c', 'c', '*', 'very bad')
• ('a', 'a', 'b')会匹配('a', 'a', '*', 'very good')

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