一、前言
本案例作為大數據框架在公共安全領域應用實踐的開篇之作，將從最基礎的數據架構體系優化講起。在接下來的章節裡將詳細描述Kafka的基本原理、Kafka增強組件以及基於Kafka的Lambda架構的具體應用場景以及相應的研發成果。
Lambda架構由Storm的作者Nathan Marz提出。旨在設計出一個能滿足。實時大數據系統關鍵特性的架構，具有高容錯、低延時和可擴展等特。
Lambda架構整合離線計算和實時計算，融合不可變（Immutability，讀寫分離和隔離 一系列構原則，可集成Hadoop，Kafka，Storm，Spark,HBase等各類大數據組件。 大數據系統的關鍵問題：如何實時地在任意大數據集上進行查詢？大數據再加上實時計算，問題的難度比較大。Lambda架構通過分解的三層架構來解決該問題：Batch Layer，Speed Layer和Serving Layer。如下圖所示意。
圖1.1 Lambda架構圖
數據流進入系統後，同時發往Batch Layer和Speed Layer處理。Batch Layer以不可變模型離線存儲所有數據集，通過在全體數據集上不斷重新計算構建查詢所對應的Batch Views。Speed Layer處理增量的實時數據流，不斷更新查詢所對應的Real time Views。Serving Layer響應用戶的查詢請求，合併Batch View和Real time View中的結果數據集到最終的數據集。
二、基於Kafka的Lambda架構
2.1 某省大數據平臺實踐案例
以某省廳大數據建設方案為例，將Kafka作為統一的數據流通道（data pipeline）。Kafka分為地市和省廳兩級，地市數據首先經過流式化處理髮送到地市的Kafka，經過標準化後，地市Kafka的再彙集到省廳Kafka。
某省大數據平臺實踐
2.2 引入Kafka的必要性
在大數據系統中，常常會碰到一個問題，整個大數據是由各個子系統組成，數據需要在各個子系統中高性能、低延遲的不停流轉。傳統的企業消息系統並不是非常適合大規模的數據處理。容易造成日誌數據難以收集，容易丟失信息，Oracle實例之間的管道無法供其它系統使用，數據架構易創建難擴展，數據質量差等問題。為了同時搞定在線應用（消息）和離線應用（數據文件，日誌），Kafka就出現了。Kafka可以起到兩個作用：
• 降低系統組網複雜度。
• 降低編程複雜度，各個子系統不再是相互協商接口，各個子系統類似插口插在插座上，Kafka承擔高速數據總線的作用。
傳統數據架構
引入Kafka後，可以構建以流為中心數據架構。Kafka是作為一個全局數據管道。每個系統都向這個中心管道發送數據或者從中獲取數據。應用程序或流處理程序可以接入管道並創建新的派生流。這些派生流又可以供其它各種系統使用。
以流為中心的數據架構
三、Kafka技術分析
3.1 Kafka的特點
Kafka可以讓合適的數據以合適的形式出現在合適的地方。Kafka的做法是提供消息隊列，讓生產者單往隊列的末尾添加數據，讓多個消費者從隊列裡面依次讀取數據然後自行處理。
Kafka消息隊列
• 分佈式系統，易於向外擴展。所有的producer、broker和consumer都會有多個，均為分佈式的。無需停機即可擴展機器。
• 提供Pub/Sub方式的海量消息處理。 據瞭解，Kafka每秒可以生產約25萬消息（50 MB），每秒處理55萬消息（110 MB）。
• 以高容錯的方式存儲海量數據流。
• 保證數據流的順序，處理關鍵更新。
• 提供消息的長時間存儲，將消息持久化到磁盤，因此可用於批量消費，例如ETL，以及實時應用程序。通過將數據持久化到硬盤以及replication防止數據丟失。
• 能夠緩存或持久化數據，支持與批處理系統（如Hadoop）的集成。
• 為實時應用程序提供低延時數據傳輸和處理。
• 支持online和offline的場景。
• 消息被處理的狀態是在consumer端維護，而不是由server端維護。當失敗時能自動平衡。
3.2 Kafka原理分析
3.2.1 Kafka總體架構
Kafka總體架構
Kafka的整體架構非常簡單，是顯式分佈式架構，producer、broker（kafka）和consumer都可以有多個。Producer，consumer實現Kafka註冊的接口，數據從producer發送到broker，broker承擔一箇中間緩存和分發的作用。broker分發註冊到系統中的consumer。broker的作用類似於緩存，即活躍的數據和離線處理系統之間的緩存。客戶端和服務器端的通信，是基於簡單、高性能且與編程語言無關的TCP協議。
基本概念：
• Topic：特指Kafka處理的消息源（feeds of messages）的不同分類。
• Partition：Topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。
• Message：消息，是通信的基本單位，每個producer可以向一個topic（主題）發佈一些消息。
• Producers：消息和數據生產者，向Kafka的一個topic發佈消息的過程叫做producers。
• Consumers：消息和數據消費者，訂閱topics並處理其發佈的消息的過程叫做consumers。
• Broker：緩存代理，Kafka集群中的一臺或多臺服務器統稱為broker。
3.2.2 Kafka關鍵技術點
3.2.2.1 zero-copy
在Kafka上，有兩個原因可能導致低效：一是太多的網絡請求，二是過多的字節拷貝。為了提高效率，Kafka把message分成一組一組的，每次請求會把一組message發給相應的consumer。 此外，為了減少字節拷貝，採用了sendfile系統調用。
3.2.2.2 Exactly once message transfer
在Kafka中僅保存了每個consumer已經處理數據的offset。這樣有兩個好處：一是保存的數據量少；二是當consumer出錯時，重新啟動consumer處理數據時，只需從最近的offset開始處理數據即可。
3.2.2.3 Push/pull
Producer 向Kafka推（push）數據，consumer 從kafka 拉（pull）數據。
3.2.2.4 負載均衡和容錯
Producer和broker之間沒有負載均衡機制。broker和consumer之間利用zookeeper進行負載均衡。所有broker和consumer都會在zookeeper中進行註冊，且zookeeper會保存他們的一些元數據信息。如果某個broker和consumer發生了變化，所有其他的broker和consumer都會得到通知。
3.2.2.5 分區
Kafka可以將主題劃分為多個分區（Partition），會根據分區規則選擇把消息均勻的分佈到不同的分區中，這樣就實現了負載均衡和水平擴展。多個訂閱者可以從一個或者多個分區中同時消費數據，以支撐海量數據處理能力。由於消息是以追加到分區中的，多個分區順序寫磁盤的總效率要比隨機寫內存還要高，是Kafka高吞吐率的重要保證之一。
Kafka分區實現負載均衡，水平拓展，高吞吐率
為了保證數據的可靠性，每個分區節點都會設置一個Leader，以及若干節點當Follower。數據寫入分區時，Leader除了自己複製一份，還會將數據複製到每個Follower上。若任一follower掛了，Kafka會再找一個follower從leader獲取數據。若Leader掛了，則從Follower中抽取一個當Leader。
Kafka分區實現數據的可靠性
3.3 Kafka的技術選型
3.3.1 Confluent Platform概述
Confluent Platform 是一個流數據平臺，能夠組織管理來自不同數據源的數據，擁有穩定高效的系統。Confluent Platform 很容易的建立實時數據管道和流應用。通過將多個來源和位置的數據集成到一箇中央數據流平臺。Confluent Platform簡化了連接數據源到Kafka、Kafka構建應用程序，以及安全、監控和管理Kafka的基礎設施。
Confluent Platform架構
3.3.2 Kafka Connect
Kafka Connect，可以更方便的創建和管理數據流管道。它為Kafka和其它系統創建規模可擴展的、可信賴的流數據提供了一個簡單的模型，通過connectors可以將大數據從其它系統導入到Kafka中，也可以從Kafka中導出到其它系統。Kafka Connect可以將完整的數據庫注入到Kafka的Topic中，或者將服務器的系統監控指標註入到Kafka，然後像正常的Kafka流處理機制一樣進行數據流處理。而導出工作則是將數據從Kafka Topic中導出到其它數據存儲系統、查詢系統或者離線分析系統等。
Kafka Connect特性包括：
• Kafka connector通用框架,提供統一的集成API
• 同時支持分佈式模式和單機模式
• REST 接口，用來查看和管理Kafka connectors
• 自動化的offset管理，開發人員不必擔心錯誤處理的影響
• 分佈式、可擴展
• 流/批處理集成
Kafka connect工作原理
3.4 Kafka端到端審計
採用開源的Chaperone技術框架來實現對kafka的端到端審計。其目標是在數據流經數據管道的每個階段，能夠抓住每個消息，統計一定時間段內的數據量，並儘早準確地檢測出數據的丟失、延遲和重複情況。
• 是否有數據丟失？是，那麼丟失了多少數據？它們是在數據管道的哪個地方丟失的？
• 端到端的延遲是多少？如果有消息延遲，是從哪裡開始的？
• 是否有數據重複？
Chaperone架構
Chaperone架構：AuditLibrary、ChaperoneService、ChaperoneCollector和WebService，它們會收集數據，並進行相關計算，自動檢測出丟失和延遲的數據，並展示審計結果。在審計過程中保證每個消息只被審計一次，在層間使用一致性的時間戳。
Chaperone模塊審計流程如下：
1. 生成審計消息：ChaperoneService通過定時向特定的Kafka主題生成審計消息來記錄狀態
2. 審計算法：AuditLibrary實現了審計算法，它會定時收集並打印統計時間窗
3. 獲取審計結果：ChaperoneCollector監聽特定的Kafka主題，並獲取所有的審計消息，存到數據庫，生成儀表盤。儀表盤展示：數據的丟失情況、消息的延遲情況、查看每個主題中心的主題狀態
4. 準確展示結果：WebService提供了REST接口來查詢Chaperone收集到的度量指標。通過這些接口，我們可以準確地計算出數據丟失的數量。
四、Kafka應用成果介紹
基於Kafka的技術特性，Kafka已成熟運用於某省廳的資源服務平臺項目，主要用於收集日誌、海量數據的微ETL，為各業務系統之間的數據共享提供一個大規模消息處理平臺，以及在各地市與省廳之間形成一個數據管道。
結合對Kafka和Kafka插件的深入研究，新德匯大數據研究院自主研發了輕量級的FSP流處理引擎，用於輕便對接流數據，高效處理和實現各類流數據延展應用。
4.1 日誌聚合
多個系統之間的日誌通過kafka匯聚，提供審計或其他監控系統進行消費。日誌聚合一般來說是從服務器上收集日誌文件，然後放到一個集中的位置（文件服務器或HDFS）進行處理。然而Kafka忽略掉文件的細節，將其更清晰地抽象成一個個日誌或事件的消息流。這就讓Kafka處理過程延遲更低，更容易支持多數據源和分佈式數據處理。比起以日誌為中心的系統比如Scribe或者Flume來說，Kafka提供同樣高效的性能和因為複製導致的更高的耐用性保證，以及更低的端到端延遲。
4.2 消息系統
系統之間解耦，通過kafka驅動各業務系統之間的數據共享與業務流程驅動。
比起大多數的消息系統來說，Kafka有更好的吞吐量，內置的分區、冗餘及容錯性，讓Kafka成為了一個很好的大規模消息處理應用的解決方案。消息系統一般吞吐量相對較低，但是需要更小的端到端延時，並常常依賴於Kafka提供的強大的持久性保障。在這個領域，Kafka足以媲美傳統消息系統，如ActiveMR或RabbitMQ。
4.3 數據管道
Kafka讓集成工作只需連接到一個單獨的管道，而無需連接到每個數據生產方與消費方。
Kafka提供數據管道，讓多個地市各種類型的數據資源，集成時不需要知道原始數據源的細節，發佈數據時也不需要知道哪個應用程序會消費和加載這些數據，增加新系統，也只需要接入現有的Kafka流數據平臺就可以。
某省廳Kafka數據管道案例
4.4 ETL流水線
未引入kafka時，數據的ETL過程需生成臨時數據庫，多次產生落地的文件，耗費內存，而且在再調用臨時數據庫時，會耗用內存。這樣厚重的架構也不具備流數據處理能力。
引入kafka後，實現微ETL。通過Kafka對接流處理引擎，簡化ELT流程，細化數據處理層次，低延時獲取目標數據。
微ETL優點：
• 無縫銜接流處理引擎，完成數據快速ETL
• kafka構建一個可伸縮的，可靠的數據流通道
• 交互低延遲
• 微ETL實現輕便的數據處理流程
傳統ETL與微ETL的對比
4.5 FSP流處理引擎
4.5.1 FSP架構
FSP架構
流處理平臺：對流數據，提供核心處理引擎，流採集工具的可配置化管理平臺
核心處理引擎：PIPELINEDB允許我們通過sql的方式，對數據流做操作，並把操作結果儲存起來；Kafka插件可擴展kafka功能，實現SQL on kafka的各類流數據的延展應用
流採集工具集：Kafkacat實現Kafka與 sqluldr、copy收集的數據的對接，實現流數據的採集
4.5.2 Kafkacat
4.5.2.1 抓取發送消息的工具
Kafkacat是NON JVM TOOL，速度快，輕便，靜態編譯小於150kb，提供元數據列表展示集群/分區/主題。
Kafkacat工作模式
4.5.2.2 通過kafkacat命令加載數據生成GP外部表
通過Kafkacat實現GP與kafka的數據對接：kafkacat工具根據外部表協議可以獲取GP和kafka的數據，並生成外部表，實現數據的並行加載。以外部表的形式實現數據格式錯誤行的容錯處理
Kafkacat 加載GP外部表
五、Kafka延展應用展望
整合NiFi與kafka，並將MiNiFi作為數據採集器布放到對端數據源，形成一條可拓展並流動的流式數據處理生產線。
Kafka與NiFi結合
5.1 NiFi介紹
NiFi是一個易用、強大、可靠的數據處理與分發系統。簡單來說，NiFi是用於自動化管理系統之間的數據流。通過與Kafka的對接，提供可視化命令與控制，實現數據流的展示與編輯處理功能，實現數據流的全程追蹤。
NiFi特點：
1.可視化命令與控制
基於Web的用戶界面，無縫體驗設計，監視，控制數據流。
2. 高擴展性
NiFi通過提供自定義類裝載器模型，來確保每個擴展組件之間的約束關係被限制在非常有限的程度。因此，在創建擴展組件時，就不用再過多關注其是否會與其他組件產生衝突。數據流處理程序能夠以可預測和可重複的模式執行。
3. 數據回壓
NiFi提供所有隊列數據的緩存，並且在隊列達到指定限制或者超時的時候，能夠提供數據回壓。
4. 高度可配置
數據丟失容錯和保證交付，低延遲和高吞吐量，動態優先級，流可以在運行時修改。
5. 安全性
系統間，NiFi可以通過雙向SSL進行數據加密。並且可以允許在發送與接收端使用共享密鑰，及其他機制對數據流進行加密與解密。
用戶與系統間，NiFi允許雙向SSL鑑定，並且提供可插入授權模式，因此可以控制用戶的登錄權限（例如：只讀權限、數據流管理者、系統管理員）。
5.2 NiFi實現統一實時採集數據的分佈式流平臺
數據實時採集器MiNiFi：
• 實現增量數據和流數據的實時採集，而不是傳統的定時採集，實現了更細緻化的數據獲取
• 可支持多種數據源，適用性強
• 實現端到端的數據採集
分佈式流平臺NiFi：
• 採集而來的數據，形成數據流，並對數據源進行自動記錄，索引，跟蹤
• 精確控制數據流
• NIFI單節點的性能是每秒處理百兆級數據，搭建NIFI集群可以提升到每秒處理G級別數據
NiFi分佈式流平臺
作者介紹：
楊剛，現任珠海市新德匯信息技術有限公司副總經理兼大數據研究院院長 15年IT從業經驗，長期從事雲和大數據的技術研發和實施工作，有深厚的電信、政務、金融等行業背景。