scala - 在 spark 数据集中加入两个聚簇表似乎最终会完全洗牌

Question

我有两个 hive 聚簇表 t1 和 t2

CREATE EXTERNAL TABLE `t1`(
  `t1_req_id` string,
   ...
PARTITIONED BY (`t1_stats_date` string)
CLUSTERED BY (t1_req_id) INTO 1000 BUCKETS
// t2 looks similar with same amount of buckets

插入部分发生在蜂巢中

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
insert overwrite table `t1` partition(t1_stats_date,t1_stats_hour)
   select *
   from t1_raw
   where t1_stats_date='2020-05-10' and t1_stats_hour='12' AND 
   t1_req_id is not null

代码如下所示：

 val t1 = spark.table("t1").as[T1]
 val t2=  spark.table("t2").as[T2]
 val outDS = t1.joinWith(t2, t1("t1_req_id) === t2("t2_req_id), "fullouter")
  .map { case (t1Obj, t2Obj) =>
    val t3:T3 = // do some logic
    t3 
  }
 outDS.toDF.write....

我在 DAG 中看到了投影-但似乎该作业仍然进行完整的数据洗牌另外，在查看执行程序的日志时，我没有看到它在一个块中读取两个表的相同存储桶-这是我所期望的寻找

有spark.sql.sources.bucketing.enabled,spark.sessionState.conf.bucketingEnabled和 spark.sql.join.preferSortMergeJoin标志

我错过了什么？如果有分桶表，为什么仍然有完全洗牌？当前的火花版本是 2.3.1

标签： scalaapache-sparkapache-spark-dataset

Answer 1

这里要检查的一种可能性是您是否有类型不匹配。例如，如果连接列的类型是 T1 中的字符串和 T2 中的 BIGINT。即使类型都是整数（例如一个是 INT，另一个是 BIGINT），Spark 仍然会在此处添加 shuffle，因为不同的类型使用不同的散列函数进行分桶。