引言

展示了如何集成Confluent Kafka，KSQL，Spring Boot和YugaByte DB來開發用于管理物聯網(IoT)傳感器數據的應用程序。

場景 - 支持物聯網的車隊管理

一家貨運公司希望跟蹤其在全國范圍內運送貨物的物聯網車輛。車輛屬于不同類型(例如18輪車，公共汽車，大型卡車)，并遵循3條交付路線(Route-37，Route-82，Route-43)。特別是，該公司希望跟蹤：

• 每條交付路線的車輛類型的總體分布。
• 最近(例如，在過去30秒內)每個貨物交付路線的這些車輛類型的子集。
• 道路封閉附近的車輛清單，以便他們可以預測交貨延誤。

應用架構

除了Confluent Kafka作為流媒體平臺，該應用程序還具有以下組件：

• 數據存儲： YugaByte DB用于存儲來自Kafka流的原始事件以及來自KSQL數據處理器的聚合。
• 數據生成器：用于模擬寫入Kafka流的車輛事件的程序。
• 數據處理器： 從Data Producer讀取KSQL，計算聚合并將結果存儲在Data Store中。
• Data Dashboard： Spring Boot應用程序，使用Web套接字，jQuery和Bootstrap顯示數據處理器的輸出。

下面是顯示這些組件如何組合在一起的架構圖。我們稱之為Confluent Kafka，KSQL和YugaByte DB堆棧或CKY堆棧。

我們現在將詳細介紹這些組件中的每一個。

數據存儲

該層存儲所有用戶數據。YugaByte DB用作數據庫，YugaByte云查詢語言(YCQL)用作數據庫API。所有數據都存儲在密鑰空間TrafficKeySpace中。有一個Origin_Table用于存儲原始事件的表。

CREATE TABLE TrafficKeySpace.Origin_Table ( vehicleId text, routeId text, vehicleType text, longitude text, latitude text, timeStamp timestamp, speed double, fuelLevel double, PRIMARY KEY ((vehicleId), timeStamp)) WITH default_time_to_live = 3600;

請注意default_time_to_live設置為3600秒的值，以確保原始事件在1小時后自動刪除。這是為了確保原始事件不會消耗數據庫中的所有存儲，并且在計算聚合后不久就會有效地從數據庫中刪除。

有三個表用于保存用于面向用戶的顯示的數據：

• Total_Traffic 交通信息
• Window_Traffic 最后30秒的流量和
• poi_traffic 對于興趣點附近的交通(道路封閉)。

數據處理器不斷更新這些表，儀表板從中讀取。

以下是這些表:

CREATE TABLE TrafficKeySpace.Total_Traffic ( routeId text, vehicleType text, totalCount bigint, timeStamp timestamp, recordDate text, PRIMARY KEY (routeId, recordDate, vehicleType));CREATE TABLE TrafficKeySpace.Window_Traffic ( routeId text, vehicleType text, totalCount bigint, timeStamp timestamp, recordDate text, PRIMARY KEY (routeId, recordDate, vehicleType));CREATE TABLE TrafficKeySpace.Poi_Traffic( vehicleid text, vehicletype text, distance bigint, timeStamp timestamp, PRIMARY KEY (vehicleid));

數據生產者

這包含生成模擬測試數據并將其發布到Kafka主題iot-data-event的程序。這模擬了使用現實世界中的消息代理從連接的車輛接收的數據。

單個數據點是JSON有效負載，如下所示：

{ "vehicleId":"0bf45cac-d1b8-4364-a906-980e1c2bdbcb

總結

以上是生活随笔為你收集整理的springboot 物联网_Confluent Kafka，KSQL，Spring Boot和分布式SQL开发物联网实战的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。