如何构建一个数据帧,它是两个kafka流的连接,其中“key(s)”列和剩余列是最新的值

我的Spark 2.4.x(pyspark)应用程序需要:

2. 一个“流表”在哪里
   • 其余列应为任一流的最新值。
3. watermarks 用过的。

这似乎是一个基本的东西,但它不完全适合我。

例子:

注意 :在下面的示例中,T1、T2和;T2时间点可以是秒/分/小时间隔。

T1)

获取1个消息负载(让我们调用它 第1页 ):
{ "SecurityCol":"Sec1", "PriceSeqNoCol":"1", "PriceCol": "101.5"}

卡夫卡沃卢梅托皮克 1条带有效负载的消息(我们称之为 第1版
{ "SecurityCol":"Sec1", "VolumeSeqNoCol":"1", "VolumeCol": "50"}

我想喝一杯 结果 DataFrame 看起来像:

+-----------+--------+---------+-------------+--------------+ 
|SecurityCol|PriceCol|VolumeCol|PriceSeqNoCol|VolumeSeqNoCol|  
+-----------+--------+---------+-------------+--------------+ 
|Sec1       |101.5   |50       |1            |1             |
+-----------+--------+---------+-------------+--------------+ 

卡夫卡普罗西汀 1条消息( ):
{ "SecurityCol":"Sec1", "PriceSeqNoCol":"2", "PriceCol": "101.6"}

结果 数据帧

+-----------+--------+---------+-------------+--------------+ 
|SecurityCol|PriceCol|VolumeCol|PriceSeqNoCol|VolumeSeqNoCol|  
+-----------+--------+---------+-------------+--------------+ 
|Sec1       |101.6   |50       |2            |1             |
+-----------+--------+---------+-------------+--------------+ 

: 第1页

T3) 卡夫卡沃卢梅托皮克 1条消息 :
{ "SecurityCol":"Sec1", "VolumeSeqNoCol":"2", "VolumeCol": "60"}

结果 数据帧

+-----------+--------+---------+-------------+--------------+ 
|SecurityCol|PriceCol|VolumeCol|PriceSeqNoCol|VolumeSeqNoCol|
+-----------+--------+---------+-------------+--------------+
|Sec1       |101.6   |60       |2            |2             |
+-----------+--------+---------+-------------+--------------+ 

注意 第1页 & 不再相关了

什么有效

1. get_json_object 现在), join 两个主题的流。
2. watermark )a 数据帧 这是所有的 价格和;接收的卷 第1部分 不仅仅是最新的。
3. 接下来是 groupBy(...).agg(last(...),...) 只是得到一行最新的值。

    dfKafka1 = spark.readStream.format("kafka"). #remaining options etc
                .load()
                .select(...)                     #pulls out fields as columns" 

    dfKafka2 = spark.readStream.format("kafka"). #remaining options etc
               .load()
               .select(...)                     #pulls out fields as columns" 

    dfResult=dfKafka1.join(dfKafka2,"SecurityCol")

#structured streaming doesnt yet allow groupBy after a join, so write to intermediate kafka topic

   dfResult.writestream.format("kafka"). #remaining options
           .trigger(processingTime="1 second")
           .start()

#load intermediate kafka topic
   dfKafkaResult=spark.readStream.format("kafka"). #remaining options
                 .load()
                 .select(...)                      #get_json_object for cols
                 .groupBy("SecurityCol")           #define the "key" to agg cols
                 .agg(last("PriceCol"),            #most recent value per col
                      last("PriceSeqNoCol"),
                      last("VolumeCol"),
                      last("VolumeSeqNoCol"))

问题

agg & last()

1. 当KafkaVolumeTopic收到一条新消息时,结果可能与
2. 进一步 orderBy /排序不能用于没有聚合的流。

限制

1. groupBy 在 参加 因为那需要 withWatermark
   • 应用程序应该能够在一天中的任何时间加入给定SecurityCol的两个主题。
     • 如果PriceTopic在上午9点收到消息,VolumeTopic在上午10点收到消息
   • 水印限制数据何时以 append 模式。所以不能在这里使用水印,因为时间是一整天。

有什么想法吗?