KSQL是一個用于Apache katkatm的流式SQL引擎。KSQL降低了進入流處理的門檻，提供了一個簡單的、完全交互式的SQL接口，用于處理Kafka的數據。你不再需要用Java或Python這樣的編程語言編寫代碼了！KSQL是開源的(Apache 2.0許可)、分布式的、可擴展的、可靠的和實時的。它支持廣泛的強大的流處理操作，包括聚合、連接、窗口、會話，等等。

一個簡單的例子

查詢流數據是什么意思，這與SQL數據庫有什么區別呢?

實際上，它與SQL數據庫有很大的不同。大多數數據庫都用于對存儲數據進行按需查找和修改。KSQL不進行查找(但是)，它所做的是連續的轉換——也就是，流處理。例如，假設我有一個來自用戶的點擊流，以及一個關于這些用戶不斷更新的帳戶信息的表。KSQL允許我對這一串單擊和用戶表進行建模，并將兩者結合在一起。即使這兩件事之一是無限的。 
因此，KSQL所運行的是連續查詢——在Kafka主題的數據流中，連續不斷地運行新數據。相反，傳統數據庫對關系數據庫的查詢是一次性查詢——在數據庫中運行一次SELECT語句獲取有限行的數據集。

KSQL的好處是什么?

很好，所以你可以不斷地查詢無限的數據流。這有什么好處?

1 實時監控實時分析

CREATE TABLE error_counts AS SELECT error_code, count(*)FROM monitoring_stream
WINDOW TUMBLING (SIZE 1 MINUTE) WHERE type = 'ERROR'

其中的一個用途是定義定制的業務級度量，這些度量是實時計算的，您可以監視和警報，就像您的CPU負載一樣。另一個用途是在KSQL中定義應用程序的正確性的概念，并檢查它在生產過程中是否會遇到這個問題。通常，當我們想到監控時，我們會想到計數器和儀表跟蹤低水平的性能統計。這些類型的測量器通?？梢愿嬖V你CPU負載很高，但是它們不能真正告訴你你的應用程序是否在做它應該做的事情。KSQL允許從應用程序生成的原始事件流中定義定制指標，無論它們是日志事件、數據庫更新還是其他類型的事件。

例如，一個web應用程序可能需要檢查，每次新客戶注冊一個受歡迎的電子郵件，創建一個新的用戶記錄，并且他們的信用卡被計費。這些功能可能分布在不同的服務或應用程序中，您可能希望監視每個新客戶在SLA中發生的每一件事，比如30秒。

2 安全性和異常檢測

CREATE STREAM possible_fraud AS SELECT card_number, count(*)
FROM authorization_attempts
WINDOW TUMBLING (SIZE 5 SECONDS)
GROUP BY card_number
HAVING count(*) > 3;

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

3 在線數據集成

CREATE STREAM vip_users AS SELECT userid, page, action FROM clickstream c LEFT JOIN users u ON c.userid = u.user_id WHERE u.level = 'Platinum';

在公司中完成的大部分數據處理都屬于數據豐富的領域:從幾個數據庫中提取數據，轉換它，將其連接到一個鍵值存儲、搜索索引、緩存或其他數據服務系統中。在很長一段時間內，用于數據集成的ETL-提取、轉換和加載-作為周期性的批處理作業執行。例如，實時轉儲原始數據，然后每隔幾個小時轉換一次，以實現高效的查詢。對于許多用例來說，這種延遲是不可接受的。KSQL與Kafka的連接器一起使用時，可以從批處理數據集成到在線數據集成。您可以使用流-表連接存儲在表中的元數據來豐富數據流，或者在將流加載到另一個系統之前對PII(個人可識別的信息)進行簡單的過濾。

4 應用程序開發

許多應用程序將輸入流轉換為輸出流。 例如，負責重新排序在線商店庫存不足的產品的流程可能會產生銷售和出貨流，以計算出訂單流。

對于用Java編寫的更復雜的應用程序來說，Kafka的原生流API可能幫助不大。但是對于簡單的應用程序，或者對Java編程不感興趣的團隊來說，一個簡單的SQL接口可能就是他們想要的。

KSQL中的核心抽象

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

1.流：流是無限制的結構化數據序列（“事實”）。 例如，我們可以有一個金融交易流，例如“Alice向Bob發送了100美元，然后查理向鮑勃發送了50美元”。 流中的事實是不可變的，這意味著可以將新事實插入到流中，但是現有事實永遠不會被更新或刪除。 流可以從Kafka主題創建，或者從現有的流和表中派生。

