基于 Spark DStream 中的消息时间戳构造窗口

Question

我正在接收来自 Kafka 的 DStream，我想按键将所有消息分组到某个滑动窗口中。

关键是此窗口需要基于每条消息中提供的时间戳(单独的字段(：

Message structure
--------------------------
key1, ..., ..., 1557678233
key1, ..., ..., 1557678234 
key2, ..., ..., 1557678235 

因此，我想考虑每个键timestamp of the first message - timestamp of the last message <= 5 分钟的消息

正如我从这个问题中看到的，这是不可行的，因为 Spark 只计算事件的系统时间。那边的人建议使用updateStateByKey，这对我来说不是很清楚......

也许我们可以使用另一种方法实现这一目标？

如何通过持续时间阈值进一步求和和过滤combineByKey函数的时间戳在combiners中包括时间戳的差异？

请

添加您的想法，或者如果您有机会面对同样的问题，请分享您的解决方案......

谢谢！

Answer 1

可能吗？毫无疑问。Apache Beam，其中提供了Apache Spark后端，可以轻松处理此类操作。

但是，这绝对不是您想要自己实现的东西，除非您拥有大量的开发资源和大量专业知识。如果你有，你可能一开始就不会问这个问题。

处理延迟事件、无序事件以及从节点故障中恢复充其量可能很棘手，因为边缘情况更多。

此外，在您实际实现它之前，它将过时 - DStream已经被视为遗留 API，并且可能迟早达到生命周期结束。同时，结构化流式处理已经可以处理开箱即用的事件时间窗口。

Answer 2

使用以下示例数据进行测试，我假设时间戳采用纪元格式 -

[key1, ..., ..., 1557678233]
[key1, ..., ..., 1557678234]
[key2, ..., ..., 1557678235]
[key2, ..., ..., 1557678240]
[key2, ..., ..., 1557678271]
[key3, ..., ..., 1557678635]
[key3, ..., ..., 1557678636]
[key3, ..., ..., 1557678637]
[key3, ..., ..., 1557678638]
[key3, ..., ..., 1557678999]

/

/-- 创建 udf 以在需要处理或拒绝记录时返回

scala> spark.udf.register("recordStatusUDF", (ts:String) => {
     |     val ts_array = ts.split(",",-1)
     |     if ((ts_array.max.trim.toLong - ts_array.min.trim.toLong) <= 300) {
     |        "process"
     |     }
     |     else { "reject" }
     | })
res83: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

/

/-- 创建架构

scala> val schema = StructType(Seq(StructField("key", StringType, true),StructField("col2", StringType, true),StructField("col3", StringType, true),StructField("epoch_ts", StringType, true)))
schema: org.apache.spark.sql.types.StructType = StructType(StructField(key,StringType,true), StructField(col2,StringType,true), StructField(col3,StringType,true), StructField(epoch_ts,StringType,true))

创建数据帧

scala> spark.createDataFrame(rdd,schema).createOrReplaceTempView("kafka_messages")

scala> spark.sql(s""" select x.key, recordStatusUDF(x.ts) as action_ind from ( select key, concat_ws(",", collect_list(epoch_ts)) as ts from kafka_messages group by key)x """).createOrReplaceTempView("action")
scala> val result = spark.sql(s""" select km.* from kafka_messages km inner join action ac on km.key = ac.key and ac.action_ind = "process" """)
result: org.apache.spark.sql.DataFrame = [key: string, col2: string ... 2 more fields]
scala> result.show(false)
+----+----+----+-----------+
|key |col2|col3|epoch_ts   |
+----+----+----+-----------+
|key1| ...| ...| 1557678233|
|key1| ...| ...| 1557678234|
|key2| ...| ...| 1557678235|
|key2| ...| ...| 1557678240|
|key2| ...| ...| 1557678271|
+----+----+----+-----------+

您可以在每个RDD(Kafka消息(上使用上述代码。希望这是有帮助的。