本文要點(diǎn)

• 大多數(shù)的流處理技術(shù)，需要開發(fā)人員使用Java或Scala等編程語言編寫代碼。

• KSQL是Apache Kafka的數(shù)據(jù)流SQL引擎，它使用SQL語句替代編寫大量代碼去實現(xiàn)流處理任務(wù)。

• KSQL基于Kafka的Stream API構(gòu)建，它支持過濾、轉(zhuǎn)換、聚合、連接、加窗操作和Sessionization(即捕獲單一會話期間的所有的流事件)等流處理操作。

• KSQL的用例涉及實現(xiàn)實時報表和儀表盤、基礎(chǔ)設(shè)施和物聯(lián)網(wǎng)設(shè)備監(jiān)控、異常檢測和欺騙行為報警等。

你會根據(jù)一分鐘前的交通信號燈過馬路嗎?當(dāng)然不會!當(dāng)前，現(xiàn)代企業(yè)或者出于競爭上的壓力，或者因為企業(yè)的客戶對產(chǎn)品或服務(wù)的交互方式有著更高的期望，它們也面對著同樣的需求。

如果人們在iPad上輕點(diǎn)按鈕就可以租賃和觀看最新的影片，那么為什么還要因為銀行賬戶吃緊而必須等待數(shù)小時?

數(shù)據(jù)在現(xiàn)代企業(yè)中處于核心地位，數(shù)據(jù)的量也在不斷增加中，并且持續(xù)快速變化。流處理技術(shù)正是支持企業(yè)實時利用這些洪流信息的一種技術(shù)。目前為重新塑造自身的業(yè)務(wù)，Netflix、奧迪、PayPal、Airbnb、Uber和紐約時報等上萬家企業(yè)已經(jīng)選擇了Apache Kafka®作為流處理平臺的事實標(biāo)準(zhǔn)。

人們的很多日�；顒樱�例如閱讀報紙、在線購物、預(yù)訂酒店或航班、搭乘出租車、玩電子游戲或是撥打電話，其后臺都已由Kafka提供支持。

為什么需要流處理?

為了說明流處理技術(shù)的作用，我在此給出一個適用于多個不同行業(yè)的很好例子。假設(shè)我們需要去實時創(chuàng)建并維護(hù)客戶的全面檔案。這樣做出于很多的原因，包括：

• 為創(chuàng)造更好的客戶體驗。例如，“這位高級客戶在過去五分鐘內(nèi)嘗試多次結(jié)賬購物車，但由于我們最近的網(wǎng)站更新錯誤而產(chǎn)生失敗。因此，我們需要立即向該客戶提供折扣，并對所造成的不良用戶體驗致歉。”

• 為盡量降低風(fēng)險。例如，“這筆新的付款操作似乎存在欺詐。因為該付款是在美國境外發(fā)起的，但客戶的手機(jī)應(yīng)用報告她身處紐約市。我們應(yīng)立即阻止這筆付款，并第一時間聯(lián)系該客戶。”

該用例需要實時匯集來自各種內(nèi)部渠道的以及一些可能外部渠道的數(shù)據(jù)，然后將這些信息整合到全面客戶檔案(也稱為客戶的“360度檔案”)中。而且一旦任何渠道有新的信息可用，檔案都會得到立即更新。

下圖描繪了我們?nèi)绾问褂肒afka實現(xiàn)該用例的高層設(shè)置。其中，客戶數(shù)據(jù)從各種來源的數(shù)據(jù)流中持續(xù)收集。全面客戶檔案保持在表中，表根據(jù)這些數(shù)據(jù)來源構(gòu)建并持續(xù)更新。所有這些操作都是實時的，并具有一定的規(guī)模。

 

 

圖1 從內(nèi)部和外部客戶數(shù)據(jù)流實時構(gòu)建全面客戶檔案

