hadoop - 为什么我的 BroadcastHashJoin 比 Spark 中的 Shuffle Hash Join 慢

Question

我在 Spark 中使用 javaHiveContext 执行连接。

大表是 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());

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

注意表存储为 Parquet 文件。

Answer 1

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

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