王 巖

北京搜房網(wǎng)絡(luò)技術(shù)有限公司

基于流式計算的數(shù)據(jù)抽取系統(tǒng)設(shè)計

王 巖

北京搜房網(wǎng)絡(luò)技術(shù)有限公司

在現(xiàn)在這個互聯(lián)網(wǎng)飛速發(fā)展的時代，企業(yè)數(shù)據(jù)也成幾何級增長，如何從海量數(shù)據(jù)中提取、整理出企業(yè)想要的數(shù)據(jù)，使看似無用的數(shù)據(jù)變得有價值，數(shù)據(jù)抽取也許是開發(fā)過程中經(jīng)常遇到的問題。通過使用流式計算的思想，對原有系統(tǒng)數(shù)據(jù)源進行分類，引入消息隊列、數(shù)據(jù)緩沖池、并行消息隊列等概念，可以很好的解決此問題。

消息隊列、數(shù)據(jù)緩沖池、流式運算、生產(chǎn)者消費者

1、傳統(tǒng)數(shù)據(jù)抽取系統(tǒng)設(shè)計介紹

面對上百萬的數(shù)據(jù)需要清理、加工成新的數(shù)據(jù)，并且在清洗數(shù)據(jù)的過程中還要與其他數(shù)據(jù)庫、第三方Web服務(wù)進行交互，如果按照傳統(tǒng)的設(shè)計思路，我們經(jīng)常會采用分頁方式及順序執(zhí)行的方式進行設(shè)計。如：通過分頁方式提取數(shù)據(jù)，然后從第一頁到最后一頁數(shù)據(jù)進行與其他數(shù)據(jù)庫進行數(shù)據(jù)清理、加工，之后再把整理好的數(shù)據(jù)以分頁的方式從第一頁到最后一頁與第三方Web服務(wù)進行交互，最后在用分頁方式將所有數(shù)據(jù)存入到數(shù)據(jù)庫中。這種方式是一種開發(fā)高效的設(shè)計方式，可以在一定程度上節(jié)省開發(fā)時間和成本，但卻是一個緊繃的結(jié)構(gòu)，通常會造成程序運行時間的浪費，而且由于步驟上下游的耦合度過高，經(jīng)常會遇到上游宕機，下游無法處理等問題。雖然看似這種設(shè)計方式可以節(jié)約開發(fā)成本，但是因為整體結(jié)構(gòu)設(shè)計緊耦，會導(dǎo)致項目后期維護難度增加。

2、流式計算的數(shù)據(jù)抽取系統(tǒng)設(shè)計

流運算是計算機程序的一種數(shù)據(jù)整理、分析的方式，在流運算當(dāng)中，高級軟件的運算法則在接收流數(shù)據(jù)時就開始對其進行分析、整理。了解了流運算的概念之后，接下來我們所要做的就是對數(shù)據(jù)、步驟進行分類、業(yè)務(wù)解耦。

2.1 數(shù)據(jù)分類

基礎(chǔ)數(shù)據(jù)源，我們將被提取的數(shù)據(jù)稱之為基礎(chǔ)數(shù)據(jù)源，基礎(chǔ)數(shù)據(jù)源可以按特征進行數(shù)據(jù)分類，如：城市、日期等。注意：選擇的特征分類不易過多，且特征可以涵蓋所有數(shù)據(jù)源信息。

數(shù)據(jù)字典，是指將一些數(shù)據(jù)字典或者數(shù)據(jù)量小且不變的數(shù)據(jù)稱之為數(shù)據(jù)字典，這種數(shù)據(jù)數(shù)據(jù)量小，但在清洗、加工數(shù)據(jù)過程中起關(guān)鍵性作用，如：城市與城市簡稱對應(yīng)表、ID與相應(yīng)名稱對應(yīng)表等。

目的數(shù)據(jù)，我們將最終得到的數(shù)據(jù)稱之為目的數(shù)據(jù)，它經(jīng)過多個步驟最終存入數(shù)據(jù)庫或者形成XML服務(wù)。