上圖的概念非常簡單，它與人們對人體神經(jīng)系統(tǒng)工作方式的理解幾乎匹配。神經(jīng)系統(tǒng)將來自眼睛、耳朵、四肢等傳感器的數(shù)據(jù)傳輸?shù)酱竽X，以便人們能夠快速做出明智的決定，例如過馬路是否安全。這就是為什么Kafka常被認(rèn)為是數(shù)字原生公司的“中樞神經(jīng)系統(tǒng)”。

然而，從目前的情況看，流處理領(lǐng)域入門的門檻還是相當(dāng)高的。當(dāng)前最廣為使用的一些流處理技術(shù)，包括Apache Kafka的Streams API，仍然需要用戶使用Java或Scala等編程語言編寫代碼，即使是實現(xiàn)最簡單的任務(wù)也是如此。對編程技巧的這種苛刻要求，已經(jīng)阻礙了許多企業(yè)充分利用流處理所能提供的優(yōu)勢。但值得慶幸的是，現(xiàn)在我們有了一種更簡單的方法。

KSQL簡介，Apache Kafka的數(shù)據(jù)流SQL引擎

KSQL于2017年推出，是Apache Kafka的數(shù)據(jù)流SQL引擎。KSQL降低了人們進(jìn)入流處理的門檻。用戶不必編寫大量的代碼，只需使用簡單的SQL語句就可以開始處理流處理。例如：

就這么簡單!雖然我們可能無法一眼看出，上面給出的KSQL流數(shù)據(jù)查詢在實現(xiàn)上是分布式的、容錯的、彈性的、可擴(kuò)展的和實時的，這些特性可以滿足現(xiàn)代企業(yè)對數(shù)據(jù)的需求。KSQL實現(xiàn)了這一目標(biāo)，它是建立在Kafka的Streams API之上的，充分地利用了Kafka在分布式流處理方面的強(qiáng)大技術(shù)基礎(chǔ)。

如果我們想使用Java或Scala直接調(diào)用Kafka的Stream API實現(xiàn)上述KSQL查詢，那么我們的應(yīng)用代碼段可能需要做如下編寫。當(dāng)然，這一代碼段還需要編譯、打包并應(yīng)用部署。

對于Java或Scala開發(fā)人員而言，Kakfa Streams API是一個強(qiáng)大的軟件庫，它實現(xiàn)了將流數(shù)據(jù)處理集成到應(yīng)用中。但是KSQL為開發(fā)人員提供了更寬廣的基礎(chǔ)，即一種僅使用SQL即可表達(dá)流數(shù)據(jù)處理需求的方式。

當(dāng)然，讀者還可以使用KSQL實現(xiàn)更多功能，不必局限于上面所展示的簡單例子。KSQL是采用Apache 2.0許可開源的，構(gòu)建于Kakfa的Streams API之上。這意味著，KSQL支持很大范圍的流數(shù)據(jù)處理操作，包括過濾、轉(zhuǎn)換、聚合、連接、加窗操作和Sessionization(即捕獲單一會話期間的所有的流事件)等。使用KSQL，可輕松實現(xiàn)：

• 驅(qū)動實時報告和儀表盤;

• 監(jiān)控基礎(chǔ)設(shè)施和物聯(lián)網(wǎng)設(shè)備;

• 檢測異常，并對欺詐活動報警;

• 分析基于會話的用戶活動;

• 執(zhí)行實時ETL;

• 以及更多操作……

下面給出幾個使用KSQL的例子。

例子一：使用KSQL實現(xiàn)在線數(shù)據(jù)集成和擴(kuò)充

企業(yè)開展的大部分?jǐn)?shù)據(jù)處理，都可歸為數(shù)據(jù)擴(kuò)充(Data Enrichment)或數(shù)據(jù)整理(Data Wrangling)，即如何從多個系統(tǒng)中提取、轉(zhuǎn)換和連接數(shù)據(jù)，并存儲到鍵值存儲、RDBMS、搜索索引、緩存等數(shù)據(jù)服務(wù)系統(tǒng)中。KSQL可與Kafka Connect的連接器一起使用，操作Oracle、MySQL、Elasticsearch、HDFS或S3等存儲系統(tǒng)，實現(xiàn)將批量數(shù)據(jù)集成轉(zhuǎn)換為實時數(shù)據(jù)集成。

