SparkSQL中产生笛卡尔积的几种典型场景以及处理策略

【前言：如果你经常使用Spark SQL进行数据的处理分析，那么对笛卡尔积的危害性一定不陌生，比如大量占用集群资源导致其他任务无法正常执行，甚至导致节点宕机。那么都有哪些情况会产生笛卡尔积，以及如何事前"预测"写的SQL会产生笛卡尔积从而避免呢？（以下不考虑业务需求确实需要笛卡尔积的场景）】

Spark SQL几种产生笛卡尔积的典型场景

首先来看一下在Spark SQL中产生笛卡尔积的几种典型SQL：

1. join语句中不指定on条件

select * from test_partition1 join test_partition2;

2. join语句中指定不等值连接

select * from test_partition1 t1 inner join test_partition2 t2 on t1.name <> t2.name;

3. join语句on中用or指定连接条件

select * from test_partition1 t1 join test_partition2 t2 on t1.id = t2.id or t1.name = t2.name;

4. join语句on中用||指定连接条件

select * from test_partition1 t1 join test_partition2 t2 on t1.id = t2.id || t1.name = t2.name;

除了上述举的几个典型例子，实际业务开发中产生笛卡尔积的原因多种多样。
同时需要注意，在一些SQL中即使满足了上述4种规则之一也不一定产生笛卡尔积。比如，对于join语句中指定不等值连接条件的下述SQL不会产生笛卡尔积:

--在Spark SQL内部优化过程中针对join策略的选择，最终会通过SortMergeJoin进行处理。

select * from test_partition1 t1 join test_partition2 t2 on t1.id = t2.id and t1.name<>t2.name;

此外，对于直接在SQL中使用cross join的方式，也不一定产生笛卡尔积。比如下述SQL：

-- Spark SQL内部优化过程中选择了SortMergeJoin方式进行处理

select * from test_partition1 t1 cross  join test_partition2 t2 on t1.id = t2.id;

但是如果cross join没有指定on条件同样会产生笛卡尔积。那么如何判断一个SQL是否产生了笛卡尔积呢？

Spark SQL是否产生了笛卡尔积？

以join语句不指定on条件产生笛卡尔积的SQL为例：

-- test_partition1和test_partition2是Hive分区表

select * from test_partition1 join test_partition2;

通过Spark UI上SQL一栏查看上述SQL执行图，如下：
SparkSQL中产生笛卡尔积的几种典型场景以及处理策略

可以看出，因为该join语句中没有指定on连接查询条件，导致了CartesianProduct即笛卡尔积。

再来看一下该join语句的逻辑计划和物理计划：

== Parsed Logical Plan ==

'GlobalLimit 1000

+- 'LocalLimit 1000

   +- 'Project [*]

      +- 'UnresolvedRelation `t`

== Analyzed Logical Plan ==

id: string, name: string, dt: string, id: string, name: string, dt: string

GlobalLimit 1000

