PySpark如何基于动态长度列表动态生成like匹配条件新增标记列

场景说明 #

现有长度动态增减的元素列表，以及PySpark DataFrame，需要新增ind列：当MSH字段包含任意一个被~包裹的列表元素时，ind取1，否则取0。
初始示例代码如下：

my_list = ['4587','9920408','9920316']

a=spark.createDataFrame([
    (101,'~1~20448~3~22901~12214~27681~9920408~20013~19957~19993~ ~ ~ ~ ~ ~'),
    (102, '~1~20448~4462~4586~24739~4587~9914381~9921471~12777~ ~ ~ ~ ~ ~ ~'),
    (103,'~1~20448~3~22901~3891~4148~9920948~14845~4230~ ~ ~ ~ ~ ~ ~'),
    (104, '~1~20448~3~22901~3891~4148~9920316~4211~4212~ ~ ~ ~ ~ ~ ~')
], ['ID', 'MSH'])

原有硬编码实现可以正常运行，但无法适配列表动态变化：

b=a.withColumn('ind', F.expr("if((msh like '%~4587~%' or msh like '%~9920408~%' or msh like '%~9920316~%'),1,0)"))

可用实现方案 #

方案1：动态拼接SQL表达式（兼容原有逻辑） #

直接遍历动态列表自动生成所有LIKE子句，用OR拼接为完整判断表达式，和原有硬编码逻辑完全一致，无兼容问题：

from pyspark.sql import functions as F

# 自动生成每个元素对应的like判断子句
like_conditions = [f"msh like '%~{elem}~%'" for elem in my_list]
# 拼接为完整的if判断表达式
judge_expr = f"if(({' or '.join(like_conditions)}), 1, 0)"
# 执行计算
b = a.withColumn('ind', F.expr(judge_expr))

• 优势：和原有写法执行逻辑、执行计划完全一致，改造成本为0
• 注意：如果列表元素包含SQL特殊字符（如单引号），需要提前做转义处理

方案2：数组交集判断（大列表下性能更优） #

先将MSH字段按~分隔拆分为数组，再通过数组交集判断是否存在匹配元素，无需拼接长SQL语句，列表元素越多性能优势越明显：

from pyspark.sql import functions as F

b = a.withColumn(
    'ind',
    F.when(
        # 拆分MSH为数组，与目标列表求交集，交集长度大于0即代表匹配成功
        F.size(
            F.array_intersect(
                F.split(F.col('MSH'), '~'),
                F.array(*[F.lit(elem) for elem in my_list])
            )
        ) > 0,
        1
    ).otherwise(0)
)

• 优势：无需拼接字符串，不存在特殊字符转义问题，大数据量/大列表场景下执行效率远高于多OR LIKE写法
• 逻辑等价性：原MSH为~分隔的字符串，拆分后自动过滤空值匹配，和%~xxx~%的匹配规则完全等价

两种方案都可以自动适配任意长度的my_list，列表增删元素时不需要修改匹配逻辑代码。

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