下面的KSQL查詢使用了流數(shù)據(jù)表連接，將存儲在數(shù)據(jù)表中的元數(shù)據(jù)擴(kuò)充到數(shù)據(jù)流中：

為符合GDPR規(guī)范，，我們需要在加載數(shù)據(jù)流到其它系統(tǒng)之前將其中的PII(個人驗證信息，personally identifiable information)數(shù)據(jù)過濾掉。在此，我們需要移除上例中建立的vip_users數(shù)據(jù)流中的usr_id域。具體做法是，將不將該域添加到結(jié)果數(shù)據(jù)流中，在結(jié)果數(shù)據(jù)流中只保留了user_country、web_page和action域：

例子二：使用KSQL實現(xiàn)實時監(jiān)控和分析

盡管實時監(jiān)控和實時分析是兩種完全不同的用例，但是它們所需要實現(xiàn)的流數(shù)據(jù)處理功能是非常類似的。KSQL可以直接對原始事件流定義一些適當(dāng)?shù)亩攘�，無論數(shù)據(jù)流是生成自數(shù)據(jù)庫更新、應(yīng)用、移動設(shè)備、車輛等來源。

下例給出的查詢基于在五分鐘窗口內(nèi)觀察到車輛遙測數(shù)據(jù)中的錯誤數(shù)，實時計算可能出故障的車輛。該例是一類特殊的聚合操作，即窗口聚合。數(shù)據(jù)首先被分組為窗口數(shù)據(jù)(在本例的查詢中，分組和加窗操作是基于輸入數(shù)據(jù)中的時間戳信息)，然后每個窗口做單獨(dú)聚合。

KSQL的另一個用法是自定義業(yè)務(wù)層面的度量，這些度量是從監(jiān)控和報警中實時計算得到的。例如，展示一個AAA電子游戲特許經(jīng)營商(“最近的游戲擴(kuò)展是否增加了游戲時間?”)的并發(fā)在線玩家數(shù)量，報告電子商務(wù)網(wǎng)站中放棄購買的購物車數(shù)量(“我們最新的在線商店更新是否更加方便了客戶結(jié)賬?”)。類似，也可以使用KSQL為用戶的業(yè)務(wù)應(yīng)用定義一個用于表示是否正確的概念，進(jìn)而檢查該概念是否符合生產(chǎn)中的要求。

上面的查詢例子正好也是一個有狀態(tài)查詢的例子。有狀態(tài)的流處理可以說是流處理中最常用的功能，同時在實現(xiàn)與正確處理上非常具有挑戰(zhàn)性。下面我將做詳細(xì)介紹。

實現(xiàn)流數(shù)據(jù)處理中的記憶：有狀態(tài)流處理

例子二中的查詢對輸入流數(shù)據(jù)執(zhí)行聚合操作。聚合操作是一種有狀態(tài)的操作，即在操作中需要維護(hù)和更新狀態(tài)。例子二的查詢在觀測到新的錯誤前，需要記住每個時間窗口和每輛車的上一次錯誤計數(shù)情況，否則就無法確定查詢結(jié)果是否會超出五分鐘窗口期內(nèi)的車輛錯誤閾值。分布式流處理的一個主要挑戰(zhàn)，就是要在保證這種有狀態(tài)操作可以高效且正確工作的同時，考慮到諸如機(jī)器崩潰、網(wǎng)絡(luò)錯誤和大規(guī)模運(yùn)行等因素。

