大型PySpark DataFrame字符串包含关联查询性能优化求助

问题根源 #

你当前的left join用了variations.contains(keyword)这种非等值关联条件，Spark没办法使用高效的哈希/排序关联算法（比如Shuffle Hash Join、Sort Merge Join），只能退化为广播嵌套循环或笛卡尔积级别的计算，面对1000万+3000万的数据集，计算量直接爆炸，跑3小时很正常。

下面是针对性的优化步骤，按优先级排序：

1. 先解决代码里的致命问题：去掉result.show() #

你代码最后调用了result.show()，这会把整个结果集（可能上亿条）拉到Driver端，直接拖垮作业。生产环境绝对不能这么做，换成持久化到存储：

# 替换result.show()
result.write.mode("overwrite").parquet("join_result.parquet")

2. 预处理数据，砍掉无效计算 #

2.1 关键词去重 #

如果keywords_df里有重复的关键词，先去重，直接减少关联次数：

keywords_df = keywords_df.dropDuplicates(["keyword"])

2.2 提前过滤不可能匹配的行 #

给两个表加长度字段，先过滤掉变体字符串长度小于关键词的行（连长度都不够，肯定不可能包含）：

from pyspark.sql import functions as F

# 给变体表加长度列
variations_df = variations_df.withColumn("var_len", F.length(F.col("variations")))
# 给关键词表加长度列
keywords_df = keywords_df.withColumn("kw_len", F.length(F.col("keyword")))

# 关联时先加长度过滤条件，减少计算量
result = keywords_df.join(
    variations_df,
    (F.col("var_len") >= F.col("kw_len")) & (F.col("variations").contains(F.col("keyword"))),
    how='left'
)

3. 利用广播优化关联策略 #

如果关键词去重后数量降到几百万级别，尝试广播关键词表，让每个Executor本地加载关键词，避免大规模Shuffle：

3.1 开启广播并调整阈值 #

from pyspark.sql.functions import broadcast

# 广播关键词表
broadcast_keywords = broadcast(keywords_df)
# 执行关联
result = broadcast_keywords.join(
    variations_df,
    (F.col("var_len") >= F.col("kw_len")) & (F.col("variations").contains(F.col("keyword"))),
    how='left'
)

同时在spark-submit里调整广播阈值（默认10MB，改成适合的大小，比如100MB）：

--conf spark.sql.autoBroadcastJoinThreshold=104857600

如果关键词表太大（比如超过1GB），别强行广播，会撑爆Executor内存，改成分批次广播关联：

# 把关键词表分成10个批次
keyword_batches = keywords_df.randomSplit([0.1]*10)
# 每个批次关联后合并结果
result_df = None
for batch in keyword_batches:
    temp_df = broadcast(batch).join(variations_df, 
        (F.col("var_len") >= F.col("kw_len")) & (F.col("variations").contains(F.col("keyword"))),
        how='left'
    )
    if result_df is None:
        result_df = temp_df
    else:
        result_df = result_df.union(temp_df)

4. 调整Spark集群参数，适配大计算量 #

4.1 调整Shuffle分区数 #

默认的spark.sql.shuffle.partitions=200太少，对于3000万级别的数据，改成Executor总核数的2-3倍（你这里30个Executor*5核=150，所以设300-450）：

--conf spark.sql.shuffle.partitions=400

4.2 禁用Sort Merge Join偏好 #

因为是非等值关联，Sort Merge Join没用，强制Spark优先用广播嵌套循环：

--conf spark.sql.join.preferSortMergeJoin=false

4.3 用Cluster模式提交 #

你当前用的是deploy-mode client，Driver在本地，会有网络瓶颈，换成Cluster模式：

--deploy-mode cluster

5. 数据分区优化，减少跨节点Shuffle #

对两个表按字符串哈希值分区，让可能匹配的行落在同一个Executor上：

# 按哈希值取模分区，分区数和Shuffle分区数一致
part_num = 400
variations_df = variations_df.repartition(part_num, F.hash("variations") % part_num)
keywords_df = keywords_df.repartition(part_num, F.hash("keyword") % part_num)

同时提前持久化分区后的表，避免重复计算：

variations_df = variations_df.cache()
keywords_df = keywords_df.cache()

6. 换用更高效的字符串匹配逻辑 #

如果你的关键词是短语（2-5词），可以把变体字符串拆成词数组，用array_contains的组合来匹配（前提是关键词作为整体短语出现在变体里，或者关键词的所有词都出现）：

# 把变体字符串拆成词数组（假设空格分隔）
variations_df = variations_df.withColumn("var_words", F.split(F.col("variations"), " "))
# 把关键词拆成词数组
keywords_df = keywords_df.withColumn("kw_words", F.split(F.col("keyword"), " "))

# 检查关键词的所有词都在变体词数组里（如果需求是关键词的词都出现，不是整体包含）
from pyspark.sql.functions import array_intersect, size
result = keywords_df.join(
    variations_df,
    size(array_intersect(F.col("kw_words"), F.col("var_words"))) == size(F.col("kw_words")),
    how='left'
)

如果必须匹配整体短语，可以用滑动窗口生成变体的n-gram短语，然后和关键词做等值关联，这会大幅提升效率：

# 生成变体字符串的n-gram短语（n对应关键词的词数，这里以3词为例）
from pyspark.ml.feature import NGram
ngram = NGram(n=3, inputCol="var_words", outputCol="var_ngrams")
variations_ngram_df = ngram.transform(variations_df)

# 然后关联条件变成关键词在变体的ngram数组里
result = keywords_df.join(
    variations_ngram_df,
    F.array_contains(F.col("var_ngrams"), F.col("keyword")),
    how='left'
)

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