为什么Spark操作Cassandra关联的简单DataFrame时会进入死循环

• 组件版本兼容bug：你使用的Spark 2.4.5与2.4.3版本的spark-cassandra-connector存在已知的优化逻辑缺陷：当从Cassandra读取的DataFrame执行sort排序操作后再参与union运算时，Spark的Catalyst优化器在尝试将排序规则下推到Cassandra数据源的过程中，会进入无限递归的优化流程，最终表现为程序死循环。
• 你代码中的filter(_=>false)使用了Scala匿名函数做行过滤，不属于Spark可识别的标准谓词逻辑，无法被优化器提前下推到数据源读取阶段，也不会截断排序后的优化链路，所以不影响死循环的触发。你补充的无过滤逻辑的测试用例也直接验证了这一点。

你可以根据自己的场景任选以下任意一种方案规避问题：

1. 切断优化链路：在排序操作后、union操作前调用cache()或persist()，让优化器无法继续向下推导排序规则：

val df = spark.read.cassandraFormat("endless_loop", "test").load()
val df1 = df.sort("id").cache() // 新增cache操作切断优化链路
val df2 = df.filter(_=>false)
val df3 = df1.union(df2)
df3.show()

2. 调整操作顺序：把排序逻辑放到union操作之后执行，避免排序后的DataFrame参与union运算：

val df = spark.read.cassandraFormat("endless_loop", "test").load()
val df2 = df.filter(_=>false)
val df3 = df.union(df2).sort("id")
df3.show()

3. 升级组件版本：该bug在Spark 3.0+以及对应适配的3.0+版本spark-cassandra-connector中已经被修复，直接升级依赖即可从根源解决问题。

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