KSQL介紹：面向Apache Kafka的開源Streaming SQL引擎

時間 2019-11-11

標籤 ksql 介紹面向 apache kafka 開源 streaming sql 引擎欄目 Apache 简体版

原文原文鏈接

我很是高興地宣佈KSQL，這是面向Apache Kafka的一種數據流SQL引擎。KSQL下降了數據流處理這個領域的准入門檻，爲使用Kafka處理數據提供了一種簡單的、徹底交互的SQL界面。你再也不須要用Java或Python之類的編程語言編寫代碼了！KSQL具備這些特色：開源（採用Apache 2.0許可證）、分佈式、可擴展、可靠、實時。它支持衆多功能強大的數據流處理操做，包括聚合、鏈接、加窗（windowing）和sessionization（捕獲單一訪問者的網站會話時間範圍內全部的點擊流事件）等等。php

文章目錄html

一個簡單的例子

若是想及時瞭解Spark、Hadoop或者Hbase相關的文章，歡迎關注微信公共賬號：iteblog_hadoopsql

查詢流式數據意味着什麼？這與SQL數據庫相比怎樣？

能夠說它實際上與SQL數據庫大不同。大多數數據庫用於對存儲的數據執行按需查找和改動。KSQL還沒法執行查詢，它所作的是連續轉換――也就是說，數據流處理。好比設想一下：我有來自用戶的點擊流和賬戶信息表（這些信息關於不斷更新的那些用戶）。KSQL讓我能夠對該點擊流和用戶表進行建模，並將二者鏈接起來，儘管那二者當中的一個是無限的。數據庫

因此，KSQL對Kafka話題中的數據流執行連續查詢――隨着新數據不斷流入，轉換在連續進行。相比之下，針對關係數據庫的查詢是一次性查詢――對數據集運行一次便可完成，就像針對數據庫中有限行的SELECT語句。apache

KSQL適用於什麼？

很好，如今你能夠連續查詢無限數據流。那有什麼好處呢？編程

一、實時監控趕上實時分析

CREATE TABLE error_counts AS緩存

SELECT error_code, count(*)FROM monitoring_stream安全

WINDOW TUMBLING (SIZE 1 MINUTE)服務器

WHERE type = 'ERROR'微信

這方面的一個用途是，定義實時計算的自定義業務級別度量指標，你能夠監控併發出警報，就像監控CPU負載那樣。另外一個用途是，在KSQL中爲應用程序定義正確性概念，並覈實它在生產環境中運行時知足這個概念。咱們一提到監控，經常想到跟蹤低級別性能統計數字的計數器（counter）和計量器（gauge）。這些種類的計量器經常能夠告訴你CPU負載很高，但其實沒法告訴你應用程序是否在作它應該作的事情。KSQL容許針對應用程序生成的原始事件流定義自定義度量指標，不管它們是日誌事件、數據庫更新仍是其餘任何類型的事件。

好比說，一個Web應用程序可能須要覈實：每當新客戶註冊，就發送歡迎電子郵件，建立新的用戶記錄，並對其信用卡計費。這些功能可能分散在不一樣的服務或應用程序中，你須要監控，確保對每一個新客戶而言，每一個操做都在某個服務級別協議（SLA）裏面（好比30秒）。

二、安全和異常檢測

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鏈接件輸入到Elastic，並在Grafana UI中加以可視化。安全用例經常酷似監控和分析。你尋找欺詐、濫用、垃圾郵件、入侵或其餘不良行爲方面的模式，而不是監控應用程序行爲或業務行爲。KSQL爲定義這些模式並查詢實時數據流提供了一種簡單、先進、實時的方法。

三、聯機數據整合

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';

許多公司進行的數據處理大部分屬於數據豐富（data enrichment）這個範疇：拿來來自幾個數據庫的數據，將其轉換，鏈接起來，並將數據存儲到鍵值存儲庫、搜索索引、緩存或其餘數據服務系統。長期以來，用於數據整合的ETL（提取、轉換和加載）做爲按期的批處理做業來加以執行。好比說，實時轉儲原始數據，而後每隔幾小時進行轉換，以實現高效查詢。對於許多用例而言，這種延遲是不可接受的。若是結合Kafka鏈接件使用，KSQL可以實現由批量數據整合轉變爲聯機數據整合。你可使用數據流-錶鏈接，藉助存儲在表中的元數據來豐富數據流，或者將數據流加載到另外一個系統以前，對PII（我的身份信息）數據執行簡單的過濾。

四、應用程序開發

許多應用程序將輸入數據流轉換成輸出數據流。好比說，負責爲在線商店從新排序庫存少的產品的進程可能須要銷售和發貨方面的數據流，才能計算訂單數據流。

至於用Java編寫的更復雜的應用程序，Kafka的原生數據流API可能正是咱們所須要的。不過針對簡單的應用程序，或者對Java編程不感興趣的團隊，簡單的SQL界面也許正是它們所尋找的。

KSQL中的核心抽象

KSQL在內部使用Kafka的Streams API（https://kafka.apache.org/documentation/streams/），它們使用一樣的核心抽象來用於Kafka端的數據流處理。KSQL中有兩個核心抽象，它們對應於Kafka Streams中的兩個核心抽象，讓你能夠處理Kafka主題：

一、STREAM：數據流是無限序列的結構化數據（「事實」，fact）。好比說，咱們可能有一個財務交易數據流，好比「Alice向Bob打款100美圓，而後Charlie向Bob打款50美圓」。數據流中的事實是不可變的，這意味着新的事實能夠插入到數據流中，但現有的事實根本沒法被更新或刪除。數據流能夠由Kafka主題來建立，或由現有的數據流和表來生成。