相比之下，無狀態(tài)操作更為簡單。計算可以在機(jī)器間自由遷移，這樣操作的代價很低，易于實現(xiàn)。而有狀態(tài)操作要實現(xiàn)計算的遷移，還需要執(zhí)行諸如將歷史狀態(tài)從故障機(jī)器移動到活動機(jī)器，并且要有效地完成，期間還可能會涉及以GB為單位的數(shù)據(jù)遷移。其中最重要的是，數(shù)據(jù)遷移必須正確地完成。例如，在例子二給出的KSQL查詢中，沒有人會希望僅僅因為相同的錯誤信息已經(jīng)得到多次處理，因此就向汽車司機(jī)發(fā)出引擎即將故障的虛假警報!

為實現(xiàn)更快的處理和更好的容錯能力，KSQL通常會運(yùn)行在多臺機(jī)器、虛擬機(jī)或容器上。那么KSQL如何解決有狀態(tài)的挑戰(zhàn)?答案是，KSQL建立是在Kafka的Streams API上的，這使得所有的KSQL查詢(包括有狀態(tài)查詢)具有如下特征：

容錯：在機(jī)器出現(xiàn)故障時，狀態(tài)和計算需要從故障機(jī)器自動遷移到活動的機(jī)器上。實現(xiàn)容錯，一方面需要持續(xù)地對從KSQL到Kafka的狀態(tài)做“流備份”，另一方面應(yīng)在需要時自動地從Kakfa將狀態(tài)恢復(fù)回KSQL。

彈性：用戶可以在操作現(xiàn)場中隨時添加并移除新機(jī)器，擴(kuò)展或縮小處理規(guī)模，而不會造成數(shù)據(jù)丟失，依然給出正確的處理結(jié)果。

擴(kuò)展性：將處理負(fù)載和狀態(tài)自動地擴(kuò)展到各臺機(jī)器，實現(xiàn)對數(shù)據(jù)的協(xié)作處理。擴(kuò)展性是通過使用Kafka的處理協(xié)議和分區(qū)數(shù)據(jù)存儲實現(xiàn)的。其中，處理任務(wù)根據(jù)數(shù)據(jù)的分區(qū)情況擴(kuò)展到各臺機(jī)器做并行處理。

由于這些屬性在KSQL中是開箱即可用的，因此用戶只需要專注于為自己的流處理需求，編寫所需的SQL語句。出于同一原因，KSQL非常適合構(gòu)建現(xiàn)代部署環(huán)境，例如基于Docker、Kubernetes或云原生的環(huán)境。

流-表二元性(Stream-Table Duality)

對數(shù)據(jù)流和表提供頭等支持，這是Kafka的一個獨(dú)有特性。讀者是否注意到，我們在前面的的例子中同時給出了數(shù)據(jù)流和表?例如，雖然例子二的輸入是一個數(shù)據(jù)流，但是該有狀態(tài)查詢的結(jié)果是一個表：

讀者可能會思考，“數(shù)據(jù)流和表兩者間有何差別?”，并且更為重要的是，“這種特性如何可用于我的日常工作中?”。簡而言之，該特性非常有用。表和數(shù)據(jù)流為用戶提供了必要的原語，可用于對數(shù)據(jù)建立推理和建模，回答對數(shù)據(jù)的業(yè)務(wù)問題。下面給出我能想到的一些最直觀的英文類比：

Kafka中的數(shù)據(jù)流是世界(或業(yè)務(wù))從一開始至今的完整歷史。它表示了過去和當(dāng)前。當(dāng)我們從當(dāng)前走向未來時，新的事件會不斷地添加到世界歷史中。在Kafka中，事件寫入、存儲并讀取自Kafka主題(Topic)。由于我們無法更改過去，因此Kafka是一種對事件不可變的、只添加的日志記錄。從分析RDBMS角度看，我們可以認(rèn)為數(shù)據(jù)流是對“事實”(Fact)的建模。

