提升您的流數據處理能力—— Greenplum的流計算功能解析

​在追求數據時效性的今天，如何高效處理低延時的流數據，逐漸成爲你們愈來愈關注的問題。 流數據處理能力已經成爲衡量大數據平臺計算實力的一個重要指標。Greenplum做爲最早進的開源大數據平臺，天生具有處理複雜問題的優點。Pivotal的研發團隊在開源Greenplum的基礎上，提供了新的高速流數據引擎gpKafka，從而將Greenplum強大的SQL處理能力引入到流計算領域。本文重點介紹目前主要的流計算模式，以及gpKafka如何將Greenplum打形成近實時的流計算引擎。

今天，有愈來愈多的人在討論流計算和流數據，如同十年前討論的大數據同樣。對大數據的含義，直到今天，不一樣人對它仍然有不一樣的理解。有人認爲大數據就是Hadoop，有人說大數據是機器學習和AI，有人說大數據是數據在雲上等等。這些有差別的理解不利於咱們對問題自己的研究和交流。因此，在咱們開始討論流數據和流計算以前，要先對其具體含義作一個相對精確的描述和解釋，從而造成一個共識。因此咱們依次解釋這三個基本概念：流數據，流計算引擎及流計算模式。

什麼是流數據

對它下定義的話，很是簡單，沒有邊界的數據。好比車輛的位置信息，設備的運行狀態報告，網站的用戶點擊信息等。儘管它的定義很簡單，流數據有幾個比較重要的特色。第一個是流數據從產生處處理，存在延遲。所以流數據有兩個時間屬性：事件時間和處理時間。而處理時間的延遲，並無嚴格要求，可能很大，可能很小，可能時大時小變化很大；而這是流數據區別於實時數據的重要方面。流數據不是實時數據，實時數據不考慮事件時間和處理時間的差異。儘管隨着硬件性能的提高，不少原生的流處理引擎已經能夠支持部分軟實時的應用場景，但流數據和實時數據自己並無什麼必然聯繫，兩者之間有交集，但屬於不一樣的應用範疇。流數據第二個特色，它自己是能夠作到強一致性的。認爲流數據是不可靠的是一種偏見，或者只是爲技術上難以實現強一致性找到藉口。但根據具體使用場景的不一樣，應用能夠根據實際需求，來決定本身須要達到的一致性目標，好比強一致，最終一致，或者最多一次，最少一次等等。

什麼是流計算引擎

定義完流數據，再理解流數據引擎就比較容易。流數據引擎是專門處理無邊界數據的計算引擎。它也有兩個特色，首先是流計算引擎必須能夠知足強一致性要求。若是某個流計算引擎在設計上沒有將強一致性做爲考慮的目標，那它一定沒法保證結果的準確性，那也就不是一個真正的流計算引擎。第二個特色是流計算引擎須要支持全部的或者大部分流計算模式。流計算模式主要有三個不一樣的屬性。咱們下面來依次介紹。

首先是時間屬性， 一共分三類，事件時間，處理時間及時間無關。那爲何要把將事件時間和處理時間分開呢？緣由是一般事件時間和處理時間之間的延時，是不固定的，且變化可能很大。舉個例子來講有一個設備在記錄咱們的位置信息，並將記錄經過移動網絡實時上傳。然而當咱們坐上飛機起飛的時候，這個設備仍在工做，但沒法將數據上傳。而在飛機降落時，它會把累計的數據一塊兒上傳。相似的狀況在實際應用中很是常見。

窗口，或者叫開窗（英文windowing）指的是如何對流數據添加虛擬的邊界，從而將無邊界的流數據轉變成一個個有邊界的數據集；它是處理無邊界數據的最經常使用方法。添加的邊界一般有兩類，時間邊界和事件邊界，分別叫作時間窗口和會話窗口；時間窗口有兩種類型，固定窗口和滑動窗口；下面用個簡單的例子解釋下三者的區別。

