为什么我的BroadcastHashJoin比Spark中的ShuffledHashJoin慢

Question

我javaHiveContext在Spark中执行联接。

大表是1,76Gb，拥有1亿条记录。

第二个表为273Mb，记录为1000万。

我得到一个，JavaSchemaRDD并呼吁count()：

String query="select attribute7,count(*) from ft,dt where ft.chiavedt=dt.chiavedt group by attribute7";

JavaSchemaRDD rdd=sqlContext.sql(query);

System.out.println("count="+rdd.count());

如果我强制abroadcastHashJoin (SET spark.sql.autoBroadcastJoinThreshold=290000000)并在具有8个核心和20Gb内存的5个节点上使用5个执行程序，则它将在100秒内执行。如果我不强制播放，它将在30秒内执行。

注意，表存储为Parquet文件。

Answer 1

问题的根源很可能是广播成本。为了简单起见，假设您在较大的RDD中有1800MB，在较小的RDD中有300MB。假设有5个执行者并且没有以前的分区，则所有数据的五分之一应该已经在正确的机器上。在标准连接的情况下，它剩下约1700MB的空间用于改组。

为了进行广播连接，必须将较小的RDD传输到所有节点。这意味着大约有1500MB的数据要传输。如果添加了与驱动程序的必需通信，则意味着必须以昂贵得多的方式移动相当数量的数据。必须首先收集广播的数据，然后才能将其转发给所有工作人员。