Kafka中的表是世界的當(dāng)前狀態(tài)(更通用的表述是某一時刻的狀態(tài))。它表示現(xiàn)在或過去的某個時刻，是世界事件歷史的一個聚合，該聚合在我們從當(dāng)前走向未來時會持續(xù)改變。表通過對數(shù)據(jù)流的處理而從流中獲取，更準(zhǔn)確地說是通過聚合這些數(shù)據(jù)流。在處理中使用了Kafka的Streams API和KSQL等工具。從分析RDBMS的角度看，我們可以認(rèn)為表是對“維度”(Dimension)的建模，保持了一個鍵的當(dāng)前值。

我們將這種內(nèi)在關(guān)系稱為“流-表二元性”(Stream-Table Duality)。如果讀者希望更深入了解這種數(shù)據(jù)流和表間的有意思關(guān)系，推薦大家閱讀我的一篇文章“Kafka和流數(shù)據(jù)處理中的數(shù)據(jù)流和表”。

稍等，那么表的概念出自何處?答案是，表來自于我們數(shù)十年在構(gòu)建應(yīng)用和服務(wù)中成功使用的數(shù)據(jù)庫。在數(shù)據(jù)庫中，表是首先需要構(gòu)建的結(jié)構(gòu)，它是各項工作的基礎(chǔ)。數(shù)據(jù)流實際上也存在于數(shù)據(jù)庫中，表現(xiàn)為構(gòu)建數(shù)據(jù)庫的交易日志(例如，MySQL的binlog，或者Oracle的Redo Log)。但這對用戶而言是不可見的，用戶并不直接操作這些數(shù)據(jù)流。我繼續(xù)使用前面的類比，一個數(shù)據(jù)庫知道現(xiàn)在，但它不知道過去。如果用戶需要過去，那么請取出備份磁帶。磁帶實際上可以看成是一種硬件流……

這樣，Kafka和流數(shù)據(jù)處理是數(shù)據(jù)庫的完全反轉(zhuǎn)。正如上文所說，我們首先要構(gòu)建數(shù)據(jù)流。而表是從數(shù)據(jù)流生成的。Pat Helland將此歸納為“所有變化均源自于不可變性”(“Immutability Changes Everything”)，“真相是日志(數(shù)據(jù)流)，數(shù)據(jù)庫是日志子集的一個緩存”。Kafka知道當(dāng)前，但也知道過去。這就是為什么紐約時報將其所有已發(fā)表的文章(可回溯至19世紀(jì)50年代的160年間的新聞報道)存儲在Kafka中，作為事實來源(Source of Truth)。

簡而言之，數(shù)據(jù)庫認(rèn)為表是最重要的，數(shù)據(jù)流次之;而Kafka認(rèn)為數(shù)據(jù)流最重要，表次之。在Kafka Streams和KSQL中，通過提供對數(shù)據(jù)流和表的原生支持，幫助用戶構(gòu)建了流數(shù)據(jù)處理和數(shù)據(jù)庫之間的橋梁。為使該特性更為強(qiáng)大，用戶可以使用Kafka Connnect將現(xiàn)有數(shù)據(jù)庫和表實時掛接到Kafka中。根據(jù)上面的陳述，我們完全可以給出這樣一個結(jié)論，即Kafka是一種“數(shù)據(jù)流關(guān)系”系統(tǒng)，而非“僅是數(shù)據(jù)流”的系統(tǒng)。

數(shù)據(jù)流和表的進(jìn)一步闡述

出于下述兩個重要原因，流-表二元性在實踐中是至關(guān)重要的。首先，企業(yè)現(xiàn)有數(shù)據(jù)庫中可能已經(jīng)存在了大量的數(shù)據(jù)，并且企業(yè)希望能將這些數(shù)據(jù)應(yīng)用于一些由流數(shù)據(jù)處理驅(qū)動的用例。其次，用戶一旦著手實現(xiàn)自己的流處理應(yīng)用，他們很快就會意識到，即使并不存在一個“真實”的數(shù)據(jù)庫，大多數(shù)用例實際上還是需要將數(shù)據(jù)建模為流和表。這是因為表代表“狀態(tài)”。無論何時要實現(xiàn)任何有狀態(tài)處理，包括執(zhí)行聚合(例如，計算某個關(guān)鍵業(yè)務(wù)度量的五分鐘平均值)或連接(例如，通過維度表連接事實“流”實現(xiàn)實時數(shù)據(jù)擴(kuò)充)，表都會涉及其中。

