Spark结构化流-比较两个流

Question

我正在使用Kafka和Spark 2.1结构化流。我有两个json格式的数据主题，例如:

topic 1:

{"id":"1","name":"tom"}
{"id":"2","name":"mark"}

topic 2:

{"name":"tom","age":"25"}
{"name":"mark","age:"35"}

我需要比较Spark中基于标记的两个流：name，当值相等时，执行一些额外的定义/函数。

如何使用Spark结构化流来做到这一点？

谢谢

Answer 1

不久前，我面临一个类似的要求：我有两个流，必须根据一些标准“连接”在一起。我使用的是一个名为mapGroupsWithState的函数。

这个函数的作用（用几句话来说，下面参考的更多细节）是以（K，V）的形式获取流，并根据每对的键将其元素累积在一个共同的状态上。然后，您可以有办法在状态完成时告诉 Spark（根据您的应用程序），甚至可以对不完整的状态进行超时。

基于您的问题的示例：

>

将Kafka主题读入Spark Stream：

val rawDataStream: DataFrame = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", bootstrapServers)
.option("subscribe", "topic1,topic2") // Both topics on same stream!
.option("startingOffsets", "latest")
.option("failOnDataLoss", "true")
.load()
.selectExpr("CAST(value AS STRING) as jsonData") // Kafka sends bytes

对数据执行一些操作（我更喜欢SQL，但您可以使用DataFrame API）将每个元素转换为键值对：

spark.sqlContext.udf.register("getKey", getKey) // You define this function; I'm assuming you will be using the name as key in your example.

val keyPairsStream = rawDataStream
.sql("getKey(jsonData) as ID, jsonData from rawData")
.groupBy($"ID")

使用mapGroupsWithState函数(我给你演示一下基本思路；您必须根据自己的需要定义myGrpFunct):

keyPairsStream
.mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout())(myGrpFunct)

就是这样！如果您正确地实现了myGrpFunct，您将得到一个合并的数据流，您可以进一步转换它，如下所示:

[“汤姆”，“身份证”：“1”，“姓名”：“汤姆”｝，“姓名”：“tom”，“年龄”：“25”｝]

["mark"，{"id"："2"，"name"："mark"}，{"name"："mark"，"age："35"}]

希望这有帮助！

一个很好的解释与一些代码片段：http://asyncified.io/2017/07/30/exploring-stateful-streaming-with-spark-structured-streaming/

Answer 2

我希望你有你的解决方案。如果不是，则可以尝试从两个主题创建两个kstream，然后连接这些kstream并将连接的数据放回一个主题。现在，您可以使用Spark结构化流将连接的数据读取为一个数据帧。现在，您可以对连接的数据应用所需的任何转换。由于结构化流不支持两个流数据帧的连接，因此可以采用这种方法完成任务。

Answer 3

遵循当前文档（Spark 2.1.1）

还不支持两个流数据集之间的任何类型的连接。

裁判：https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#unsupported-操作

此时此刻，我认为你需要依赖于@igofred的回答所提出的Spark Streaming。