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

近實時分析的場景

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

• 分析現實世界中正在發生的事件的能力，結合歷史數據和實時流數據進行匯總分析、預測和明細查詢
• 絕對實時和批量不可調和，"近實時" 的意思是這是人機交互中能感受的尺度（秒級），而不是機器自動處理的實時性量級（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

 

   

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