Spark 2.2.0 是不是支持 Streaming Self-Join？

Posted 2023-04-15

【中文标题】Spark 2.2.0 是不是支持 Streaming Self-Join？

【英文标题】：Does Spark 2.2.0 support Streaming Self-Joins?Spark 2.2.0 是否支持 Streaming Self-Join？

【发布时间】：2018-02-18 13:13:11

【问题描述】：

我了解 Spark 2.2.0 不支持两个不同数据帧的 JOINS，但我正在尝试进行自连接，因此只有一个流。下面是我的代码

val jdf = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "join_test")
    .option("startingOffsets", "earliest")
    .load();

jdf.printSchema

打印以下内容

root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)

现在我在阅读完这个 SO post 后运行下面的连接查询

jdf.as("jdf1").join(jdf.as("jdf2"), $"jdf1.key" === $"jdf2.key")

我得到以下异常

org.apache.spark.sql.AnalysisException: cannot resolve '`jdf1.key`' given input columns: [timestamp, value, partition, timestampType, topic, offset, key];;
'Join Inner, ('jdf1.key = 'jdf2.key)
:- SubqueryAlias jdf1
:  +- StreamingRelation DataSource(org.apache.spark.sql.SparkSession@f662b5,kafka,List(),None,List(),None,Map(startingOffsets -> earliest, subscribe -> join_test, kafka.bootstrap.servers -> localhost:9092),None), kafka, [key#243, value#244, topic#245, partition#246, offset#247L, timestamp#248, timestampType#249]
+- SubqueryAlias jdf2
   +- StreamingRelation DataSource(org.apache.spark.sql.SparkSession@f662b5,kafka,List(),None,List(),None,Map(startingOffsets -> earliest, subscribe -> join_test, kafka.bootstrap.servers -> localhost:9092),None), kafka, [key#243, value#244, topic#245, partition#246, offset#247L, timestamp#248, timestampType#249]

【参考方案1】：

我认为如果我们尝试加入相同的流数据帧或不同的数据帧，它不会产生任何差异。所以，它不会被支持。

有两种方法可以实现。

首先，您可以加入静态和流式数据帧。因此，读取一次作为批处理数据，然后读取一次作为流式 df。 第二种解决方案，您可以使用 Kafka 流。它提供了流数据的连接。

【讨论】：

您可能是对的，但错误说明了其他内容。它说无法解析“jdf1.key”。不是说不支持JOINS！

它可能对你有帮助，***.com/questions/46412884/…

知道了！但我很想看到 UnsupportedOperationException ......相反，我在上面的代码中看到了其他内容。