前言

近年來，隨著社會服務(wù)信息化的高速發(fā)展，在互聯(lián)網(wǎng)、物聯(lián)網(wǎng)、金融、物流、電磁等各方面數(shù)據(jù)都呈現(xiàn)指數(shù)級的增長。大數(shù)據(jù)的傳輸是大數(shù)據(jù)處理基本流程的重要一環(huán)，高性能的數(shù)據(jù)傳輸可以為后續(xù)數(shù)據(jù)分析特別是實(shí)時(shí)分析提供保障。本文簡要介紹了主流的大數(shù)據(jù)傳輸方法和多源異構(gòu)數(shù)據(jù)傳輸?shù)脑O(shè)計(jì)方案，為大家提供參考。

[[329866]]

1、大數(shù)據(jù)傳輸相關(guān)背景

2003年起，Google公司相繼發(fā)表了Google FS、MapReduce、BigTable等3個(gè)系統(tǒng)(框架)的論文，說明了這3個(gè)產(chǎn)品的詳細(xì)設(shè)計(jì)方法，為后來全球的大數(shù)據(jù)發(fā)展奠定了基礎(chǔ)。由于數(shù)據(jù)量和效率的問題，傳統(tǒng)的單機(jī)存儲與計(jì)算已經(jīng)不適應(yīng)時(shí)代的發(fā)展，多節(jié)點(diǎn)的分布式存儲逐漸取而代之，這種方法可以在多個(gè)廉價(jià)的節(jié)點(diǎn)上同時(shí)存儲和并行計(jì)算，并且提供了很好的容錯(cuò)能力。

隨著大數(shù)據(jù)技術(shù)的不斷發(fā)展，更多高性能的處理框架走上了歷史舞臺，形成了大數(shù)據(jù)生態(tài)系統(tǒng)。例如分布式存儲有HDFS、Hbase、hive等，分布式計(jì)算有MapReduce、Spark、Storm等，而作為該生態(tài)系統(tǒng)的重要組成部分，數(shù)據(jù)傳輸模塊必不可少，現(xiàn)在比較流行的有Kafka、Logstash、Sqoop等。

在數(shù)據(jù)傳輸?shù)倪^程中，不論是類似將文件導(dǎo)入數(shù)據(jù)庫的離線數(shù)據(jù)傳輸，還是類似實(shí)時(shí)采集數(shù)據(jù)傳輸?shù)綌?shù)據(jù)庫進(jìn)行計(jì)算的實(shí)時(shí)傳輸，我們都希望具有高速優(yōu)質(zhì)的傳輸效率，同時(shí)，還要求數(shù)據(jù)傳輸達(dá)到良好的安全性、穩(wěn)定性、可靠性。另一方面，對于實(shí)時(shí)性要求比較高的，例如金融股票、數(shù)據(jù)可視化等方面需要獲得快速的響應(yīng)，而對于傳入數(shù)據(jù)倉庫保存的可以有一定延遲。

基于最基本的用戶需求，大數(shù)據(jù)傳輸機(jī)制應(yīng)當(dāng)遵循以下原則：

(1)模型安全性。大數(shù)據(jù)計(jì)算一般是由幾十個(gè)甚至上百個(gè)節(jié)點(diǎn)組成的，在獲取數(shù)據(jù)的時(shí)候，節(jié)點(diǎn)與數(shù)據(jù)源之間，節(jié)點(diǎn)與節(jié)點(diǎn)之間，都會有占有較大的I/O使用率，數(shù)據(jù)傳輸之間必須滿足必要的安全性。對于保密要求較高的數(shù)據(jù)，更要建立全面的數(shù)據(jù)保護(hù)措施，以防數(shù)據(jù)泄露。

(2)傳輸可靠性。隨著計(jì)算存儲設(shè)備和數(shù)據(jù)傳輸通道的不斷升級，數(shù)據(jù)的傳輸速度和效率逐漸提高。在獲取數(shù)據(jù)源的時(shí)候，數(shù)據(jù)管道必須提供一個(gè)可靠的傳輸，以達(dá)到至少交付一次的保證。

