大數據的近實時分析系統架構

近實時分析的場景

近實時分析 – 對變化中的數據?供快速分析能力

• 分析現實世界中正在發生的事件的能力，結合歷史數據和實時流數據進行彙總分析、預測和明細查詢
• 絕對實時和批量不可調和，"近實時" 的意思是這是人機交互中能感覺的尺度（秒級），而不是機器自動處理的實時性量級（ns / us級）
• 數據價值從非結構化到結構化，分析從非範式到範式。SQL是結構化分析的最終手段，可是：
  • 彙總分析（順序掃?）與明細查詢（隨機掃描）
  • 小數據量下都不是問題；可是放在海量數據下看，兩種負載難以調和
  • 海量數據和實時流窗口上的SQL引擎實現也徹底不一樣
• 更接近實時Hadoop上是徹底可行的，可是實時性要求會帶來架構上的巨大變化

典型場景

須要同時支持順序和隨機讀／寫的應用場景

●在線交互式BI分析/決策輔助

○ 場景舉例: 貸後風險實時監測，實時資產偏好視圖，歷史風險偏好趨勢，市場監測

○ 應用類型: 須要準實時的同步插入/修改，同時彙總分析和單條查詢

● 時間序列數據

○ 場景舉例: 股市行情數據; 欺詐檢測和預防; 風險監控；線上實時反欺詐

○ 應用類型:須要實時捕獲流數據，同時結合已有的T+1數據進行彙總、分析和計算

● 機器日誌數據分析

○ 場景舉例: 臺機監控、故障預警

○ 應用類型:須要過濾大量流數據，同時結合已有的T+1數據進行彙總、分析和計算

   

更實時的、交互式BI

傳統數倉中增長實時彙總分析能力

   

   

物聯網（IoT）產生的實時分析和預測

   

車聯網

• 歷史分析
  • 開發人員但願知道如何優化充電性能
  • 新版本軟件升級後隨着時間推移是如何影響汽車性能的？
• 實時洞察
  • 客戶但願知道是不是未成年人在駕駛。他們加速多快？

    時速多少？他們在哪裏？

  • 汽車設備——好比在服務前或服務中拿到最新的診斷數據包

   

源於互聯網的Lambda 架構

   

Lambda 架構

   

   

企業應用中Lambda的典型實現方式

   

   

車聯網的實時數據處理

Hbase Provides:

• Fast, Random Read & Write Access

• "Mini-scans"

• Scale-out architecture capable of serving Big Data to many users

   

車輛網歷史數據分析方案

   

   

構建在混合架構上的分析管道

   

可是，HBase+HDFS混合架構的複雜性無處不在

同時供高性能的順序掃?和隨機查詢，避免使用HBase+HDFS混合架構的複雜性：

• 開發：必須編寫複雜的代碼來管理兩個系統之間的數據傳輸及同步

• 運維：必須管理跨多個不一樣系統的一致性備份、安全策略以及監控

• 業務：新數據從達到HBase到HDFS中有時延，不能立刻供分析

在實際運行中，系統一般會遇到數據延時達到，所以須要對過去的數據進行修正等。若是使用不可更改的存儲（如HDFS文件），將會很是不便。

   

Lambda 複雜性一：同步

   

Lambda 複雜性二：錯誤難以診斷

   

   

Lambda Pros & Cons

• Pros

  • 成功將不一樣領域的開源框架嫁接到一個統

  一架構內，應對不一樣類型的混合負載

  • Batch Layer可應對數據的無限擴展

  • Speed Layer可?供準實時的響應性能

•  Cons - Complexity

  • 須要大量的數據在不一樣存儲和格式中遷移，形成維護困難

  • 數據結構從新聲明或者數據修改都很困難

  • Batch Layer和Speed Layer須要維護兩套代碼，但其實現邏輯須要一致

  • 意外錯誤的捕獲、處理和衝正很是複雜

  • 前端查詢的複雜度很是大，須要合併數據集

基於Kudu實現簡單的近實時分析

當前的欺詐檢測架構：存儲架構太複雜

   

 可是怎 樣處理下面的問題 ？

● 怎麼有效處理轉換過程當中的錯誤？

● 如何定義將HBase數據轉換成Parquet格式做業的週期？

● 從數據進入到報表中能體現之間的時延如何量化？

● 做業流程怎麼保障不被其餘操做打斷？

使用Kudu的Hadoop實時數據分析

   

改進點 :

● 只要一套系 統

●不須要後臺定 時的批處理任務

● 輕鬆應對數據遲到和數據修正

●新數據當即用於在分析和 業務運營

 

Kudu: 在快速變化的數據上進行快速分析

   

   

Kudu的設計目標

 • 掃描大數據量時吞吐率 高(列式存儲和多副本機制)

目標 : 相對Parquet的掃?性能差距在2x以內

• 訪問少許數據時延時低(主鍵索引和多數佔優複製機制)

目標 : SSD上讀寫延時不超過1毫秒

• 相似的數據庫語義(初期支持單行記錄的ACID)

• 關係數據模型

• SQL查詢
• "NoSQL"風格的掃?/插入/更新(Java客戶端)