下面給出一個流和表的例子。該例子使用KSQL實時計算用戶地理位置的變更次數(shù)。例如，Strava這樣的移動應(yīng)用允許用戶手動簽到某個位置，并自動定期發(fā)送地理位置更新。查詢的輸入是一個地理位置更新數(shù)據(jù)流，輸出結(jié)果是一個不斷更新的表。由于COUNT()是一種聚合操作，因此查詢是一個有狀態(tài)操作，即為了累加當(dāng)前計數(shù)，首先必須記住當(dāng)前的計數(shù)值!下面給出KSQL查詢，它每秒執(zhí)行會數(shù)次地理位置更新。對于每秒數(shù)十萬次乃至更多此更新，操作也是同樣的。

在下一個例子中，我們根據(jù)訂單狀態(tài)計算“訂單”流的每小時匯總情況。這也是一個實踐中常見的用例。同樣，計算的結(jié)果是一個表('orders_hourly_aggregates')。一旦有新訂單到達(dá)，該表就會持續(xù)更新。該查詢還展示了一些可在KSQL中使用的標(biāo)量函數(shù)。

功能齊備(Batteries Included)的流數(shù)據(jù)處理

Kafka提供了一個功能齊備的流媒體平臺，可用于構(gòu)建應(yīng)用和系統(tǒng)。無論實施簡單的流數(shù)據(jù)擴(kuò)充，還是實現(xiàn)類似于欺詐檢測或360度用戶配置文件等更為復(fù)雜的操作，我們都需要一個易于使用的流處理解決方案，這正是所有功能和核心數(shù)據(jù)結(jié)構(gòu)齊備Kafka，特別是Kafka包括對流和表的頭等支持。如果缺乏這種支持，用戶最終需要構(gòu)建一些不必要的復(fù)雜架構(gòu)，將流(或僅支持流的)處理技術(shù)與Cassandra或MySQL等遠(yuǎn)程數(shù)據(jù)存儲結(jié)合在一起，才能啟用有狀態(tài)處理，并且可能還必須添加Hadoop / HDFS才能啟用支持容錯的處理。那么用戶需要同時拋接多少個科技球?

總結(jié)

本文是一次對使用KSQL(Apache Kafka的流SQL引擎)進(jìn)行流處理的旋風(fēng)之旅。文中給出了多個具體的例子，從更高層面介紹了KSQL是如何解決有狀態(tài)流處理的挑戰(zhàn)，以及Kafka和KSQL是如何通過對數(shù)據(jù)流和表提供很好的支持，為搭建數(shù)據(jù)流和數(shù)據(jù)庫世界之間的橋梁提供幫助。KSQL更易于讀者端到端地實現(xiàn)自己的用例。

如果讀者對KSQL產(chǎn)生了濃厚的興趣，我推薦如下資源：

下載KSQL，并按快速指南操作。

閱讀KSQL文檔。文檔中詳細(xì)介紹了KSQL的結(jié)構(gòu)、SQL語法指南，并提供了更多的例子和教程，其中包括一些基于Docker的變體。

加入GitHub上的KSQL社區(qū)，或者加入Confluent上Slack社區(qū) Community Slack組的“#ksql”頻道。

本文作者：Michael Noll 

查看英文原文： Democratizing Stream Processing with Apache Kafka and KSQL

標(biāo)簽： Mysql 安全 代碼 電子商務(wù) 電子商務(wù)網(wǎng) 電子商務(wù)網(wǎng)站 媒體 數(shù)據(jù)庫 搜索 通信 網(wǎng)絡(luò)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。