可以拆分和分解b列，然后进行左连接。然后收集d值并用逗号连接。

import pyspark.sql.functions as F

result = source_df.withColumn(
    'B_split',
    F.explode(F.split('B', ','))
).alias('s').join(
    lookup_df.alias('l'),
    F.expr('s.B_split = l.C'),
    'left'
).drop('C').na.fill(
    'EMPTY', ['D']
).groupBy(
    source_df.columns
).agg(
    F.concat_ws(',', F.collect_list('D')).alias('new_col')
)

result.show()
+---+-----+---+-----------+
|  A|    B|  T|    new_col|
+---+-----+---+-----------+
|foo|a,b,c|sam|   h1,h2,h3|
|faz|b,a,f|sam|h2,h1,EMPTY|
|bar|k,a,c|bob|EMPTY,h1,h3|
+---+-----+---+-----------+

这是另一种不必爆炸第一个Dataframe的方法。直接加入，如： ',B,' LIKE '%,C,%' . 然后groupby A 将Map收集到Map列中，并在列的拆分上使用transform B 为每个元素获取相应的替换项。
最后，使用 array_join 函数，您可以从转换的数组结果中获得逗号分隔的列表，并将空值替换为 EMPTY :

from pyspark.sql import functions as F

df1 = df.join(
    lookup_df,
    F.expr("concat(',', B, ',') like concat('%,', C, ',%')"),
    "left"
).groupby("A").agg(
    *[F.first(c).alias(c) for c in df.columns if c != "A"],
    F.map_from_entries(
        F.collect_list(
            F.struct(F.col("C"), F.col("D"))
        )
    ).alias("mappings")
).select(
    F.col("A"),
    F.col("B"),
    F.expr("array_join(transform(split(B, ','), x -> mappings[x]), ',', 'EMPTY')").alias("new_col"),
    *[F.col(c) for c in df.columns if c not in ("A", "B")]
)

df1.show()

# +---+-----+-----------+---+

# |  A|    B|    new_col|  T|

# +---+-----+-----------+---+

# |faz|b,a,f|h2,h1,EMPTY|sam|

# |bar|k,a,c|EMPTY,h1,h3|bob|

# |foo|a,b,c|   h1,h2,h3|sam|

# +---+-----+-----------+---+