Kudu的使用

相似SQL 模式的表

• 有限的列數 (不一樣於HBase/Cassandra)

• 數據類型: BOOL, INT8, INT16, INT32, INT64, FLOAT, DOUBLE, STRING, BINARY,TIMESTAMP

• 一部分列構成聯合主鍵

• ALTER TABLE快速返回

"NoSQL" 風格的 Java和C++ APIs

• Insert(), Update(), Delete(), Upsert(), Scan()

與MapReduce, Spark 和Impala 的無縫對接

• 將對接更多處理引擎!

車輛網：一致的架構處理異構的數據分析管道

   

在CDH技術堆棧上的準實時分析技術

   

基於OGG 的數據庫日誌解析和Apache Kudu 的實時分析

•Kudu Adapter (Handler)幫助保持DB和Kudu之間基於日誌解析的數據同步。

•使用OGG Java API將DB事務解碼爲kudu特定的事務。

•使用KUDU API在Kudu結束執行事務操做。

•Kudu Adapter (Handler) 支持數據的Inserts, Updates, Upsert 及Deletes 事務操做.

   

更通用的實時數據處理集成／分析架構

• 與Apache Spark Streaming 集成進行real-time 的數據分析

• 處理完的數據再接入Kafka進行進一步的處理和供下游系統進一步分析

使用案例分享

   

小米(MI) 簡介

   

   

使用案例1

移動服務監聽及跟蹤工具

目標：

收集從移動App及後臺服務發起的RPC程序調用重要的跟蹤事件

服務監聽及錯誤處理工具

需求：

   

• 高寫吞吐

  >50億條/天的寫能力，且持續增加

• 快速查詢最新記錄並作響應

  快速定位錯誤並做出響應

• 可以確保單條記錄快速查詢

  更容易進行差錯

     

沒有Kudu 以前大 數據分析處理

   

   

在kudu以前，咱們的大數據分析pipeline大概是有這幾種：

1. 數據源-> scribe打日誌到HDFS -> MR/Hive/Spark -> HDFS

Parquet -> Impala -> 結果service這個數據流通常用來分析各類日誌。

2. 數據源 -> 實時更新HBase/Mysql -> 天天批量導出Parquet->

Impala -> 結果serve這個數據流通常用來分析狀態數據，也就是通常須要隨機更新的數據，好比用戶profile之類。

 這兩條數據流主要有幾個問題：

1. 數據從生成到可以被高效查詢的列存儲，整個數據流延遲比較大，通常是小時級別到一天；
2. 不少數據的日誌到達時間和邏輯時間是不一致的，通常存在一些隨機延遲。
3. 好比不少mobile app統計應用，這些tracing event發生後，極可能過一段時間才被後端tracing server收集到。咱們常常看到一些hive查詢，分 析一天或者一小時的數據，可是要讀2-3天或者多個小時的日誌，而後過濾出實際想要的記錄。
4. 對於一些實時分析需求，有一些能夠經過流處理來解決，不過他確定沒用SQL方便，另外流式處理只能作固定的數據分析，對ad-hoc查詢無能爲力kudu的特色正好能夠來配合impala搭建實時ad-hoc分析應用。

大數據分析管道－ 由於Kudu

改進後的數據流大概是這個樣子：

1. 數據源 -> kafka -> ETL(Storm) -> kudu -> Impala

2. 數據源 -> kudu -> Impala

3. 數據流1 主要是爲須要進一步作ETL的應用使用的，另外kafka能夠當作一個buffer，當寫吞吐有毛刺時，kafka能夠作一個緩衝。若是應用嚴格的實時需求，就是隻要數據源寫入就必須可以查到，就須要使用數據流2。

 

案例1: Benchmark

環境：

• 71 節點集羣
• 硬件

  CPU: E5-2620 2.1GHz * 24 core Memory: 64GB

  Network: 1Gb Disk: 12 HDD

• 軟件

  Hadoop2.6/Impala 2.1/Kudu

數據：

   

1 天的服務器端跟蹤數據

   

~26億行記錄

~270 bytes/行

每條記錄17 字段, 5 關鍵字段

案例 1: Benchmark 結果

使用 impala 進行批加載 (INSERT INTO):

   

查詢延時：

   

* HDFS parquet 文件複製因子= 3 , kudu 表複製因子= 3

*結果爲每條查詢執行5次並取平均值 

案例2: 京東案例

   

Jd.com 中國第二大在線電商

1. 使用Kafka實時收集數據

   • 點擊流日誌

   • 應用/瀏覽器Trace日誌

   • 每條記錄約70字段

2. 6/18 sale day

   • 150億筆交易

   • 高峯期每秒一千萬條數據插入

   • 集羣200臺服務器

3. 查詢使用JDBC -> Impala -> Kudu

   

案例3 某互聯網金融 使用 Kafka 、Spark 、Kudu 和Impala 構建

   

 業務需求：

• 根據當前客戶的操做行爲進行風險等級實時分析，防範金融風險

架構說明：