(3)網(wǎng)絡(luò)自適應(yīng)性。用戶和分析設(shè)備可以根據(jù)自身的需求，適應(yīng)數(shù)據(jù)傳輸?shù)姆?wù)，最大化對接數(shù)據(jù)格式，達(dá)到良好的對接效果。

2、主流傳輸方法

目前在大數(shù)據(jù)的廣泛應(yīng)用中，Kafka、Logstash、Sqoop等都是傳輸數(shù)據(jù)的重要途徑，這里簡要介紹傳輸原理。

2.1Kafka

Kafka最初由Linkedin公司開發(fā)，是一個(gè)分布式、分區(qū)的、多副本的、多訂閱者，基于zookeeper協(xié)調(diào)的分布式日志系統(tǒng)，常見可以用于web/nginx日志、訪問日志，消息服務(wù)等等，Linkedin于2010年將該系統(tǒng)貢獻(xiàn)給了Apache基金會并成為頂級開源項(xiàng)目。

Kafka主要設(shè)計(jì)特點(diǎn)如下：

• 以時(shí)間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時(shí)間的訪問性能。
• 高吞吐率。即使在非常廉價(jià)的商用機(jī)器上也能做到單機(jī)支持每秒100K條消息的傳輸。
• 支持Kafka Server間的消息分區(qū)，及分布式消費(fèi)，同時(shí)保證每個(gè)part內(nèi)的消息順序傳輸。
• 同時(shí)支持離線數(shù)據(jù)處理和實(shí)時(shí)數(shù)據(jù)處理。
• Scale out:支持在線水平擴(kuò)展。

圖1 kafka的架構(gòu)

圖1展示了一個(gè)典型的kafka集群的架構(gòu)，每個(gè)集群中都包含若干個(gè)生產(chǎn)者(producer)，這些生產(chǎn)者可以是來自數(shù)據(jù)采集設(shè)備的硬件數(shù)據(jù)源，亦可以是服務(wù)器產(chǎn)生的日志信息等等;每個(gè)集群中都有若干的服務(wù)代理(broker)，每個(gè)服務(wù)代理一般安裝在一個(gè)節(jié)點(diǎn)服務(wù)器上，kafka支持平行擴(kuò)展，集群中服務(wù)代理的數(shù)量越多，吞吐量也會越高。生產(chǎn)者生產(chǎn)的數(shù)據(jù)可以向一個(gè)指定的topic中寫入，消費(fèi)者可以根據(jù)自己的需求，向指定的topic中拉取數(shù)據(jù)。

為了進(jìn)一步提高數(shù)據(jù)傳輸?shù)耐掏侣?，kafka將每個(gè)topic分為若干個(gè)part，每個(gè)part下面都會存儲對應(yīng)的數(shù)據(jù)和索引文件。當(dāng)創(chuàng)建topic時(shí)，可以指定part的數(shù)量，part數(shù)量越多，系統(tǒng)的吞吐量就會越大，但是也會占用更多的資源。kafka收到生產(chǎn)者發(fā)送的數(shù)據(jù)后，就跟根據(jù)一定的均衡策略，將數(shù)據(jù)存放到某一個(gè)part下，等待消費(fèi)者來消費(fèi)數(shù)據(jù)。

除此之外，kafka還為數(shù)據(jù)建立了副本，當(dāng)數(shù)據(jù)節(jié)點(diǎn)發(fā)生意外時(shí)，其他的副本通過一定的機(jī)制擔(dān)起主part的作用，從而使系統(tǒng)具有高可用性。kafka提供了至少一次的交付保證，生產(chǎn)者發(fā)送數(shù)據(jù)到節(jié)點(diǎn)，節(jié)點(diǎn)會反饋該消息是否存儲，若未收到確認(rèn)信息，生產(chǎn)者則會重復(fù)發(fā)送該信息;同樣的，消費(fèi)者消費(fèi)數(shù)據(jù)發(fā)送收到的反饋，節(jié)點(diǎn)記錄被消費(fèi)的位置，下次消費(fèi)則從該位置開始。這些機(jī)制都保證了至少一次的可靠交付。

