Spark如何广播CSV生成的Dataset以避免Shuffle？

当然可以！针对你提到的多小型查找表场景，广播绝对是避免shuffle、提升关联操作性能的绝佳方案，下面给你详细拆解怎么做以及相关要点：

Spark专门提供了broadcast函数来实现这个需求，你只需要把加载好的小表用它包装一下，Spark就会自动把这个数据集的副本广播到所有Executor节点的内存中，后续和大表做关联时就不会触发shuffle了。示例代码如下：

import org.apache.spark.sql.functions;

// 加载你的小型查找表
Dataset<Row> item = spark.read()
    .option("delimiter", "|")
    .option("header","true")
    .csv(fName)
    .cache();

// 手动广播这个表
Dataset<Row> broadcastedItem = functions.broadcast(item);

之后用broadcastedItem和其他数据集做join操作时，Spark会采用广播哈希Join策略——大表的每个分区直接和本地内存里的小表副本做关联，完全不需要跨节点的数据 shuffle。

除了手动调用函数，你也可以通过配置让Spark自动识别并广播小表：

• 全局配置：在spark-defaults.conf里设置spark.sql.autoBroadcastJoinThreshold参数，单位是字节（默认值是10MB，也就是10485760）。只要你的查找表大小低于这个阈值，Spark就会自动帮你广播它。
• 代码内临时调整：如果需要针对当前作业单独调整阈值，可以在代码里动态设置：

  // 把自动广播阈值设置为20MB
  spark.conf().set("spark.sql.autoBroadcastJoinThreshold", "20971520");
  

不过如果你的表确实很小，手动调用broadcast会更稳妥，能强制Spark采用广播策略，避免因统计信息不准确导致的自动策略失效。

正常的join操作（比如Sort Merge Join）需要把两个表的数据按照关联键重新分区，也就是shuffle——这会产生大量的网络传输和磁盘IO。而广播小表后，每个Executor内存里都有一份小表的完整副本，大表的每个分区只需要在本地和小表做关联，完全不需要跨节点移动数据，自然就避免了shuffle。

• 只给真正小型的表用广播：如果表太大，广播会占用过多Executor内存，反而可能引发OOM或者拖慢整体性能，所以一定要确保你的查找表符合“小表”的定义。
• cache()和广播不冲突：你已经调用了cache()来缓存表，这会避免重复读取源文件；而广播是把缓存后的表副本发送到各个Executor，两者配合使用能进一步提升效率。

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