Key表示咱們須要觀察的對象，每一行表示這個對象發生的事件，因此，這裏咱們有來自這三個事件數據流。固定窗口，表示以固定的時間間隔劃分數據流，每次處理的是這段時間內的數據。滑動窗口每次也是按照固定時間間隔執行處理函數，假設這個時間間隔爲T1，每次處理的數據是從當前時間開始，以前一段時間以內的數據，假設這個時間段爲T2。一般T1小於T2。當T1等於T2時，滑動窗口退化爲固定窗口。當T1大於T2，滑動窗口等價於對數據進行下采樣。而會話窗口的含義是，須要處理的事件，有明確的開始和結束的標誌。開始和結束標誌之間的一系列事件稱爲一個會話，而對數據的處理則是以會話爲單位。基於會話的場景是使用最多狀況， 實現時能夠轉換爲滑動窗口的方式。

介紹完時間和窗口，最後一個屬性就是計算的類型，咱們稱它爲運算，也就是咱們須要對流數據執行什麼樣的處理。從簡單到複雜，依次是，流數據的內鏈接，也就是在兩個流數據中找到共同的事件或相似的事件；第二個是數據的變換和過濾，好比簡單去重，單位轉換，到複雜的加密，脫敏等；第三種是最複雜的流數據聚合，即須要執行某種聚合函數從而識別出數據流的某些特徵。目前來看，執行流運算最佳的工具就是SQL。

爲何SQL是適合流計算的最佳工具？SQL有強大的表達和計算能力，有完善的標準和衆多的廠商支持。它從誕生那一天就遇到各類各樣的挑戰者，SQL解決複雜分析問題的能力是通過檢驗的。最重要的一點，這些問題其實都是SQL早已解決過一遍的問題。這也不難理解爲何當下趨勢就是愈來愈多的所謂「真正的流計算引擎」也陸陸續續提供了SQL的支持，例如spark，flink，ksql等。

如今來總結流數據處理的模式，簡單說就是這三個屬性的排列組合。 絕大部分流計算問題均可以劃分到這裏面的某一個類別，這些也是流計算引擎須要支持的計算模式。例如在基於事件時間的會話窗口上，進行網絡攻擊檢測；或者以固定窗口，對時間無關的數據進行加工變換等等。

既然，你們都但願在流計數據上執行SQL，一個簡單的思路當就是讓已有的SQL引擎來支持對流數據的查詢。好比你們熟悉的Pipelinedb，Timescaledb等等，他們都是在Postgres上進行擴展，將其改形成流數據引擎。Greenplum也是一樣，通過一些加強，就能夠將它變成分佈式流計算引擎。 接下來咱們從gpKafka開始看一下Greenplum如何知足流計算的應用場景。

Greenplum是最早進的開源大數據平臺，原生支持數倉分析，機器學習，文本分析，地理信息系統等功能，普遍應用在各個行業。

Kafka最初是設計爲一個分佈式的日誌系統，並普遍用於流數據的消息中間件場景。隨着Kafka擴展組件愈來愈多，它也慢慢演變成爲一個完整的流數據計算平臺。Kafka的核心組件遵循Unix設計哲學「do one thing and do it well」，於是在實現時犧牲了不重要的功能，確保了最重要的三個功能：高速，可靠，可擴展。

這是kafka的邏輯結構，topic是存放消息的隊列。每一個topic包含一個或多個partition，partition的數量決定了消費時的並行程度。producer生成消息，消息發送給某個topic，或者直接發給topic中特定的partition。consumer消費消息，consumer以組（consumer group）爲單位消費同一個topic的消息。同一個topic能夠有多組消費者同時消費，每一個組對應不一樣的app。在同一個組內部，每一個partition的消息，同時只能由一個消費者進行消費。不存在同一組內的多個消費者消費同一個partition的狀況。所以，增長producer的數量必定會提升發送端的並行度，但增長consumer的數量，則不必定會增長接受端的並行度，由於實際消費端的並行度的上限是由partition的數據決定的。

