您可以使用like或contains操作符，并使用reduce创建条件链。

anomalous_blocks = ['blk_A','blk_C','blk_D']
label_condition = reduce(lambda a, b: a | b, 
                         [func.col('log').like('%'+k+'%') for k in anomalous_blocks]
                         )
# Column<'((log LIKE %blk_A% OR log LIKE %blk_C%) OR log LIKE %blk_D%)'>
data_sdf. \
    withColumn('label', label_condition). \
    show()
# +---+---------------+-----+
# | id|            log|label|
# +---+---------------+-----+
# |  1|Test logX blk_A| true|
# |  2|Test logV blk_B|false|
# |  3|Test logF blk_D| true|
# |  4|Test logD blk_F|false|
# |  5|Test logB blk_K|false|
# |  6|Test logY blk_A| true|
# |  7|Test logE blk_C| true|
# +---+---------------+-----+

您可以在pyspark.sql.Column上使用isin方法来实现这一点，而不需要UDF（请注意，我稍微修改了anomalous_blocks列表的内容，以便与df的内容完全匹配。由于您使用的是N >> M，因此这应该非常便宜）：

df = spark.createDataFrame(
    [
        (1, "Test log blk_A"),
        (2, "Test log blk_B"),
        (3, "Test log blk_D"),
        (4, "Test log blk_F"),
        (5, "Test log blk_K"),
        (6, "Test log blk_A"),
        (7, "Test log blk_C")
    ],
    ["id", "log"]
)
anomalous_blocks = ['blk_A','blk_C','blk_D']
# Solution starts here
adapted_anomalous_blocks = ["Test log " + x for x in anomalous_blocks]
output = df.withColumn("Label", df.log.isin(adapted_anomalous_blocks))
output.show()
+---+--------------+-----+
| id|           log|Label|
+---+--------------+-----+
|  1|Test log blk_A| true|
|  2|Test log blk_B|false|
|  3|Test log blk_D| true|
|  4|Test log blk_F|false|
|  5|Test log blk_K|false|
|  6|Test log blk_A| true|
|  7|Test log blk_C| true|
+---+--------------+-----+