用于傳感器分析的KSQL UDF。利用KSQL的新的API特性，用Java輕松地構建UDF / UDAF函數(shù)，從而使用Apache Kafka進行連續(xù)流處理。用例：聯(lián)網(wǎng)汽車——使用深度學習的實時流媒體分析。

我為混合機器學習基礎設施構建了一個場景，利用Apache Kafka作為可伸縮的中樞神經(jīng)系統(tǒng)。使用公共云在極端尺度下訓練分析模型（如通過谷歌ML引擎在谷歌云平臺（GCP）上使用TensorFlow和TPUs。預測（即模型推斷）是在本地Kafka基礎設施的邊緣前提下執(zhí)行的（例如利用Kafka流或KSQL進行流分析）。

這篇文章的重點是在前提部署。我用KSQL UDF創(chuàng)建了一個用于傳感器分析的Github項目。它利用KSQL的新API特性輕松地使用Java構建UDF / UDAF函數(shù)，對傳入事件進行連續(xù)流處理。

用例：聯(lián)網(wǎng)汽車——使用深度學習的實時流媒體分析

連續(xù)處理來自連接設備（本例中的汽車傳感器）的數(shù)百萬個事件：

我建立了不同的分析模型。他們在公共云上接受訓練，利用TensorFlow、H2O和谷歌ML引擎。模型創(chuàng)建不是這個示例的重點。最終的模型已經(jīng)準備好投入生產(chǎn)，并可以部署進行實時預測。

模型服務可以通過模型服務器或原生嵌入到流處理應用程序中來完成。查看模型部署中RPC與流處理的權衡和“TensorFlow + gRPC + Kafka流”示例。

演示：使用MQTT、Kafka和KSQL在邊緣進行模型推斷

Github項目生成汽車傳感器數(shù)據(jù)，通過Confluent MQTT代理將其轉發(fā)到Kafka集群進行KSQL處理和實時分析。

這個項目主要是通過MQTT將數(shù)據(jù)輸入Kafka，通過KSQL對數(shù)據(jù)進行處理：

Confluent MQTT代理的一大優(yōu)點是可以簡單地實現(xiàn)物聯(lián)網(wǎng)場景，而不需要MQTT代理。您可以通過MQTT代理直接將消息從MQTT設備轉發(fā)到Kafka。這大大減少了工作和成本。如果您“只是”希望在Kafka和MQTT設備之間進行通信，那么這是一個完美的解決方案。

如果你想看這個故事的其他部分（與像Elasticsearch / Grafana這樣的sink應用的集成），請看看Github項目“KSQL流物聯(lián)網(wǎng)數(shù)據(jù)”。通過Kafka Connect和Elastic connector實現(xiàn)了與ElasticSearch和Grafana的集成。

KSQL UDF 源代碼

開發(fā)udf非常容易。只需在一個UDF類中實現(xiàn)一個Java方法：

下面是KSQL UDF異常檢測的完整源代碼。（Anomaly Detection KSQL UDF.）

如何運行與Apache Kafka和MQTT代理演示？

在Github項目中描述了執(zhí)行演示的所有步驟。

您只需要安裝Confluent Platform，然后按照以下步驟部署UDF、創(chuàng)建MQTT事件并通過利用分析模型的KSQL處理它們。

我使用mosquito to生成MQTT消息。當然，您也可以使用任何其他MQTT客戶機。這就是開放和標準化協(xié)議的最大好處。
責任編輯:pj