Kafka是分佈式集羣，Kafka集羣由Broker組成，每個Broker都是能夠獨立提供服務的進程實例。消息以分區爲組織單元，保存在Broker上。數據會有備份，稱爲replica， 備份保存在另外一個broker上，所以備份的數目不會超過Broker的數目，而分區數沒有這個限制，同一Broker上能夠有同一topic的多個分區。

在全部的備份中，有一個Leader對外提供讀寫服務，其它稱爲replica。只有在leader掛掉的狀況下，纔會從剩下的replica中選出新的Leader提供服務。爲最大化Kafka的吞吐，在流處理管道架構設計上必須考慮如何與Kafka自己的設計相匹配。例如在Broker數目固定的狀況下，增長topic和增長partition，本質上都是橫向擴展，更須要關注的是如何避免各個partition中數據的傾斜。

gpKafka是Greenplum的Kafka數據加載組件，官方名稱爲「Greenplum-Kafka integration」。它把Kafka高速的流處理能力和Greenplum強大的SQL執行能力聯合起來，大大下降了數據處理的延時，從而將Greenplum引入到近實時應用的場合。gpKafka是經Confluent官方認證的數據加載解決方案，支持Kafka從基本的數據加載到高級的元數據管理功能。gpKafka支持exactly-one等強一致性的使用場景，可運行在Cloudfoundry或者K8s上。 此外在接下來的版本里，gpkafka會增長的兩個最重要功能，橫向擴展和向Kafka卸載數據。

gpKafka利用gpss從Kafka的特定topic中讀取數據，而後轉發給Greenplum集羣。Gpss全稱是Greenplum Stream Server，是Greenplum的下一代數據加載解決方案，相比於gpfdist，GPSS會提供流數據支持及API接口，有更好的擴展性，支持更豐富的功能，並開放更細粒度的任務控制接口。

Gpss在讀取Kafka中的消息時，爲topic中每個partition建立一個reader做爲consumer。而後把來自於同一個topic的消息聚集到一塊兒，再經過Gpfdist協議轉發給各個Segment。Gpfdist協議是在HTTP協議基礎上作了加強，用於實現向Greenplum的Segment節點直接發送數據。Gpfdist和Gpss都使用的Gpfdist協議。

Gpss本身實現了Gpfdsit協議的服務端，自己並不包含Gpfdist可執行程序。Gpss有專門的controller服務來管理batch，controller決定什麼時候開始或結束一個加載batch，什麼時候執行數據轉換函數。GpKafka目前支持兩種定義batch的方式，基於時間的和基於消息數目的。GpKafka把從一個Kafka topic到一個Greenplum表的加載任務，定義爲一個job，每一個job經過yaml格式的配置來文件定義。每一個Gpss進程能夠同時執行多個job。最大同時運行的job數量取決於運行Gpss進程的機器的系統資源。

gpKafka支持的運算類型，包括消息數據進入Greenplum以前，通過的全部變換和處理。

灰色部分是正在實現中的功能、會在以後版本支持，主要用於數據脫敏、解密等。Formatter用於解析消息，除了消息長度外，Kafka自己並不對消息的內容作任何額外要求，不會修改和查看消息內容，消息能夠是任何格式，好比json，csv，avro，甚至二進制數據。應用程序徹底能夠根據本身的場景和優化目標決定使用哪一種文件格式。Json或csv文本可讀性強，但沒有壓縮，數據量大；壓縮格式會明顯的增長吞吐，但會增長收發端的CPU負載。一般在Kafka的絕大部分使用場景，網絡是明顯的瓶頸，比較推薦使用壓縮格式。Confluent官方推薦的格式avro。

Formatter的目的就是將這些不一樣格式的消息進行相應的變換，從而獲得Greenplum能夠識別的內容。例如對csv是直接解析，而avro格式則要先轉換成json。

