如何在Spark中执行多DataFrame连接？大表连接性能优化求助

咱们先拆解下你遇到的问题：4个各1000万行的Cassandra表，基于4个列做连接，运行慢大概率是数据倾斜、分区策略不合理、Cassandra读取未优化或者连接策略没选对导致的。下面给你一步步来优化：

一、先优化Cassandra数据读取环节 #

从Cassandra读数据时，默认的分区可能不匹配你的连接键，先做这几点：

• 对齐Spark与Cassandra的分区：如果你的Cassandra表分区键包含连接用的time/regionid这类字段，读表时指定分区器，让Spark分区和Cassandra分区对齐，减少后续shuffle：

  df1 = sql.read.format("org.apache.spark.sql.cassandra").\
    option("spark.cassandra.input.partitioner.class", "com.datastax.spark.connector.rdd.partitioner.DynamicClusterPartitioner").\
    load(keyspace="db", table="table1").\
    select('time','regionid','wilayaid','siteid','col1', 'col2','col3','col4','col5')
  

• 提前过滤数据：如果不需要全表数据，先用where限定范围（比如时间、区域），直接减少后续处理的数据量：

  df1 = sql.read.format("org.apache.spark.sql.cassandra").\
    load(keyspace="db", table="table1").\
    where("time >= '2024-01-01' and regionid in ('r1','r2')").\
    select('time','regionid','wilayaid','siteid','col1', 'col2','col3','col4','col5')
  

• 调整读取并行度：增大Cassandra单次读取行数，同时匹配Spark shuffle分区数（建议设置为集群核数的2-3倍）：

  # 读取时调整fetch size
  df1 = sql.read.format("org.apache.spark.sql.cassandra").\
    option("spark.cassandra.input.fetch.size_in_rows", "10000").\
    load(keyspace="db", table="table1")
  
  # 设置shuffle分区数
  sql.setConf("spark.sql.shuffle.partitions", "300")
  

二、解决数据倾斜问题 #

多列连接最容易出现的就是数据倾斜——某几个连接键组合对应的数据量特别大，导致单个任务卡死：

• 先检查连接键分布：统计每个连接键组合的行数，定位异常大的分组：

  from pyspark.sql.functions import desc
  df1.groupBy('time','regionid','wilayaid','siteid').count().orderBy(desc("count")).show(20)
  

• 拆分大分组单独处理：如果发现某个组合数据量远超其他，给该组合的连接键加随机后缀（比如time加_rand），拆分成多个小分组连接后再合并，其余正常分组按原逻辑连接。
• 广播小表：如果其中某个DataFrame经过过滤后尺寸远小于其他（比如只有几百万行），用broadcast把小表广播到所有节点，避免大shuffle：

  from pyspark.sql.functions import broadcast
  joined_df = df1.join(broadcast(df2), on=['time','regionid','wilayaid','siteid'], how='inner')
  

  注意：广播表建议不超过1GB，否则会占用过多内存。

三、优化连接策略 #

Spark会自动选择连接策略，但你可以根据场景手动调整：

• 大表连接用Sort Merge Join：如果所有表都超过1GB，开启Sort Merge Join（适合大表且连接键有序的场景）：

  sql.setConf("spark.sql.join.preferSortMergeJoin", "true")
  

  可以提前对每个DataFrame按连接键排序，进一步提升连接效率。
• 严格检查连接条件：确保连接条件是完整的4个列，避免漏写导致意外的笛卡尔积（直接让数据量爆炸）。

四、其他通用优化 #

• 持久化中间结果：如果多个连接用到同一个DataFrame，先持久化到内存+磁盘，避免重复读取Cassandra：

  from pyspark.storagelevel import StorageLevel
  df1.persist(StorageLevel.MEMORY_AND_DISK)
  df1.count() # 触发持久化
  

• 调整资源配置：给Executor和Driver分配足够内存，比如：

  # 提交任务时指定
  spark-submit --executor-memory 8g --driver-memory 4g --executor-cores 4 your_script.py
  

你可以先从数据倾斜检查和Cassandra读取优化入手，这两个是大表连接慢最常见的原因。

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