CREATE STREAM pageviews (viewtime BIGINT, userid VARCHAR, pageid VARCHAR)

WITH (kafka_topic='pageviews', value_format=’JSON’);

2.、TABLE：表是STREAM或另外一個TABLE的視圖，它表示不斷變化的事實的集合。好比說，咱們可能有一個表，含有最新的財務信息，好比「Bob當前的賬戶餘額是150美圓」。它至關於傳統的數據庫表，可是由數據加窗之類的數據流語義加以豐富。表中的事實是不可變的，這意味着新的事實能夠插入到表中，但現有的事實根本沒法被更新或刪除。表能夠由Kafka主題來建立，或由現有的數據流和表來生成。

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中做爲TABLE來讀取的話題，這方面的一個例子是獲取用戶元數據的主題，其中每一個事件表示特定用戶ID的最新元數據，不管是用戶的姓名、地址仍是喜愛選擇。

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

不妨看一個實際的演示。此演示顯示了你如何將KSQL用於實時監控、異常檢測和警報。針對點擊流數據的實時日誌分析有好幾種形式。在本文例子中，咱們標記出了在測試Web服務器上佔用太多帶寬的惡意用戶會話。監控惡意用戶會話是sessionization的許多應用之一。不過籠統地說，會話是用戶行爲分析的基本模塊。一旦你按照特定的會話標識符將用戶和事件關聯起來，就能夠構建許多類型的分析機制，從簡單的度量指標（好比訪問次數），到更復雜的度量指標（好比客戶轉換漏斗和事件流），不一而足。咱們在演示的最後環節顯示瞭如何在Elastic支持的Grafana儀表板上實時顯示KSQL查詢的輸出結果。

視頻播放器

00:00

05:09

內部結構

若是想及時瞭解Spark、Hadoop或者Hbase相關的文章，歡迎關注微信公共賬號：iteblog_hadoop

有一個KSQL服務器進程執行查詢。一組KSQL進程做爲一個集羣來運行。能夠經過啓動KSQL服務器的更多實例來動態添加更多的處理能力。這些實例具備容錯性：若是一個實例失效，另外幾個會接過它處理的工做。使用交互式KSQL命令行客戶軟件來啓動查詢，客戶軟件經過REST API向集羣發送命令。命令行讓你能夠檢查可用的數據流和表，執行新的查詢，檢查運行中查詢的狀態，並終止運行中查詢。在內部，KSQL是使用Kafka的Streams API構建的；它繼承了Kafka的彈性可擴展性、先進的狀態管理及容錯功能，還支持Kafka最近推出的只處理一次（exactly-once proecessing）語義。KSQL服務器嵌入這個機制，另外添加了分佈式SQL引擎（包括一些新穎的功能，好比提高查詢性能的字節碼自動生成）以及用於查詢和控制的REST API。

Kafka + KSQL顛覆數據庫

過去咱們談論了顛覆數據庫（https://www.confluent.io/blog/turning-the-database-inside-out-with-apache-samza/），如今咱們經過爲由內到外發生變化的數據庫添加SQL層來實現顛覆。

在關係數據庫中，表是核心抽象，日誌是實現細節。而在以事件爲中心的世界，數據庫已被顛覆，核心抽象不是表，而是日誌。這些表只是來源於日誌，隨着新數據進入到日誌，表不斷更新。中央日誌是Kafka，KSQL是引擎，讓你能夠建立所需的物化視圖，並將它們表示爲持續更新的表。而後，你能夠針對這類流式表運行時間點查詢（KSQL即將發佈該功能），爲日誌中的每一個鍵得到最新值，採起持續不斷的方式。

若是想及時瞭解Spark、Hadoop或者Hbase相關的文章，歡迎關注微信公共賬號：iteblog_hadoop

使用Kafka和KSQL完全顛覆數據庫，這對公司中能夠用數據流方式來表示和處理的全部數據派什麼用場帶來了很大的影響。Kafka日誌是數據流的核心存儲抽象，容許進入到離線數據倉庫的相同數據如今可用於數據流處理。其餘一切數據是基於日誌的流式物化視圖，不管是各類數據庫、搜索索引仍是公司中的其餘數據服務系統。如今可使用KSQL，以數據流的方式，執行建立這些派生視圖所需的全部數據豐富和ETL。監控、安全、異常及威脅檢測、分析以及故障應對能夠實時執行，而不是爲時太晚才執行。全部這些可供任何人使用，只要藉助一種對你的全部Kafka數據而言簡單又熟悉的SQL界面：KSQL。

若是想及時瞭解Spark、Hadoop或者Hbase相關的文章，歡迎關注微信公共賬號：iteblog_hadoop

KSQL的下一站是什麼？

咱們在發佈開發者預覽版的KSQL，開始圍繞它構建社區，並徵集反饋意見。咱們計劃與開源社區合做，增添另外幾項功能，從而將它變成一種可準備部署到生產環境的系統：從KSQL的質量、穩定性和可操做性，直到支持更豐富的SQL語法（包括進一步的聚合功能和針對連續表的時間點SELECT），對迄今爲止計算的數據執行快速查詢。

如何獲取KSQL？

你能夠試一試KSQL快速入門和上述演示來實際體驗一下。歡迎你反饋缺乏什麼功能，或者哪些方面能夠改進：歡迎到Confluent Community Slack上的#KSQL頻道發表任何想法或反饋，若是你發現了錯誤，歡迎提交GitHub問題單；咱們很樂意與早期採用者密切合做，因此請踊躍參與。咱們期待與開源社區的其他人合做，讓KSQL變成一項出色的技術。

本文英文原文：https://www.confluent.io/blog/ksql-open-source-streaming-sql-for-apache-kafka/