Formatter以後的處理流程是transform，指的是具體對數據執行何種操做，在接下來會對其有專門的介紹。 最後一步操做是後處理，其主要目的是根據前一步變換的結果，對須要入庫的數據進行篩選。它經過指定的過濾條件，來保留須要的列或者數據。例如咱們只須要跟蹤分析某一個特定用戶行爲，就能夠在加載前識別出有效數據從而避免額外的數據清洗工做。

Transform是gpKafka中最強大最靈活的功能。它通過在外部表上執行函數的方式，在落盤以前將數據進行必要的轉換。例如提取圖像信息，非結構化到結構化數據的轉換，甚至執行機器學習函數等。Greenplum支持用各類經常使用語言來實現自定義函數，除了SQL外，還包括pl/C，pl/R，pl/python，pl/Java，pl/perl等，方便不一樣背景的用戶在實現時充分權衡開發效率和運行效率。此外，Transform的變換函數在各個segment上並行執行，不存在單點瓶頸。

介紹完運算，接下來介紹時間和窗口。 數據處理時間就是執行transform的時間，能夠經過now()函數來獲取。而事件時間一般保存在消息的記錄中，能夠在Mapping中指定。gpKafka經過minibatch的控制數據加載，batch的時間間隔就是數據加載時每次窗口移動的長度，經過配置文件的MINIMAL_INTERVAL指定。POST_BATCH_SQL用來指定窗口函數。窗口函數與transform函數相似，能夠是任意合法的SQL或者自定義函數，窗口大小能夠經過窗口函數的參數決定，或者由窗口函數自己來控制。

gpKafka默認將數據進行轉換後加載到Greenplum中，所以能夠保留全部的流原始數據。從而能夠方便的進行滑動窗口或者session運算。換句話說，gpKafka是經過滑動窗口的方式來實現對會話窗口的支持。

這裏用一個簡單的例子來展現滑動窗口到底是怎麼實現的。這個窗口函數很是簡單，只是計算窗口時間的消息個數。它的輸入參數i，表示了滑動窗口的窗口長度，經過一個簡單的where條件實現過濾。

如今讓咱們簡單總結一下：gpKafka完整支持流計算的批處理模式，支持區分事件時間和處理時間，支持固定窗口及滑動窗口，能夠經過時間窗口模擬會話窗口， 並能夠在這些窗口上執行各類高級的SQL操做，使用Greenplum強大的分析引擎。所以Greenplum能夠勝任不少流計算的應用場景，下面咱們看一個典型的例子。

這是一個用Greenplum5和gpKafka進行網絡日誌分析的典型應用，用於找到潛在的安全風險，識別網絡攻擊等。客戶監控全部的網絡通訊，將抓到的pcap包交給Kafka，而後經過gpKafka持續加載到Greenplum中利用Madlib進行分析訓練。最後將訓練好的模型發送給spark，進行低延時的各類運算。在Greenplum6中，對併發性和短小查詢作了大幅的性能優化，所以在Greenplum6之後的版本中，配合resource group作好資源隔離，不少計算也能夠直接在Greenplum中完成。現在，Greenplum正慢慢演進爲一個全功能的大數據計算平臺，在傳統的數倉以外，Greenplum一定會在更多領域發揮愈來愈大的做用。

本文根據李陽在2019年PostgreSQL杭州沙龍活動中的演講內容整理。

參考文獻

[1] Akidau, Tyler, Slava Chernyak, and Reuven Lax. Streaming Systems: The What, Where, When, and How of Large-Scale Data Processing. First edition. Beijing Boston Farnham Sebastopol Tokyo: O’Reilly, 2018.
[2] Psaltis, Andrew G. Streaming Data: Understanding the Real-Time Pipeline. Shelter Island, NY: Manning Publications, 2017.
[3] https://sookocheff.com/post/k...
[4] https://medium.com/@rem.baba/...