CREATE STREAM pageviews (viewtime BIGINT, userid VARCHAR, pageid VARCHAR) WITH (kafka_topic='pageviews', value_format=’JSON’);

2。表:一個表是一個流或另一個表的視圖，它代表了一個不斷變化的事實的集合。例如，我們可以擁有一個包含新財務信息的表，例如“Bob的經常帳戶余額為$150”。它相當于傳統的數據庫表，但通過流化等流語義來豐富。表中的事實是可變的，這意味著可以將新的事實插入到表中，現有的事實可以被更新或刪除?？梢詮腒afka主題中創建表，也可以從現有的流和表中派生表。

CREATE TABLE users (registertime BIGINT, gender VARCHAR, regionid VARCHAR, userid VARCHAR) WITH (kafka_topic='users', value_format='DELIMITED');

KSQL簡化了流應用程序，因為它完全集成了表和流的概念，允許使用表示現在發生的事件的流來連接表示當前狀態的表。 Apache Kafka中的一個主題可以表示為KSQL中的STREAM或TABLE，具體取決于主題處理的預期語義。 例如，如果要將主題中的數據作為一系列獨立值讀取，則可以使用CREATE STREAM。此類流的一個例子是捕獲頁面視圖事件，其中每個頁面視圖事件都不相關且獨立于另一個頁面視圖事件。另一方面，如果您希望將某個主題中的數據讀取為可更新的值的集合，那么您將使用CREATE TABLE。在KSQL中應該讀取一個主題的示例，它捕獲用戶元數據，其中每個事件代表特定用戶id的新元數據，如用戶的姓名、地址或首選項。

KSQL:實時點擊流分析和異常檢測

讓我們來看一個真正的例子。這個例子展示如何使用KSQL進行實時監視、異常檢測和警報。對clickstream數據的實時日志分析可以采取多種形式。在本例中，我們將標記在web服務器上消耗過多帶寬的惡意用戶會話。監視惡意用戶會話是會話化的眾多應用之一。但從廣義上說，會話是用戶行為分析的基礎。一旦您將用戶和事件關聯到一個特定的會話標識符，您就可以構建許多類型的分析，從簡單的度量，例如訪問計數。我們通過展示如何在Elastic支持的Grafana儀表板上實時顯示KSQL查詢的輸出，來結束這個例子。

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

看看里面

有一個KSQL服務器進程執行查詢。一組KSQL進程作為集群運行。您可以通過啟動更多的KSQL server實例來動態添加更多的處理能力。這些實例是容錯的:如果一個失敗了，其他的就會接管它的工作。查詢是使用交互式的KSQL命令行客戶端啟動的，該客戶端通過REST API向集群發送命令。命令行允許檢查可用的流和表，發出新的查詢，檢查狀態并終止正在運行的查詢。KSQL內部是使用Kafka的流API構建的;它繼承了它的彈性可伸縮性、先進的狀態管理和容錯功能，并支持Kafka近引入的一次性處理語義。KSQL服務器將此嵌入到一個分布式SQL引擎中(包括一些用于查詢性能的自動字節代碼生成)和一個用于查詢和控制的REST API。

Kafka + KSQL將數據庫轉出來

過去我們已經討論過將數據庫轉入內部，現在我們通過向內向外的DB添加一個SQL層來實現。

在關系數據庫中，表是核心抽象，日志是一個實現細節。 在以數據庫為中心的事件世界中，核心抽象不是表; 它是日志。 這些表只是從日志導出的，并隨著新數據到達日志而不斷更新。 中央日志是Kafka，KSQL是引擎，允許您創建所需的物化視圖，并將其表示為不斷更新的表。 
然后，您可以以這種流式表格方式運行即時查詢（即將在KSQL中），以便以持續的方式獲取日志中每個鍵的新值。

使用Kafka和KSQL將數據庫轉出，對一家公司的所有數據都有很大的影響，這些數據可以自然地以流媒體方式進行表示和處理。Kafka日志是流數據的核心存儲抽象，允許進入您的離線數據倉庫的相同數據現在可以用于流處理。其他一切都是在日志上的一個流化的物化視圖，它是各種數據庫、搜索索引，或者是公司的其他數據服務系統。創建這些派生視圖所需的所有數據和ETL，現在都可以使用KSQL以流媒體方式完成。監控、安全、異常和威脅檢測、分析和對故障的響應都可以實時進行，而當時間太晚了。所有這些都可以通過一個簡單而又熟悉的SQL接口來使用所有Kafka的數據:KSQL。