本地RDD注册的表与DB表执行join操作的底层运行机制是什么

实际默认运行逻辑 #

你遇到的慢是因为默认情况下Spark会先全量拉取DB端的查询结果到本地集群后，再执行和本地RDD表的关联计算，不会自动把关联逻辑下推到DB端执行，原因如下：

• Spark JDBC数据源的默认下推能力仅支持简单的列裁剪、谓词过滤，当你把DB侧多表Join的查询结果注册为Spark临时表后，Spark会将其视为普通的分布式数据源，无法感知外部数据库的内部表结构、索引规则，也没法把存在Spark内存中的本地RDD数据推给外部DB做关联，自然不可能把整个Join逻辑下发到DB执行，你认为后者不合理的判断是正确的。
• 如果DB侧的多表Join返回结果数据量较大，整个任务的耗时基本都花在了全量数据的网络传输上，哪怕DB侧单独执行多表Join很快，传输大量数据也会导致整体速度极慢。

针对性优化方案 #

因为你的本地RDD仅20条，属于极小数据集，可以通过以下两种方式快速解决性能问题：

• 手动下推关联条件：把本地RDD里的关联键值直接拼到读取DB的SQL语句中，比如关联键是用户ID，就把20个ID拼成user_id in (id1,id2,...id20)的条件加到DB侧的查询里，让DB直接返回符合关联条件的少量数据，再拉到Spark侧做后续处理，耗时基本和DB单独执行查询差不多。
• 启用Spark广播Join：将小体量的本地RDD表广播到所有Executor节点，避免关联时的Shuffle开销，同时如果配置了JDBC并行拉取参数，还能在每个拉取分区里直接做关联过滤，代码示例如下：

import org.apache.spark.sql.functions.broadcast

// 将本地小RDD转成DataFrame后标记为广播
val smallBroadcastDF = broadcast(localRdd.toDF("join_key", "local_col1", "local_col2"))
// 和JDBC读取的DB大表执行关联
val joinResult = jdbcReadDF.join(smallBroadcastDF, Seq("join_key"), "inner")

如果需要进一步提升DB数据拉取速度，还可以配置JDBC的partitionColumn、lowerBound、upperBound、numPartitions参数，开启多线程并行拉取DB数据。

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