Spark Structured Streaming CSV流与rate流join加persist触发执行异常求解

你遇到的问题本质是 Spark Structured Streaming 不支持直接对流式DataFrame/DataSet调用persist()/cache()方法，以下是对应你的三个疑问的解答：

1. 关于异常提示的疑问
   persist是Spark批处理专属API，调用时会触发Spark以批处理模式校验执行计划，而你的输入是readStream生成的流数据源，批处理执行路径天然不支持流源，所以才抛出“带流源的查询必须调用writeStream.start()执行”的错误。这个错误不是要求你在流DataFrame生成后直接调用start，而是告诉你当前的执行路径（批处理的缓存逻辑）不兼容流源，必须走完整的流查询链路：readStream -> 转换逻辑 -> writeStream -> start，中间不能插入批处理专属操作。
2. 关于无法打印执行计划的疑问
   异常是在你调用persist的瞬间触发的，此时你还没有构建完完整的流查询逻辑，后续的join、输出逻辑还没有和CSV流关联起来，Spark在校验缓存操作的时候就直接报错终止了，自然拿不到完整的物理执行计划。
3. 关于去掉persist能跑但性能差的疑问
   去掉persist后，Spark不会触发提前的批模式校验，可以正常构建完整的流查询链路所以能运行。性能差的原因是你配置了maxFilesPerTrigger=1，每个微批次都会重新读取CSV文件、解析格式，没有缓存的前提下每次都要重复执行HDFS读取、CSV解析的逻辑，开销会非常高。

可行解决方案 #

如果你用到的CSV数据是静态不变的维度数据，最合理的方案就是你测试过的批处理读取方式：用read加载CSV数据后调用persist缓存，再和rate流做流静态Join，这是Spark官方原生支持的模式，静态数据只会加载一次，性能最优。
如果你的CSV数据需要动态增量更新，必须用流模式读取，目前Structured Streaming暂不支持流数据集的缓存操作，可以通过以下方式优化性能：

• 调大maxFilesPerTrigger的数值，减少微批次的触发频率和文件读取次数
• 提前将CSV文件转换为Parquet、ORC等列存格式，降低文件解析开销
• 增加HDFS的短路读配置，或者使用分布式缓存层加速文件读取效率

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