第三方數(shù)據(jù)源，因為基礎(chǔ)數(shù)據(jù)源無法直接作為最終的目的數(shù)據(jù)，需要通過其他數(shù)據(jù)庫進行數(shù)據(jù)整理、加工，我們稱之為第三方數(shù)據(jù)源，這種數(shù)據(jù)源在整個數(shù)據(jù)整理的過程中，有可能會有多次并且是不同數(shù)據(jù)庫的第三方數(shù)據(jù)源。

第三方Web服務(wù)，通過基礎(chǔ)數(shù)據(jù)源的某一列的值進行Http請求獲得相應(yīng)數(shù)據(jù)，稱之為第三方Web服務(wù)。

2.2 流式運算設(shè)計

流式運算設(shè)計的主要思想是解耦和步驟的拆分，數(shù)據(jù)的分類其實是對業(yè)務(wù)的整理和解耦，并且按照不同類型的數(shù)據(jù)操作進行步驟的拆分。流式運算所采用的核心技術(shù)是消息隊列。

消息隊列中間件是大型系統(tǒng)中不可缺少的組件，它采用數(shù)據(jù)先進先出的原則，解決系統(tǒng)邏輯耦合，異步消息，流量削鋒等問題，實現(xiàn)高性能，高可用，可伸縮的架構(gòu)。常見的消息隊列中間件有：MSMQ、Kafka、RabitMQ等。消息隊列常見的模型有：點對點和發(fā)布、訂閱。消息隊列核心設(shè)計思想是生產(chǎn)者、消費者模式，在實際的軟件開發(fā)過程中，經(jīng)常會碰到如下場景：某個模塊負責(zé)產(chǎn)生數(shù)據(jù)，這些數(shù)據(jù)由另一個模塊來負責(zé)處理(此處模塊是廣義的，可以是類、函數(shù)、線程等)。產(chǎn)生數(shù)據(jù)的模塊，就形象地稱為生產(chǎn)者；而處理數(shù)據(jù)的模塊，就稱為消費者。消息隊列具有可靠緩存，可以防止數(shù)據(jù)丟失，防止數(shù)據(jù)堆積；消息隊列還具有流式模型，因為它的生產(chǎn)者、消費者模型也完全符合這種storm(流)模式；消息隊列還具有消息總線功能，它以一種總線的方式出現(xiàn)，負責(zé)消息的傳遞和分發(fā)，直接減少了模塊間的耦合程度。

對于原有系統(tǒng)的改造中，因為業(yè)務(wù)數(shù)據(jù)的多樣性和復(fù)雜性，我們需要使用多個消息隊列組合使用，將不同的數(shù)據(jù)分類固化到不同的消息隊列當(dāng)中，如：第一個消息隊列提取全部的數(shù)據(jù)字典和分頁方式提取基礎(chǔ)數(shù)據(jù)源，第二個消息隊列與第三方數(shù)據(jù)進行交互處理數(shù)據(jù)，第三個消息隊列與Web服務(wù)進行交互處理數(shù)據(jù)等等。同時，為了更好的提升每個步驟的處理效率，我們引入了數(shù)據(jù)緩沖池和并行消息隊列的概念。數(shù)據(jù)緩沖池是消息隊列與消息隊列之前通過程序運用緩存技術(shù)構(gòu)建的一個數(shù)據(jù)緩沖池，當(dāng)數(shù)據(jù)緩沖池達到一定數(shù)量的時候開啟下一個消息隊列進行數(shù)據(jù)運輸、處理，例如：如果數(shù)據(jù)緩沖池閾值設(shè)置為100，那么如果當(dāng)某個消息隊列每次處理20條數(shù)據(jù)式，這個消息隊列必須運行5次后，使得數(shù)據(jù)緩沖池大小達到100以后，才能開啟它后面的消息隊列。并行消息隊列的概念是指每個消息隊列可以橫向擴充，例如：當(dāng)數(shù)據(jù)緩沖池中有100條數(shù)據(jù)時，每條消息隊列處理20條數(shù)據(jù)，為了提高運行速度，可以通過代碼增加了5條消息隊列，那么當(dāng)程序運行時候5條消息隊列并行處理，這5條消息隊列稱之為并行消息隊列。注意，并行消息隊列采用多線程開發(fā)的思想，并行消息隊列越多會對這個消息隊列的第三方數(shù)據(jù)庫或者第三方Web服務(wù)造成負載壓力，配置不合理反而會造成性能問題。

