使用PySpark实现两个大型CSV文件内连接的高效方案咨询

你当前用Spark做内连接的方向是完全正确的，没必要退回去手写原生MapReduce。Spark本身就是在MapReduce计算模型上做了大量优化的分布式引擎，内置的执行优化、内存计算、Shuffle优化比手写MR的性能高得多，你现在的代码问题只是没做针对性调优，不是选型错误。

你现在直接读CSV的写法在超大文件场景下浪费非常多性能，先改这两点：

• 读文件时显式指定Schema，不要用Spark默认的自动类型推断。自动推断需要先全量扫描一遍文件才能确定字段类型，TB级数据光这一步就能浪费几十分钟。而且定义Schema的时候只列你实际要用的字段，不需要的字段Spark会直接跳过，不读入内存：

from pyspark.sql.types import StructType, StructField, StringType
# df1只保留需要的id、name字段
df1_schema = StructType([
    StructField("id", StringType(), nullable=False),
    StructField("name", StringType(), nullable=True)
])
df1 = spark.read.csv('df1.csv', sep=r'\t', header=True, schema=df1_schema).select("id", "name")

# df2只保留需要的id、title字段
df2_schema = StructType([
    StructField("id", StringType(), nullable=False),
    StructField("title", StringType(), nullable=True)
])
df2 = spark.read.csv('df2.csv', sep=r'\t', header=True, schema=df2_schema).select("id", "title")

• 如果有数据过滤规则，比如过滤空id、不符合规则的name/title，直接在读取后立刻加filter，不要等Join完再过滤，尽可能把参与Shuffle的数据量压到最小。

Spark SQL本身已经做了很多Join优化，但超大表场景可以手动干预进一步提效：

• 先判断两个表的大小，如果其中一个表解压后在2GB以内（具体阈值看你集群Executor的内存配置），直接用广播Join，完全省掉大表的Shuffle过程，性能会提升一个数量级：

# 注册临时表后，SQL里加广播提示即可
query = """SELECT /*+ BROADCAST(d2) */ d1.id, d2.id, d1.name, d2.title
            FROM d1
            INNER JOIN d2
            ON d1.id = d2.id;
            """

• 如果两个表都是TB级超大表，没法广播，就先按Join键id做重分区，把相同id的数据提前分到同一个分区，避免Join阶段的二次Shuffle，分区数设置为集群总CPU核数的2~3倍即可：

# 比如集群总共有500个核，就设成1000~1500个分区
df1 = df1.repartition(1200, "id")
df2 = df2.repartition(1200, "id")

• 如果遇到个别id对应数据量极大的数据倾斜问题，直接对倾斜键加随机前缀做拆分Join，Spark有成熟的处理方案，不需要自己从零写逻辑。

很多人以为手写MR更底层就更快，实际完全不是：

• 原生MR每次Shuffle的中间结果必须落磁盘，没有Spark的内存计算、Tungsten堆外内存优化、全链路代码生成，IO和执行效率天生比Spark差一大截
• 你手写MR实现内连接，本质就是重复实现Spark内置的SortMerge Join逻辑，要自己处理分区、排序、分组、数据倾斜、异常处理，几百行代码写出来的实现，大概率跑不过Spark默认优化过的Join逻辑
• 你后续还要对Join结果做其他处理，Spark可以把整个计算链路串起来做全局优化，MR的话每一步计算都要把结果落盘，再读入下一个任务，额外IO开销非常大。

如果这两个CSV需要定期重复处理，第一次跑完就把数据转成Parquet或者ORC列式存储格式，下次处理直接读列式文件，性能会比读CSV快5~10倍：

• 列式存储压缩比更高，自带列裁剪、谓词下推能力，读的时候不需要扫描全量字段
• 存的时候直接按id做分区，后续Join的时候完全不需要做Shuffle，直接读取对应分区的数据就能完成Join。

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