在安全性方面，kafka使用了SSL或者SASL驗(yàn)證來自客戶端(生產(chǎn)者和消費(fèi)者)以及其他broker和工具到broker的鏈接身份，在傳輸?shù)倪^程中也可以選擇對數(shù)據(jù)進(jìn)行加密，對客戶端的讀寫授權(quán)，雖然可能會導(dǎo)致集群性能下降，但對于保密性較高的數(shù)據(jù)來說，是可以接受的。

2.2Logstash

Logstash 是免費(fèi)且開放的服務(wù)器端數(shù)據(jù)處理管道，能夠從多個(gè)來源采集數(shù)據(jù)，與此同時(shí)這根管道還可以讓你根據(jù)自己的需求在中間加上濾網(wǎng)轉(zhuǎn)換過濾數(shù)據(jù)，然后將數(shù)據(jù)發(fā)送到用戶指定的數(shù)據(jù)庫中。

圖2 Logstash數(shù)據(jù)傳輸

圖3 Logstash 結(jié)構(gòu)

Logstash將數(shù)據(jù)流中每一條數(shù)據(jù)稱之為一個(gè)event，處理流水線有三個(gè)主要角色完成：inputs –> filters –> outputs，原始數(shù)據(jù)進(jìn)入logstash后在內(nèi)部流轉(zhuǎn)并不是以原始數(shù)據(jù)的形式流轉(zhuǎn)，在input處被轉(zhuǎn)換為event，在output event處被轉(zhuǎn)換為目標(biāo)格式的數(shù)據(jù)。

當(dāng)有一個(gè)輸入數(shù)據(jù)時(shí)，input會從文件中取出數(shù)據(jù)，然后通過json codec將數(shù)據(jù)轉(zhuǎn)換成logstash event。這條event會通過queue流入某一條pipline處理線程中，首先會存放在batcher中。當(dāng)batcher達(dá)到處理數(shù)據(jù)的條件(如一定時(shí)間或event一定規(guī)模)后，batcher會把數(shù)據(jù)發(fā)送到filter中，filter對event數(shù)據(jù)進(jìn)行處理后轉(zhuǎn)到output，output就把數(shù)據(jù)輸出到指定的輸出位置。輸出后還會返回ACK給queue，包含已經(jīng)處理的event，queue會將已處理的event進(jìn)行標(biāo)記。

假如 Logstash 節(jié)點(diǎn)發(fā)生故障，Logstash 會通過持久化隊(duì)列來保證至少將運(yùn)行中的事件送達(dá)一次。那些未被正常處理的消息會被送往死信隊(duì)列 (dead letter queue) 以便做進(jìn)一步處理。由于具備了這種吸收吞吐量的能力，無需采用額外的隊(duì)列層，Logstash 就能平穩(wěn)度過高峰期。此外，還能充分確保采集管道的安全性。

3、多源異構(gòu)數(shù)據(jù)傳輸設(shè)計(jì)

在數(shù)據(jù)不斷壯大的過程中，我們往往會根據(jù)自身的需求，收集不同類型的數(shù)據(jù)，存儲在不同的數(shù)據(jù)庫中，使用數(shù)據(jù)時(shí)也會從不同的數(shù)據(jù)源讀取數(shù)據(jù)進(jìn)行分析和處理。這些不同的存儲方式、不同的采集的系統(tǒng)、不同的數(shù)據(jù)格式，從簡單的文件數(shù)據(jù)庫到復(fù)雜的網(wǎng)絡(luò)數(shù)據(jù)庫，共同構(gòu)成了異構(gòu)數(shù)據(jù)源。為了將數(shù)據(jù)統(tǒng)一處理，根據(jù)可視化等現(xiàn)實(shí)需求，就需要將各個(gè)異構(gòu)數(shù)據(jù)源通過一個(gè)引擎銜接起來，為數(shù)據(jù)的大批量處理和展示提供更為標(biāo)準(zhǔn)化的讀取方式。

目前，以異構(gòu)數(shù)據(jù)批處理為目標(biāo)的應(yīng)用有springbatch、kettle、datax等，他們各自有各自的特點(diǎn)：

