KSQL - 使用 GEO_DISTANCE 计算 2 条消息的距离

Question

我有一个 kafka 主题，主题中的每条消息都有纬度/经度和事件时间戳。创建了一个引用主题的流，并希望使用 geo_distance 计算 2 点之间的距离。 例子

GpsDateTime            lat              lon
2016-11-30 22:38:36,    32.685757,  -96.735942
2016-11-30 22:39:07,    32.687347,  -96.732841
2016-11-30 22:39:37,    32.68805,   -96.729726 

我想在上面的流上创建一个新流并用距离丰富它。

GpsDateTime            lat              lon          Distance
2016-11-30 22:38:36,    32.685757,  -96.735942        0
2016-11-30 22:39:07,    32.687347,  -96.732841        0.340
2016-11-30 22:39:37,    32.68805,   -96.729726        0.302

是否可以使用 KSQL 达到预期的效果？或者如何在处理新消息时参考之前的消息？

Answer 1

首先，这些读数是否来自某种设备？如果是这样，您是否有他们的唯一 ID (UUID)？我会把它放到你的流中，所以它就像UUID, GpsDateTime, lat, lon。

您将需要创建一个相当基本的 Kafka Streams 应用程序。在这个应用程序中，您将从流中的最新读数存储到 StoreBuilder 中。然后，当收到来自 Kafka 的新消息时，您将检索此最新值，进行计算，然后将新的 lat,long 值存储到 StoreBuilder。

当然，我不清楚您是否只想永远有一个纬度、经度值，而您所有的后续值都是从第一次读数中计算出来的。或者，如果您想进行滚动计算，始终比较上次读数和当前读数之间的距离。

不管怎样，你可以在实践中看到这段代码：https://github.com/confluentinc/kafka-streams-examples/blob/5.0.0-post/src/test/java/io/confluent/examples/streams/StateStoresInTheDSLIntegrationTest.java

此示例是一个字数统计示例，但可以根据您的用例快速转换。

静态最终类 WordCountTransformerSupplier（第 78 行）将成为您的 LatLongDistanceComputation。

您将创建具有适当类型的 StoreBuilder（第 154 行）（无论您将纬度/经度存储为什么）。

第 165 行是实际从流入的值流中读取项目的位置。

当然，您还需要编辑 inputTopic 和 outputTopic（第 66-67 行）以及其他一些内容。