KSQL是一個(gè)用于Apache katkatm的流式SQL引擎。KSQL降低了進(jìn)入流處理的門(mén)檻，提供了一個(gè)簡(jiǎn)單的、完全交互式的SQL接口，用于處理Kafka的數(shù)據(jù)。你不再需要用Java或Python這樣的編程語(yǔ)言編寫(xiě)代碼了！KSQL是開(kāi)源的（Apache 2.0許可）、分布式的、可擴(kuò)展的、可靠的和實(shí)時(shí)的。它支持廣泛的強(qiáng)大的流處理操作，包括聚合、連接、窗口、會(huì)話(huà)，等等。

　　一個(gè)簡(jiǎn)單的例子

　　查詢(xún)流數(shù)據(jù)是什么意思，這與SQL數(shù)據(jù)庫(kù)有什么區(qū)別呢？

　　實(shí)際上，它與SQL數(shù)據(jù)庫(kù)有很大的不同。大多數(shù)數(shù)據(jù)庫(kù)都用于對(duì)存儲(chǔ)數(shù)據(jù)進(jìn)行按需查找和修改。KSQL不進(jìn)行查找（但是），它所做的是連續(xù)的轉(zhuǎn)換——也就是，流處理。例如，假設(shè)我有一個(gè)來(lái)自用戶(hù)的點(diǎn)擊流，以及一個(gè)關(guān)于這些用戶(hù)不斷更新的帳戶(hù)信息的表。KSQL允許我對(duì)這一串單擊和用戶(hù)表進(jìn)行建模，并將兩者結(jié)合在一起。即使這兩件事之一是無(wú)限的。

　　因此，KSQL所運(yùn)行的是連續(xù)查詢(xún)——在Kafka主題的數(shù)據(jù)流中，連續(xù)不斷地運(yùn)行新數(shù)據(jù)。相反，傳統(tǒng)數(shù)據(jù)庫(kù)對(duì)關(guān)系數(shù)據(jù)庫(kù)的查詢(xún)是一次性查詢(xún)——在數(shù)據(jù)庫(kù)中運(yùn)行一次SELECT語(yǔ)句獲取有限行的數(shù)據(jù)集。

　　KSQL的好處是什么？

　　很好，所以你可以不斷地查詢(xún)無(wú)限的數(shù)據(jù)流。這有什么好處？

　　1 實(shí)時(shí)監(jiān)控實(shí)時(shí)分析 CREATETABLEerror_counts ASSELECTerror_code，count（*）FROMmonitoring_stream WINDOW TUMBLING （SIZE1MINUTE） WHEREtype =‘ERROR’

　　其中的一個(gè)用途是定義定制的業(yè)務(wù)級(jí)度量，這些度量是實(shí)時(shí)計(jì)算的，您可以監(jiān)視和警報(bào)，就像您的CPU負(fù)載一樣。另一個(gè)用途是在KSQL中定義應(yīng)用程序的正確性的概念，并檢查它在生產(chǎn)過(guò)程中是否會(huì)遇到這個(gè)問(wèn)題。通常，當(dāng)我們想到監(jiān)控時(shí)，我們會(huì)想到計(jì)數(shù)器和儀表跟蹤低水平的性能統(tǒng)計(jì)。這些類(lèi)型的測(cè)量器通?？梢愿嬖V你CPU負(fù)載很高，但是它們不能真正告訴你你的應(yīng)用程序是否在做它應(yīng)該做的事情。KSQL允許從應(yīng)用程序生成的原始事件流中定義定制指標(biāo)，無(wú)論它們是日志事件、數(shù)據(jù)庫(kù)更新還是其他類(lèi)型的事件。

　　例如，一個(gè)web應(yīng)用程序可能需要檢查，每次新客戶(hù)注冊(cè)一個(gè)受歡迎的電子郵件，創(chuàng)建一個(gè)新的用戶(hù)記錄，并且他們的信用卡被計(jì)費(fèi)。這些功能可能分布在不同的服務(wù)或應(yīng)用程序中，您可能希望監(jiān)視每個(gè)新客戶(hù)在SLA中發(fā)生的每一件事，比如30秒。