3 流式運算架構(gòu)設(shè)計（見圖1）

在第一個消息隊列啟動之前，數(shù)據(jù)字典通過全量的方式，加載到程序緩存當(dāng)中。因為基礎(chǔ)數(shù)據(jù)源通過不同特征(如：城市)進行了分類，程序通過多線程的方式，開啟不同線程執(zhí)行各自的程序，即北京、上海、天津等城市并行執(zhí)行后續(xù)消息隊列操作，以下舉例為分類為北京的數(shù)據(jù)，第一個消息隊列(Queue1)開始以分頁的方式對基礎(chǔ)數(shù)據(jù)源進行抽取。當(dāng)?shù)谝粋€數(shù)據(jù)緩沖池滿了以后，開啟第二個消息隊列(Queue2)，程序與第三方數(shù)據(jù)源進行交互并對數(shù)據(jù)進行加工。當(dāng)?shù)诙€數(shù)據(jù)緩沖池滿了以后開啟第三個消息隊列(Queue3)，并通過程序請求第三方Web服務(wù)對數(shù)據(jù)進行加工。最后當(dāng)?shù)谌齻€數(shù)據(jù)緩沖池滿了以后，開啟第四個消息隊列，第四個隊列(Queue4)對最后形成目的數(shù)據(jù)，并入庫或者形成xml服務(wù)。如果過程中還有其它第三方數(shù)據(jù)源或者第三方Web服務(wù)，可以縱向增加消息隊列(QueueN)即增加了數(shù)據(jù)處理步驟。如果當(dāng)某個消息隊列時間消耗過長或者阻塞時候，可以通過增加并行消息隊列，提高并發(fā)，例如：Queue31、Queue32。在流式運算數(shù)據(jù)抽取系統(tǒng)中，需要在程序中增加運行日志，運行日志主要包括每個消息隊列的開始時間和結(jié)束時間，每個步驟中出錯日志，以及處理多少條過濾多少條，因為流式運算系統(tǒng)，運行日志起到對整體程序健康度的監(jiān)測，通過監(jiān)測每個消息隊列的運行日志中的運行時間，可以在耗時較長的消息隊列增加并行消息隊列來提高整體的運算速度。

圖1

這種流式運算數(shù)據(jù)抽取系統(tǒng)的設(shè)計，實現(xiàn)程序的縱向拓展和橫向拓展，通過增加消息隊列實現(xiàn)縱向拓展，使數(shù)據(jù)處理之間解耦。通過增加并行消息隊列實現(xiàn)橫向拓展，提高單步并發(fā)。在實際應(yīng)用中這種設(shè)計方式，可以大大提高程序每分鐘處理量，降低程序之間的耦合性，并且不會出現(xiàn)某一部分數(shù)據(jù)出現(xiàn)問題導(dǎo)致整個程序無法產(chǎn)出目的數(shù)據(jù)。在開發(fā)方面，流式運算方式將步驟解耦，搭建好整體框架以后可以很好的分配相關(guān)人員開發(fā)相關(guān)業(yè)務(wù)，大大提高了開發(fā)質(zhì)量和開發(fā)效率。

4、結(jié)束語

流式運算是處理大數(shù)據(jù)手段之一，通過多個消息隊列可以搭建一個強壯的流式運算框架，再根據(jù)需求開發(fā)不同業(yè)務(wù)，就形成了流式計算的數(shù)據(jù)抽取系統(tǒng)，這種系統(tǒng)屬于后臺系統(tǒng)，多與系統(tǒng)中的計劃任務(wù)結(jié)合使用，通過計劃任務(wù)定時定點啟動流式計算數(shù)據(jù)抽取系統(tǒng)，實現(xiàn)數(shù)據(jù)的篩選整理，入庫或產(chǎn)出相關(guān)的數(shù)據(jù)服務(wù)，為后續(xù)業(yè)務(wù)系統(tǒng)提供基礎(chǔ)數(shù)據(jù)源進行分析展示。