Spark流-流连接报错：已用等值谓词仍提示不支持无等值连接

报错原因 #

你的等值连接条件t.k = t2.k中，t.k是通过lit(10)生成的常量值，并非流本身的动态列。Spark的流-流（stream-stream）join要求等值连接条件必须是两个流各自的原生/动态生成的属性列之间的匹配，仅用常量与单一流列的等值判断不被视为有效流流join的等值谓词，因此触发报错Stream-stream join without equality predicate is not supported。

解决方法 #

方案1：将常量列改为流动态列 #

把temp流的k列改为基于rate流自身的动态字段生成，比如利用rate流自带的value列生成动态值：

val df = spark.readStream
  .format("rate")
  .option("rowsPerSecond", "1")
  .load()

df.withColumn("k", col("value") % 20) // 用value取模生成动态k值，可根据业务调整规则
  .createOrReplaceTempView("temp")

此时t.k是随流数据变化的动态列，与t2.k的等值连接符合流流join的要求。

方案2：转为流-批join（若业务允许） #

如果业务上不需要temp2作为流处理，可将其改为批读取，这样就不再受流流join的等值条件限制：

// 替换原stream-read为batch-read
spark.read.schema(schema).parquet("core/src/test/resources/hdfs/test")
  .createOrReplaceTempView("temp2")

额外注意事项 #

如果坚持使用流流join，除了合法的等值条件外，建议为两个流添加**水印（Watermark）**配置，避免状态存储无限增长：

// 为temp流添加水印（基于rate流自带的timestamp字段）
val tempDf = df.withColumn("k", col("value") % 20)
  .withWatermark("timestamp", "10 minutes")
tempDf.createOrReplaceTempView("temp")

// 为temp2流添加水印（需确保数据中有timestamp字段，若无则需先通过逻辑生成）
val temp2Df = spark.readStream.schema(schema).parquet("core/src/test/resources/hdfs/test")
  .withWatermark("timestamp", "10 minutes")
temp2Df.createOrReplaceTempView("temp2")

内容的提问来源于stack exchange，提问作者D. belvedere