本文將主要介紹如何利用Lambda架構(gòu)來跟蹤數(shù)據(jù)實(shí)時(shí)更新的項(xiàng)目實(shí)現(xiàn)，以一個(gè)新聞服務(wù)功能為例。

當(dāng)前股票市場的交易者可以了解豐富的股票交易信息。從金融新聞到傳統(tǒng)的報(bào)紙和雜志再到博客和社交媒體，匯聚著海量的數(shù)據(jù)，遠(yuǎn)比股票交易者想關(guān)注的股 票信息要大得多，這就需要為股票交易者提供信息的有效過濾。這里將開發(fā)一個(gè)新聞服務(wù)功能給股票證券投資交易者使用，并為股票交易者提供個(gè)性化新聞。

這個(gè)新聞服務(wù)就叫"自動(dòng)獲取金融新聞"，輸入各個(gè)數(shù)據(jù)源的金融新聞，也同時(shí)輸入用戶實(shí)時(shí)股票交易信息。不管何時(shí)，在股票交易者所擁有資產(chǎn)證券中占比 較大的公司，它們的新聞一到達(dá)，將會(huì)顯示到股票交易者的儀表板上。隨著大量股票交易者進(jìn)行交易，相應(yīng)的交易信息會(huì)發(fā)送過來，所以希望擁有一個(gè)大數(shù)據(jù)系統(tǒng)來 存儲(chǔ)所有交易者的歷史交易信息作為真實(shí)數(shù)據(jù)源，然而，處理海量數(shù)據(jù)會(huì)非常慢以至于不能進(jìn)行實(shí)時(shí)的數(shù)據(jù)更新。為了達(dá)到實(shí)時(shí)跟蹤和維持?jǐn)?shù)據(jù)結(jié)果為***這兩個(gè)要求，可以采用Lambda架構(gòu)來實(shí)現(xiàn)。

Lambda架構(gòu)優(yōu)勢

在傳統(tǒng)SQL系統(tǒng)，更新一個(gè)表只是對已存在字段的值進(jìn)行更改，這在少量的服務(wù)器上的數(shù)據(jù)庫工作的很好，可以水平擴(kuò)展到從庫或者備份庫。但是當(dāng)數(shù)據(jù)庫 擴(kuò)展到大量數(shù)據(jù)服務(wù)器上時(shí)，硬件崩潰等情況下恢復(fù)數(shù)據(jù)到失敗點(diǎn)就比較困難和耗時(shí)，而且由于歷史不在數(shù)據(jù)庫中，僅僅存在log日志，數(shù)據(jù)崩潰將導(dǎo)致一些不可見的數(shù)據(jù)錯(cuò)誤，即臟數(shù)據(jù)。

而相對應(yīng)地，一個(gè)分布式、多副本消息隊(duì)列的大數(shù)據(jù)系統(tǒng)可以保證數(shù)據(jù)一旦進(jìn)入系統(tǒng)就不會(huì)丟失，即使在硬件或者網(wǎng)絡(luò)失敗的情況下。存儲(chǔ)更新的所有歷史可 以重建真實(shí)的數(shù)據(jù)源，并能保證每次批處理之后結(jié)果正確，然而，為了在實(shí)時(shí)數(shù)據(jù)更新后得到***完整的數(shù)據(jù)集，需要重新處理整個(gè)歷史數(shù)據(jù)集，將會(huì)耗費(fèi)太長的時(shí) 間。為了解決這個(gè)問題，可以在Lambda架構(gòu)中增加一個(gè)實(shí)時(shí)組件，此組件只存儲(chǔ)數(shù)據(jù)更新的當(dāng)前值，可以保證快速實(shí)時(shí)得到結(jié)果，工作過程類似于傳統(tǒng)的 SQL系統(tǒng)。實(shí)時(shí)處理層的臟數(shù)據(jù)將會(huì)被后續(xù)批處理覆蓋掉，這個(gè)高可用、最終一致性的系統(tǒng)可以實(shí)現(xiàn)準(zhǔn)確的結(jié)果。當(dāng)前值的任何錯(cuò)誤，實(shí)時(shí)處理層的報(bào)告，硬件或 者網(wǎng)絡(luò)錯(cuò)誤，數(shù)據(jù)崩潰，或者軟件Bug等將會(huì)在下一次批處理時(shí)自動(dòng)修復(fù)。

自動(dòng)獲取金融新聞項(xiàng)目的數(shù)據(jù)管道

整個(gè)數(shù)據(jù)管道流動(dòng)如圖1：

圖1

輸入數(shù)據(jù)格式為JSON，主要來自綜合交易信息和Twitter新聞。JSON格式的消息會(huì)push到Kafka，并被批處理層(batch layer)和實(shí)時(shí)處理層(real-time layer)消費(fèi)。使用Kafka作為數(shù)據(jù)管道的輸入起點(diǎn)，是因?yàn)镵afka可以保證即使在硬件或者網(wǎng)絡(luò)失敗的情況下，消息也會(huì)被傳輸?shù)秸麄€(gè)系統(tǒng)。

在批處理層，Camus(Linkin開源的項(xiàng)目，現(xiàn)已更名為Gobblin)消費(fèi)所有Kafka過來的消息并保存到HDFS上，然后Spark處理所有的交易歷史計(jì)算每個(gè)股票交易者持有的股票準(zhǔn)確數(shù)量，對應(yīng)的結(jié)果會(huì)寫入Cassandra數(shù)據(jù)庫。