Springbatch是spring提供的一個(gè)輕量級、全面的批處理數(shù)據(jù)處理框架，無需用戶交互即可最有效地處理大量信息的自動化，復(fù)雜處理，并且提供了可重用的功能，這些功能對于處理大量的數(shù)據(jù)至關(guān)重要。

Kettle是一款國外開源的ETL工具，他可以通過Spoon來允許你運(yùn)行或者轉(zhuǎn)換任務(wù)，支持從不同的數(shù)據(jù)源讀取、操作和寫入數(shù)據(jù)，在規(guī)定的時(shí)間間隔內(nèi)用批處理的模式自動運(yùn)行。

Datax一個(gè)異構(gòu)數(shù)據(jù)源離線同步工具，致力于實(shí)現(xiàn)包括關(guān)系型數(shù)據(jù)庫(MySQL、Oracle等)、HDFS、hive、ODPS、HBase、FTP等各種異構(gòu)數(shù)據(jù)源之間穩(wěn)定高效的數(shù)據(jù)同步功能。

下面介紹一種輕量級的ETL工具，主要作用就是從不同源獲取數(shù)據(jù)，然后做統(tǒng)一的處理，最后再寫入各種目標(biāo)源。它基本特性是：

基于Springboot開發(fā)，輕量級別、快速、簡單，入門門檻低

擴(kuò)展性強(qiáng)，各個(gè)模塊均是獨(dú)立的，可以以插件的形式進(jìn)行開發(fā)

可以通過UI界面來構(gòu)建任務(wù)并操作，總體監(jiān)控平臺的數(shù)據(jù)實(shí)時(shí)情況

基于Disruptor做緩沖，同時(shí)使用redis等內(nèi)存緩存，保證高速處理任務(wù)

該ETL工具將整個(gè)系統(tǒng)分為如下模塊：Input、Reader、Transport、Convert、Writer和Output，在系統(tǒng)上層已經(jīng)定義好各個(gè)模塊的接口，開發(fā)者根據(jù)自己的需要個(gè)性化定義自己的模塊，只需繼承上層接口即可實(shí)現(xiàn)模塊的嵌入。系統(tǒng)運(yùn)行的簡化基本流程如圖4所示。

圖4 ETL工具運(yùn)行簡化流程

這里所有的模塊都有一定的標(biāo)準(zhǔn)來接入系統(tǒng)，然后使用各數(shù)據(jù)源提供的API來讀寫數(shù)據(jù)，例如輸入可以從文件讀取、mysql、hbase、hdfs、kafka、http等，輸出同樣支持這些數(shù)據(jù)源，最終解決異構(gòu)數(shù)據(jù)源相互傳輸數(shù)據(jù)不兼容的問題。

系統(tǒng)在應(yīng)對緩沖和讀寫速度上均設(shè)置可選的策略，可以基于java的調(diào)度器，綜合當(dāng)前輸入輸出的任務(wù)數(shù)量，來調(diào)整輸入輸出線程池以及線程的數(shù)量，以使數(shù)據(jù)的傳輸達(dá)到最大的性能。

4、總 結(jié)

現(xiàn)在數(shù)據(jù)采集的設(shè)備無處不在，在各種格式的數(shù)據(jù)匯入不同數(shù)據(jù)倉庫、數(shù)據(jù)倉庫之間互相接入數(shù)據(jù)都需要一個(gè)高效、可靠、安全的數(shù)據(jù)通道，本文介紹了大數(shù)據(jù)傳輸?shù)囊恍┍尘爸R，同時(shí)簡要描述了當(dāng)前主流數(shù)據(jù)傳輸工具的應(yīng)用和個(gè)性化異構(gòu)數(shù)據(jù)引擎的設(shè)計(jì)問題。本文參考了一些文獻(xiàn)和網(wǎng)絡(luò)資源，對他們的觀點(diǎn)和技術(shù)對本文的貢獻(xiàn)表示感謝。

參考文獻(xiàn)[1] https://www.cnblogs.com/qingyunzong/p/9004509.html

[2] https://blog.csdn.net/chenleiking/article/details/73563930[3]https://gitee.com/starblues/rope/wikis/pages?sort_id=1863419&doc_id=507971