本文将主要介紹如何利用lambda架構來跟蹤資料實時更新的項目實作，以一個新聞服務功能為例。
目前股票市場的交易者可以了解豐富的股票交易資訊。從金融新聞到傳統的報紙和雜志再到部落格和社交媒體，彙聚着海量的資料，遠比股票交易者想關注的股 票資訊要大得多，這就需要為股票交易者提供資訊的有效過濾。這裡将開發一個新聞服務功能給股票證券投資交易者使用，并為股票交易者提供個性化新聞。
這個新聞服務就叫"自動擷取金融新聞"，輸入各個資料源的金融新聞，也同時輸入使用者實時股票交易資訊。不管何時，在股票交易者所擁有資産證券中占比 較大的公司，它們的新聞一到達，将會顯示到股票交易者的儀表闆上。随着大量股票交易者進行交易，相應的交易資訊會發送過來，是以希望擁有一個大資料系統來 存儲所有交易者的曆史交易資訊作為真實資料源，然而，處理海量資料會非常慢以至于不能進行實時的資料更新。為了達到實時跟蹤和維持資料結果為最新這兩個要求，可以采用lambda架構來實作。
lambda架構優勢
在傳統sql系統，更新一個表隻是對已存在字段的值進行更改，這在少量的伺服器上的資料庫工作的很好，可以水準擴充到從庫或者備份庫。但是當資料庫 擴充到大量資料伺服器上時，硬體崩潰等情況下恢複資料到失敗點就比較困難和耗時，而且由于曆史不在資料庫中，僅僅存在log日志，資料崩潰将導緻一些不可見的資料錯誤，即髒資料。
而相對應地，一個分布式、多副本消息隊列的大資料系統可以保證資料一旦進入系統就不會丢失，即使在硬體或者網絡失敗的情況下。存儲更新的所有曆史可 以重建真實的資料源，并能保證每次批處理之後結果正确，然而，為了在實時資料更新後得到最新完整的資料集，需要重新處理整個曆史資料集，将會耗費太長的時 間。為了解決這個問題，可以在lambda架構中增加一個實時元件，此元件隻存儲資料更新的目前值，可以保證快速實時得到結果，工作過程類似于傳統的 sql系統。實時處理層的髒資料将會被後續批處理覆寫掉，這個高可用、最終一緻性的系統可以實作準确的結果。目前值的任何錯誤，實時處理層的報告，硬體或 者網絡錯誤，資料崩潰，或者軟體bug等将會在下一次批處理時自動修複。
自動擷取金融新聞項目的資料管道
整個資料管道流動如圖1：
圖1
輸入資料格式為json，主要來自綜合交易資訊和twitter新聞。json格式的消息會push到kafka，并被批處理層(batch layer)和實時處理層(real-time layer)消費。使用kafka作為資料管道的輸入起點，是因為kafka可以保證即使在硬體或者網絡失敗的情況下，消息也會被傳輸到整個系統。
在批處理層，camus(linkin開源的項目，現已更名為gobblin)消費所有kafka過來的消息并儲存到hdfs上，然後spark處理所有的交易曆史計算每個股票交易者持有的股票準确數量，對應的結果會寫入cassandra資料庫。
在流式處理層，spark streaming實時消費kafka消息，但并不像storm那樣完全實時，spark streaming可以達到500ms的micro-batch資料流處理。spark streaming可以重用批處理層的spark代碼，并且micro-batch資料流處理可以得到足夠小的延遲。
批處理層和實時處理層的結果都會寫入到cassandra資料庫，并通過flask提供一個web接口服務。随着海量交易資料寫入系統，cassandra資料庫的快速寫入能力基本可以滿足。
如何排程實時處理層和批處理層的結果
當最新的消息進入大資料系統，web接口提供的結果服務總能保持最新，綜合批處理層和實時層的處理結果。用一個例子來展示如何簡單的使用批處理結果和實時處理結果。
從下圖2看到，有三個資料庫表：一個存儲批處理結果(圖2中batch表);一個存儲自上次批處理完成時間點到目前時間的實時交易資料，即增量資料(圖2中real time 2表);另外一個存儲最新資料，即狀态表(圖2中高亮的real time 1表)。
任何軟體、硬體或者網絡問題引起批處理結果異常，都通過單獨一個資料庫表記錄資料增量，并在批處理成功後更新為對應的批處理結果數來保證最終資料一緻性。
在這個例子中，假設第一輪批處理起始時間點為t0，一個交易者做了一筆交易後獲得了3m公司的5000股股票。
圖2
在t0時間點，批處理開始，處理完之後最新結果存儲在real time 1表，目前值為5000股。
圖3
在批處理過程中，交易者賣掉3m公司1000股股票，real time 1表更新資料值為4000股，同時real time 2表存儲從t0到目前的增量-1000股，如圖4所示。
圖4
當批處理結束，三個表的值分别為5000，4000，-1000。這時，交換active資料庫表為real time 2表，進行合并批處理結果和實時結果獲得最新結果值。然後重置real time 1表為0，後續用來存儲從t1時間點開始的增量資料。接下來新的一輪以存儲最新資料的real time 2表為起點，循環前面的過程。
圖5
圖6
圖7
以上每步處理過程完全成功并寫入資料庫，可以保證展示給交易者的資料準确性。資料集 處理時間取決于資料集大小，處理任務的計劃按序處理而不是按自然天時間。在一個系統中需要工作流支援複雜處理、多任務依賴和資源共享。這裡采用 airbnb的項目airflow，可以排程程式和監控工作流。airflow把task和上遊各種依賴建構成一個有向無環圖(dag)，基于 python實作，可以把多個任務寫成bash腳本，bash指令能直接調用任何子產品，并且bash腳本可以被airflow使用，這樣使得 airflow易操作。airflow程式設計接口比基于xml配置的排程系統oozie簡單;airflow的bash腳本編碼量比luigi要少很多，luigi的每個job都是一個python工程。每步合并實時和批量資料的job運作都是前一步成功完成退出後。
最後簡單總結一下，lambda架構涉及批量處理層和實時處理層處理曆史資料以及實時更新的資料。 為了lambda架構的實作切實可行，資料處理要設計成批處理層和實時處理層結合。本項目中，有一個“備用”資料庫表專門用來存儲輸入的總數，而不從批處 理層讀取資料，并允許對批處理層和實時處理層的結果進行簡單的聚合。以上就是用lambda架構實作的一個高可用、高資料最終一緻性的系統。
本文作者：俠天
來源：51cto