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

EN

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

您将需要创建一个相当基本的Kafka流应用程序。在这个应用程序中，您将把最近从流中读取的内容存储到StoreBuilder中。然后，当从Kafka收到新消息时，您将检索这个最新值，进行计算，然后将新的lat长值存储到StoreBuilder中。

当然，我不清楚你是否只想拥有一个lat，长值，所有后续的值都是从一读开始计算的。或者，如果您希望有一个滚动计算，您总是比较之间的距离，最后和当前的读数。

无论如何，您可以在以下网站上看到这段代码：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行)(无论您以何种方式存储lat/lon )。

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

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