在流式處理層，Spark Streaming實(shí)時(shí)消費(fèi)Kafka消息，但并不像Storm那樣完全實(shí)時(shí)，Spark Streaming可以達(dá)到500ms的micro-batch數(shù)據(jù)流處理。Spark Streaming可以重用批處理層的Spark代碼，并且micro-batch數(shù)據(jù)流處理可以得到足夠小的延遲。

批處理層和實(shí)時(shí)處理層的結(jié)果都會(huì)寫入到Cassandra數(shù)據(jù)庫，并通過Flask提供一個(gè)web接口服務(wù)。隨著海量交易數(shù)據(jù)寫入系統(tǒng)，Cassandra數(shù)據(jù)庫的快速寫入能力基本可以滿足。

如何調(diào)度實(shí)時(shí)處理層和批處理層的結(jié)果

當(dāng)***的消息進(jìn)入大數(shù)據(jù)系統(tǒng)，web接口提供的結(jié)果服務(wù)總能保持***，綜合批處理層和實(shí)時(shí)層的處理結(jié)果。用一個(gè)例子來展示如何簡單的使用批處理結(jié)果和實(shí)時(shí)處理結(jié)果。

從下圖2看到，有三個(gè)數(shù)據(jù)庫表：一個(gè)存儲(chǔ)批處理結(jié)果(圖2中Batch表);一個(gè)存儲(chǔ)自上次批處理完成時(shí)間點(diǎn)到當(dāng)前時(shí)間的實(shí)時(shí)交易數(shù)據(jù)，即增量數(shù)據(jù)(圖2中Real Time 2表);另外一個(gè)存儲(chǔ)***數(shù)據(jù)，即狀態(tài)表(圖2中高亮的Real Time 1表)。

任何軟件、硬件或者網(wǎng)絡(luò)問題引起批處理結(jié)果異常，都通過單獨(dú)一個(gè)數(shù)據(jù)庫表記錄數(shù)據(jù)增量，并在批處理成功后更新為對應(yīng)的批處理結(jié)果數(shù)來保證最終數(shù)據(jù)一致性。

在這個(gè)例子中，假設(shè)***輪批處理起始時(shí)間點(diǎn)為t0，一個(gè)交易者做了一筆交易后獲得了3M公司的5000股股票。

圖2

在t0時(shí)間點(diǎn)，批處理開始，處理完之后***結(jié)果存儲(chǔ)在Real Time 1表，當(dāng)前值為5000股。

圖3

在批處理過程中，交易者賣掉3M公司1000股股票，Real Time 1表更新數(shù)據(jù)值為4000股，同時(shí)Real Time 2表存儲(chǔ)從t0到當(dāng)前的增量-1000股，如圖4所示。

圖4

當(dāng)批處理結(jié)束，三個(gè)表的值分別為5000，4000，-1000。這時(shí)，交換active數(shù)據(jù)庫表為Real Time 2表，進(jìn)行合并批處理結(jié)果和實(shí)時(shí)結(jié)果獲得***結(jié)果值。然后重置Real Time 1表為0，后續(xù)用來存儲(chǔ)從t1時(shí)間點(diǎn)開始的增量數(shù)據(jù)。接下來新的一輪以存儲(chǔ)***數(shù)據(jù)的Real Time 2表為起點(diǎn)，循環(huán)前面的過程。

圖5

圖6

圖7

以上每步處理過程完全成功并寫入數(shù)據(jù)庫，可以保證展示給交易者的數(shù)據(jù)準(zhǔn)確性。數(shù)據(jù)集 處理時(shí)間取決于數(shù)據(jù)集大小，處理任務(wù)的計(jì)劃按序處理而不是按自然天時(shí)間。在一個(gè)系統(tǒng)中需要工作流支持復(fù)雜處理、多任務(wù)依賴和資源共享。這里采用 Airbnb的項(xiàng)目Airflow，可以調(diào)度程序和監(jiān)控工作流。Airflow把task和上游各種依賴構(gòu)建成一個(gè)有向無環(huán)圖(DAG)，基于 Python實(shí)現(xiàn)，可以把多個(gè)任務(wù)寫成Bash腳本，Bash命令能直接調(diào)用任何模塊，并且Bash腳本可以被Airflow使用，這樣使得 Airflow易操作。Airflow編程接口比基于XML配置的調(diào)度系統(tǒng)Oozie簡單;Airflow的Bash腳本編碼量比Luigi要少很多，Luigi的每個(gè)job都是一個(gè)python工程。每步合并實(shí)時(shí)和批量數(shù)據(jù)的job運(yùn)行都是前一步成功完成退出后。

***簡單總結(jié)一下，Lambda架構(gòu)涉及批量處理層和實(shí)時(shí)處理層處理歷史數(shù)據(jù)以及實(shí)時(shí)更新的數(shù)據(jù)。 為了Lambda架構(gòu)的實(shí)現(xiàn)切實(shí)可行，數(shù)據(jù)處理要設(shè)計(jì)成批處理層和實(shí)時(shí)處理層結(jié)合。本項(xiàng)目中，有一個(gè)“備用”數(shù)據(jù)庫表專門用來存儲(chǔ)輸入的總數(shù)，而不從批處 理層讀取數(shù)據(jù)，并允許對批處理層和實(shí)時(shí)處理層的結(jié)果進(jìn)行簡單的聚合。以上就是用Lambda架構(gòu)實(shí)現(xiàn)的一個(gè)高可用、高數(shù)據(jù)最終一致性的系統(tǒng)。