　　2 安全性和異常檢測(cè) CREATESTREAM possible_fraud ASSELECTcard_number， count （*） FROM authorization_attempts WINDOW TUMBLING （SIZE 5 SECONDS） GROUP BY card_number HAVING count（*）》 3;

　　這是您在上面的演示中看到的一個(gè)簡(jiǎn)單的版本：KSQL查詢(xún)，它將事件流轉(zhuǎn)換為數(shù)值時(shí)間序列，使用Kafka-Elastic連接器將其注入到彈性中，并在Grafana UI中可視化。安全用例通常看起來(lái)很像監(jiān)視和分析。而不是監(jiān)視應(yīng)用程序的行為或業(yè)務(wù)行為，您正在尋找欺詐、濫用、垃圾郵件、入侵或其他不良行為的模式。KSQL提供了一種簡(jiǎn)單、復(fù)雜和實(shí)時(shí)的方式來(lái)定義這些模式和查詢(xún)實(shí)時(shí)流。

　　3 在線(xiàn)數(shù)據(jù)集成 CREATESTREAM vip_users ASSELECTuserid， page，actionFROMclickstream c LEFTJOINusers u ONc.userid = u.user_id WHEREu.level=‘Platinum’;

　　在公司中完成的大部分?jǐn)?shù)據(jù)處理都屬于數(shù)據(jù)豐富的領(lǐng)域：從幾個(gè)數(shù)據(jù)庫(kù)中提取數(shù)據(jù)，轉(zhuǎn)換它，將其連接到一個(gè)鍵值存儲(chǔ)、搜索索引、緩存或其他數(shù)據(jù)服務(wù)系統(tǒng)中。在很長(zhǎng)一段時(shí)間內(nèi)，用于數(shù)據(jù)集成的ETL-提取、轉(zhuǎn)換和加載-作為周期性的批處理作業(yè)執(zhí)行。例如，實(shí)時(shí)轉(zhuǎn)儲(chǔ)原始數(shù)據(jù)，然后每隔幾個(gè)小時(shí)轉(zhuǎn)換一次，以實(shí)現(xiàn)高效的查詢(xún)。對(duì)于許多用例來(lái)說(shuō)，這種延遲是不可接受的。KSQL與Kafka的連接器一起使用時(shí)，可以從批處理數(shù)據(jù)集成到在線(xiàn)數(shù)據(jù)集成。您可以使用流-表連接存儲(chǔ)在表中的元數(shù)據(jù)來(lái)豐富數(shù)據(jù)流，或者在將流加載到另一個(gè)系統(tǒng)之前對(duì)PII（個(gè)人可識(shí)別的信息）進(jìn)行簡(jiǎn)單的過(guò)濾。

　　4 應(yīng)用程序開(kāi)發(fā)

　　許多應(yīng)用程序?qū)⑤斎肓鬓D(zhuǎn)換為輸出流。 例如，負(fù)責(zé)重新排序在線(xiàn)商店庫(kù)存不足的產(chǎn)品的流程可能會(huì)產(chǎn)生銷(xiāo)售和出貨流，以計(jì)算出訂單流。

　　對(duì)于用Java編寫(xiě)的更復(fù)雜的應(yīng)用程序來(lái)說(shuō)，Kafka的原生流API可能幫助不大。但是對(duì)于簡(jiǎn)單的應(yīng)用程序，或者對(duì)Java編程不感興趣的團(tuán)隊(duì)來(lái)說(shuō)，一個(gè)簡(jiǎn)單的SQL接口可能就是他們想要的。

　　KSQL中的核心抽象

　　KSQL在內(nèi)部使用Kafka的Streams API，并且它們共享與Kafka流處理相同的核心抽象。 KSQL有兩個(gè)核心抽象，它們映射到Kafka Streams中的兩個(gè)核心抽象，并允許您操縱Kafka主題：