+- LocalLimit 1000

   +- Project [id#84, name#85, dt#86, id#87, name#88, dt#89]

      +- SubqueryAlias `t`

         +- Project [id#84, name#85, dt#86, id#87, name#88, dt#89]

            +- Join Inner

               :- SubqueryAlias `default`.`test_partition1`

               :  +- HiveTableRelation `default`.`test_partition1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#84, name#85], [dt#86]

               +- SubqueryAlias `default`.`test_partition2`

                  +- HiveTableRelation `default`.`test_partition2`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#87, name#88], [dt#89]

== Optimized Logical Plan ==

GlobalLimit 1000

+- LocalLimit 1000

   +- Join Inner

      :- HiveTableRelation `default`.`test_partition1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#84, name#85], [dt#86]

      +- HiveTableRelation `default`.`test_partition2`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#87, name#88], [dt#89]

== Physical Plan ==

CollectLimit 1000

+- CartesianProduct

   :- Scan hive default.test_partition1 [id#84, name#85, dt#86], HiveTableRelation `default`.`test_partition1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#84, name#85], [dt#86]

   +- Scan hive default.test_partition2 [id#87, name#88, dt#89], HiveTableRelation `default`.`test_partition2`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe, [id#87, name#88], [dt#89]

通过逻辑计划到物理计划，以及最终的物理计划选择CartesianProduct，可以分析得出该SQL最终确实产生了笛卡尔积。

Spark SQL中产生笛卡尔积的处理策略

在之前的文章中《Spark SQL如何选择join策略》已经介绍过，Spark SQL中主要有ExtractEquiJoinKeys（Broadcast Hash Join、Shuffle Hash Join、Sort Merge Join，这3种是我们比较熟知的Spark SQL join）和Without joining keys（CartesianProduct、BroadcastNestedLoopJoin）join策略。

那么，如何判断SQL是否产生了笛卡尔积就迎刃而解。

1. 在利用Spark SQL执行SQL任务时，通过查看SQL的执行图来分析是否产生了笛卡尔积。如果产生笛卡尔积，则将任务杀死，进行任务优化避免笛卡尔积。【不推荐。用户需要到Spark UI上查看执行图，并且需要对Spark UI界面功能等要了解，需要一定的专业性。（注意：这里之所以这样说，是因为Spark SQL是计算引擎，面向的用户角色不同，用户不一定对Spark本身了解透彻，但熟悉SQL。对于做平台的小伙伴儿，想必深有感触）】

2. 分析Spark SQL的逻辑计划和物理计划，通过程序解析计划推断SQL最终是否选择了笛卡尔积执行策略。如果是，及时提示风险。具体可以参考Spark SQL join策略选择的源码：

def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {

// --- BroadcastHashJoin --------------------------------------------------------------------

// broadcast hints were specified

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)

if canBroadcastByHints(joinType, left, right) =>

        val buildSide = broadcastSideByHints(joinType, left, right)

Seq(joins.BroadcastHashJoinExec(

          leftKeys, rightKeys, joinType, buildSide, condition, planLater(left), planLater(right)))

// broadcast hints were not specified, so need to infer it from size and configuration.

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)

if canBroadcastBySizes(joinType, left, right) =>

        val buildSide = broadcastSideBySizes(joinType, left, right)

Seq(joins.BroadcastHashJoinExec(

          leftKeys, rightKeys, joinType, buildSide, condition, planLater(left), planLater(right)))

// --- ShuffledHashJoin ---------------------------------------------------------------------

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)

if !conf.preferSortMergeJoin && canBuildRight(joinType) && canBuildLocalHashMap(right)

           && muchSmaller(right, left) ||

           !RowOrdering.isOrderable(leftKeys) =>

Seq(joins.ShuffledHashJoinExec(

          leftKeys, rightKeys, joinType, BuildRight, condition, planLater(left), planLater(right)))

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)

if !conf.preferSortMergeJoin && canBuildLeft(joinType) && canBuildLocalHashMap(left)

           && muchSmaller(left, right) ||

           !RowOrdering.isOrderable(leftKeys) =>

Seq(joins.ShuffledHashJoinExec(

          leftKeys, rightKeys, joinType, BuildLeft, condition, planLater(left), planLater(right)))

// --- SortMergeJoin ------------------------------------------------------------

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)

if RowOrdering.isOrderable(leftKeys) =>

        joins.SortMergeJoinExec(

          leftKeys, rightKeys, joinType, condition, planLater(left), planLater(right)) :: Nil

// --- Without joining keys ------------------------------------------------------------

// Pick BroadcastNestedLoopJoin if one side could be broadcast

case j @ logical.Join(left, right, joinType, condition)

if canBroadcastByHints(joinType, left, right) =>

        val buildSide = broadcastSideByHints(joinType, left, right)

        joins.BroadcastNestedLoopJoinExec(

          planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

case j @ logical.Join(left, right, joinType, condition)

if canBroadcastBySizes(joinType, left, right) =>

        val buildSide = broadcastSideBySizes(joinType, left, right)

        joins.BroadcastNestedLoopJoinExec(

          planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

// Pick CartesianProduct for InnerJoin

case logical.Join(left, right, _: InnerLike, condition) =>

        joins.CartesianProductExec(planLater(left), planLater(right), condition) :: Nil

case logical.Join(left, right, joinType, condition) =>

        val buildSide = broadcastSide(

left.stats.hints.broadcast, right.stats.hints.broadcast, left, right)

// This join could be very slow or OOM

        joins.BroadcastNestedLoopJoinExec(

          planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

// --- Cases where this strategy does not apply ---------------------------------------------

case _ => Nil

    }

此外，在业务开发中，要不断总结归纳产生笛卡尔积的情况，形成知识文档，以便在后续业务开发中避免类似的情况出现。

除了笛卡尔积效率比较低，BroadcastNestedLoopJoin效率也相对低效，尤其是当数据量大的时候还很容易造成driver端的OOM，这种情况也是需要极力避免的。

秒客网

SparkSQL中产生笛卡尔积的几种典型场景以及处理策略

相关文章