• 數據源Stream API的數據由Kafka接入
• Spark Streaming消費Kafka數據，並注入到Kudu中
• 流數據接入Spark Streaming做業進行實時處理，並使用Mlib進行預測
• 預測的結果保存到Kudu
• 客戶使用Impala或Spark SQL進行交互式結果查詢
• 分析工具使用JDBC接口訪問數據進行分析

   

案例4 某銀行使用 Kafka 、Kudu 和Impala 實現準實時數據倉庫應用

   

 業務需求：

• 數倉應用創建在多維分析模型上，維度表須要根據須要保留歷史記錄

架構說明：

• 數據源Stream API的數據由Kafka接入到Spark Streaming或Flume，並保存到Kudu
• 經過Impala對維度數據進行SCD操做：
  • SCD I： 存在即update
  • SCD II： 存在則先insert一條新記錄，並更新歷史記錄，如End Time 或 Flag
• 客戶使用Impala進行交互式結果查詢
• 分析工具使用JDBC接口訪問數據進行分析

案例5 使用 Kafka、Kudu 和Impala實現準實時數據倉庫分析應用

 

業務要求：

客戶流式計算測試要求實現Hadoop產品從KAFKA快速加載數據，主要有2個應用場景：

• Append模式即簡單將記錄添加到數據表中，相似MySQL的insert into，並須要保證數據不重複。

• Insert_update模式即對於有主鍵的數據表，若是新的記錄的主鍵在數據表中已存在，則用新的記錄update舊的記錄，若是新的記錄的主鍵 在數據表中不存在，則將新的記錄insert導數據表中。

設計實現思路：

1. 基於本次流式計算的測試要求，不管是Append仍是Insert_update原本均可以經過使用HBase來實現，由於HBase的Rowkey能夠保證數據的惟一性約束，達到Append去重的目的，而HBase的API也支持經過Rowkey去更新已經存在的數據。

2. 可是在本次流計算測試的性能場景要求中，還須要測試混合負載，須要進行數據集的統計查詢，即在入庫的同時須要進行大量的SQL統計分析查詢，還包括join操做。這樣HBase確定沒法知足，由於HBase只適合於隨機插入以及簡單的Rowkey條件查詢。

因此咱們最終選擇了Kudu來完成，既能夠知足快速的隨機插入，包括去重和更新操做，同時支持併發的SQL查詢的混合負載要求。

   

整體架構設計

   

• Append：Kudu是用於存儲結構化的表。表有預約義的帶類型的列（Columns），每張表有一個主鍵（primary key）。主鍵帶有惟一性（uniqueness）限制，可做爲索引用來支持快速的隨機查詢。若是咱們使用insert()方法，經過Kudu主鍵的惟一性，咱們能夠達到去重的目的，當有重複數據導入的時候，Kudu自身會經過主鍵判斷，若是存在，則直接丟棄。

• Insert_update：Kudu源生提供了Upsert()方法，直接能夠達到本次測試insert_update的目的，即根據主鍵判斷，若是存在則更新該數據，不然則做爲新的數據插入到Kudu

   

測試數據集及測試結果

本次測試主要用到了五個表：

HDFS中的表主要用來SQL混合負載的join表，並驗證Impala跨存儲執行性能。

   

 Append，無重複

   

Upsert，去重

   

Upsert，去重時有SQL查詢的混合負載

   

穩定的入庫速度

   

0min-6min，

指數行情：63萬條/秒，

現貨行情:18萬條/秒，

委託：40萬條/秒，

成交：35萬條/秒，

總的吞吐在：160萬條/秒

 

案例6: 某車企的實時車輛網分析平臺

   

應用場景4：企業大數據中心 技術架構

   

規劃中的技術框架

   

性能基準測試

TPC-H (Analytics benchmark)

• 集羣由75個TS 和一個master構成

  • 每一個節點12 塊硬盤, 128GRAM

  • Kudu 0.5.0+Impala 2.2+CDH 5.4

  • TPC-H Scale Factor 100 (100GB)

• 分析語句舉例(6表關聯統計分析):

SELECT n_name, sum(l_extendedprice * (1 - l_discount)) as revenue FROM customer, orders, lineitem, supplier, nation, region WHERE c_custkey = o_custkey AND l_orderkey = o_orderkey AND l_suppkey = s_suppkey AND c_nationkey = s_nationkey AND s_nationkey = n_nationkey AND n_regionkey = r_regionkey AND r_name = 'ASIA' AND o_orderdate >= date '1994-01-01' AND o_orderdate < '1995-01-01' GROUP BY n_name ORDER BY revenue desc;

   

- 對內存數據,Kudu性能比Parquet高31% (幾何平均)

- 對硬盤數據,Parquet性能應該比Kudu更好(larger IO requests)

Kudu vs Phoenix

• 10 節點集羣 (9 worker, 1 master)

• HBase 1.0, Phoenix 4.3

• TPC-H LINEITEM 表(60億行記錄)

   

   

與NoSQL數據庫PK隨機查詢性能 (YCSB)

• YCSB 0.5.0-snapshot

• 10 節點集羣

(9 worker, 1 master)

• HBase 1.0

• 1億條記錄, 1千萬 ops

 

   

多用戶併發查詢下性能最好