　　1.流：流是無(wú)限制的結(jié)構(gòu)化數(shù)據(jù)序列（“事實(shí)”）。 例如，我們可以有一個(gè)金融交易流，例如“Alice向Bob發(fā)送了100美元，然后查理向鮑勃發(fā)送了50美元”。 流中的事實(shí)是不可變的，這意味著可以將新事實(shí)插入到流中，但是現(xiàn)有事實(shí)永遠(yuǎn)不會(huì)被更新或刪除。 流可以從Kafka主題創(chuàng)建，或者從現(xiàn)有的流和表中派生。

　　CREATESTREAM pageviews （viewtime BIGINT， userid VARCHAR， pageid VARCHAR）WITH（kafka_topic=‘pageviews’， value_format=’JSON’）;

　　2。表：一個(gè)表是一個(gè)流或另一個(gè)表的視圖，它代表了一個(gè)不斷變化的事實(shí)的集合。例如，我們可以擁有一個(gè)包含最新財(cái)務(wù)信息的表，例如“Bob的經(jīng)常帳戶(hù)余額為$150”。它相當(dāng)于傳統(tǒng)的數(shù)據(jù)庫(kù)表，但通過(guò)流化等流語(yǔ)義來(lái)豐富。表中的事實(shí)是可變的，這意味著可以將新的事實(shí)插入到表中，現(xiàn)有的事實(shí)可以被更新或刪除?？梢詮腒afka主題中創(chuàng)建表，也可以從現(xiàn)有的流和表中派生表。

　　CREATETABLEusers （registertime BIGINT， gender VARCHAR， regionid VARCHAR， userid VARCHAR） WITH（kafka_topic=‘users’， value_format=‘DELIMITED’）;

　　KSQL簡(jiǎn)化了流應(yīng)用程序，因?yàn)樗耆闪吮砗土鞯母拍?，允許使用表示現(xiàn)在發(fā)生的事件的流來(lái)連接表示當(dāng)前狀態(tài)的表。 Apache Kafka中的一個(gè)主題可以表示為KSQL中的STREAM或TABLE，具體取決于主題處理的預(yù)期語(yǔ)義。 例如，如果要將主題中的數(shù)據(jù)作為一系列獨(dú)立值讀取，則可以使用CREATE STREAM。此類(lèi)流的一個(gè)例子是捕獲頁(yè)面視圖事件，其中每個(gè)頁(yè)面視圖事件都不相關(guān)且獨(dú)立于另一個(gè)頁(yè)面視圖事件。另一方面，如果您希望將某個(gè)主題中的數(shù)據(jù)讀取為可更新的值的集合，那么您將使用CREATE TABLE。在KSQL中應(yīng)該讀取一個(gè)主題的示例，它捕獲用戶(hù)元數(shù)據(jù)，其中每個(gè)事件代表特定用戶(hù)id的最新元數(shù)據(jù)，如用戶(hù)的姓名、地址或首選項(xiàng)。

　　KSQL：實(shí)時(shí)點(diǎn)擊流分析和異常檢測(cè)

　　讓我們來(lái)看一個(gè)真正的例子。這個(gè)例子展示如何使用KSQL進(jìn)行實(shí)時(shí)監(jiān)視、異常檢測(cè)和警報(bào)。對(duì)clickstream數(shù)據(jù)的實(shí)時(shí)日志分析可以采取多種形式。在本例中，我們將標(biāo)記在web服務(wù)器上消耗過(guò)多帶寬的惡意用戶(hù)會(huì)話(huà)。監(jiān)視惡意用戶(hù)會(huì)話(huà)是會(huì)話(huà)化的眾多應(yīng)用之一。但從廣義上說(shuō)，會(huì)話(huà)是用戶(hù)行為分析的基礎(chǔ)。一旦您將用戶(hù)和事件關(guān)聯(lián)到一個(gè)特定的會(huì)話(huà)標(biāo)識(shí)符，您就可以構(gòu)建許多類(lèi)型的分析，從簡(jiǎn)單的度量，例如訪(fǎng)問(wèn)計(jì)數(shù)。我們通過(guò)展示如何在Elastic支持的Grafana儀表板上實(shí)時(shí)顯示KSQL查詢(xún)的輸出，來(lái)結(jié)束這個(gè)例子。

　　您也可以按照我們的指示，親自完成例子，并查看代碼。