王 巖

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

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

王 巖

北京搜房網(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)常遇到的問題。通過使用流式計(jì)算的思想，對原有系統(tǒng)數(shù)據(jù)源進(jìn)行分類，引入消息隊(duì)列、數(shù)據(jù)緩沖池、并行消息隊(duì)列等概念，可以很好的解決此問題。

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

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

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

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

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

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

基礎(chǔ)數(shù)據(jù)源，我們將被提取的數(shù)據(jù)稱之為基礎(chǔ)數(shù)據(jù)源，基礎(chǔ)數(shù)據(jù)源可以按特征進(jìn)行數(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ù)源，因?yàn)榛A(chǔ)數(shù)據(jù)源無法直接作為最終的目的數(shù)據(jù)，需要通過其他數(shù)據(jù)庫進(jìn)行數(shù)據(jù)整理、加工，我們稱之為第三方數(shù)據(jù)源，這種數(shù)據(jù)源在整個數(shù)據(jù)整理的過程中，有可能會有多次并且是不同數(shù)據(jù)庫的第三方數(shù)據(jù)源。

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

2.2 流式運(yùn)算設(shè)計(jì)

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

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

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

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

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

圖1

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

4、結(jié)束語

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