【51CTO.com快譯】

[[175601]]

物聯(lián)網(wǎng)數(shù)據(jù)有望發(fā)掘獨(dú)特的、前所未有的業(yè)務(wù)洞察力，不過(guò)前提是企業(yè)能夠成功地管理從眾多物聯(lián)網(wǎng)數(shù)據(jù)源流入的數(shù)據(jù)。許多企業(yè)試圖從物聯(lián)網(wǎng)項(xiàng)目獲得價(jià)值，但經(jīng)常遇到的一個(gè)問(wèn)題是數(shù)據(jù)漂移(data drift)：源設(shè)備和數(shù)據(jù)處理基礎(chǔ)設(shè)施經(jīng)常發(fā)生不可預(yù)測(cè)的變化，因而導(dǎo)致數(shù)據(jù)的結(jié)構(gòu)、內(nèi)容或含義發(fā)生變化。

無(wú)論流式處理還是批量處理，數(shù)據(jù)通常經(jīng)由眾多工具，從數(shù)據(jù)源進(jìn)入到最后的存儲(chǔ)位置。這條鏈上任何地方的變化都會(huì)導(dǎo)致下流系統(tǒng)中出現(xiàn)不完整、不準(zhǔn)確或不一致的數(shù)據(jù)，無(wú)論是源系統(tǒng)的模式發(fā)生變化、編碼字段值的含義發(fā)生變化，還是參與數(shù)據(jù)生成的軟件組件出現(xiàn)升級(jí)或添加。

這種數(shù)據(jù)漂移的影響可能危害特別大，因?yàn)樗鼈兂３ｉL(zhǎng)時(shí)間沒(méi)有被發(fā)現(xiàn)，因而讓低逼真度數(shù)據(jù)污染了數(shù)據(jù)存儲(chǔ)和隨后的分析。在被發(fā)現(xiàn)之前，使用這種有問(wèn)題的數(shù)據(jù)會(huì)導(dǎo)致錯(cuò)誤的發(fā)現(xiàn)結(jié)果和拙劣的業(yè)務(wù)決定。等到最后發(fā)現(xiàn)了問(wèn)題，通常借助數(shù)據(jù)科學(xué)家的手動(dòng)數(shù)據(jù)清理和準(zhǔn)備來(lái)加以解決，這給數(shù)據(jù)分析增添了硬性成本、機(jī)會(huì)成本和延誤。

StreamSets Data Collector

使用StreamSets Data Collector來(lái)構(gòu)建和管理大數(shù)據(jù)攝取管道將有助于緩解數(shù)據(jù)漂移的影響，同時(shí)大大縮短花在數(shù)據(jù)清理上的時(shí)間。我們?cè)诒疚闹袑⒅鸩浇榻B一種典型的使用場(chǎng)合：實(shí)時(shí)攝取物聯(lián)網(wǎng)傳感器生成的數(shù)據(jù)，饋入到HDFS，以便分析，并使用Impala或Hive實(shí)現(xiàn)可視化。

不用編寫一行代碼，StreamSets Data Collector就能從眾多數(shù)據(jù)源攝取流數(shù)據(jù)和批量數(shù)據(jù)。StreamSets Data Collector可執(zhí)行轉(zhuǎn)換，并在數(shù)據(jù)流傳輸過(guò)程中清理數(shù)據(jù)，然后寫入到眾多目的地。管道部署到位后，你就能獲得細(xì)粒度的數(shù)據(jù)流度量指標(biāo)、檢測(cè)異常數(shù)據(jù)，并發(fā)出警報(bào)，那樣你就能密切關(guān)注管道性能。StreamSets Data Collector可以獨(dú)立運(yùn)行，也可以部署在Hadoop集群上，它提供了支持眾多類型的數(shù)據(jù)源和目的地的連接件。

下列使用場(chǎng)合涉及從貨運(yùn)集裝箱實(shí)時(shí)生成的數(shù)據(jù)。

數(shù)據(jù)漂移的第一個(gè)例子體現(xiàn)在貨運(yùn)集裝箱使用的物聯(lián)網(wǎng)傳感器。由于長(zhǎng)期以來(lái)的升級(jí)，生產(chǎn)一線的傳感器運(yùn)行三種不同固件版本中的一種。每種版本添加新的數(shù)據(jù)字段，改變模式。為了從該傳感器數(shù)據(jù)獲得價(jià)值，我們用來(lái)攝取信息的系統(tǒng)必須能夠兼顧這種多樣性。

清潔和傳送數(shù)據(jù)

我們的管道從RabbitMQ系統(tǒng)讀取數(shù)據(jù)，該系統(tǒng)負(fù)責(zé)從生產(chǎn)一線的傳感器接收MQTT消息。我們進(jìn)行核實(shí)，確保我們收到的消息正是想要處理的那些消息。為此，我們使用數(shù)據(jù)流選擇器處理程序，為入站消息指定數(shù)據(jù)規(guī)則。然后，我們使用該規(guī)則宣布與該規(guī)則的標(biāo)準(zhǔn)匹配的所有數(shù)據(jù)都傳送到下游，但是不匹配的任何數(shù)據(jù)一概被丟棄。

然后，我們使用另一個(gè)數(shù)據(jù)流選擇器，根據(jù)設(shè)備的固件版本來(lái)傳送數(shù)據(jù)。與版本1匹配的所有記錄走一條路徑，與版本2匹配的所有記錄走另一條路徑，以此類推。我們還指定了一條默認(rèn)的全部捕獲(catch-all)規(guī)則，將任何異常發(fā)送到一條“錯(cuò)誤”路徑。針對(duì)現(xiàn)代數(shù)據(jù)流，我們完全預(yù)料到數(shù)據(jù)會(huì)出現(xiàn)意外的變化，于是我們?cè)O(shè)立了一種從容的錯(cuò)誤處理機(jī)制：把異常記錄引到本地文件、Kafka數(shù)據(jù)流或輔助管道。那樣一來(lái)，我們就能保持管道正常運(yùn)行，同時(shí)事后重新處理不符合主要目的的數(shù)據(jù)。

不妨從為固件版本3處理數(shù)據(jù)入手，這增添了緯度/經(jīng)度數(shù)據(jù)。我們馬上想要確保那些字段出現(xiàn)在數(shù)據(jù)集中，而且數(shù)據(jù)包含有效的值。由于位置字段是一個(gè)嵌套的結(jié)構(gòu)，我們想要對(duì)它作扁平化處理，最終丟棄嵌套的數(shù)據(jù)。

同樣，固件版本2包括新的方位字段(raw、pitch和roll)，我們可以以一種類似的方式來(lái)核實(shí)和清潔它。

最后，所有設(shè)備版本都包含溫度和濕度讀數(shù)。首先，我們轉(zhuǎn)換這些讀數(shù)的數(shù)據(jù)類型。濕度轉(zhuǎn)換成了雙精度浮點(diǎn)型(double)，濕度轉(zhuǎn)換成了整型(integer)，日期轉(zhuǎn)換成了Unix時(shí)間戳。

然后我們使用腳本處理程序來(lái)編寫一些自定義邏輯，比如將華氏度值轉(zhuǎn)換成攝氏度。StreamSets腳本處理程序支持Jython、Groovy和JavaScript。

清理數(shù)據(jù)(也就是根據(jù)固件版本和最終用途來(lái)傳送數(shù)據(jù))后，我們把它發(fā)送到幾個(gè)HDFS目的地。

配置目的地

StreamSets本身支持許多數(shù)據(jù)格式，比如明文、分隔文本、JSON、Protobuf和Avro。在該例子中，我們將把數(shù)據(jù)寫入到一個(gè)經(jīng)過(guò)壓縮的Avro文件。

HDFS目的地可以靈活配置。你可以按照企業(yè)政策的要求來(lái)配置安全、動(dòng)態(tài)配置輸出文件的路徑和位置，甚至決定寫入多個(gè)Cloudera CDH版本。

一旦你設(shè)計(jì)好了管道，就可以切換至預(yù)覽模式，使用數(shù)據(jù)樣本來(lái)測(cè)試和調(diào)試數(shù)據(jù)流。你可以逐步調(diào)試每一個(gè)處理程序，在任何階段分析數(shù)據(jù)狀態(tài)。

比如說(shuō)，我們可以在下面看到，reading_date和temperature的數(shù)據(jù)類型被轉(zhuǎn)換成了長(zhǎng)整型和雙精度浮點(diǎn)型。如果執(zhí)行了轉(zhuǎn)換數(shù)據(jù)的運(yùn)算，StreamSets也會(huì)提醒你。

你還可以把異?；?ldquo;極端情況”數(shù)據(jù)注入到數(shù)據(jù)流，看看它對(duì)你的數(shù)據(jù)流有何影響。預(yù)覽模式讓你可以輕松地調(diào)試復(fù)雜的管道，不需要把管道放入到生產(chǎn)環(huán)境。

執(zhí)行管道

現(xiàn)在我們準(zhǔn)備執(zhí)行管道，開始將數(shù)據(jù)攝入到我們的集群中。點(diǎn)擊“開始”按鈕，用戶界面就會(huì)切換至執(zhí)行模式。

這時(shí)候，StreamSets Data Collector開始攝取數(shù)據(jù)，在內(nèi)存中處理數(shù)據(jù)，然后將數(shù)據(jù)發(fā)送到目的地。屏幕底部的監(jiān)控窗口顯示了各個(gè)實(shí)時(shí)度量指標(biāo)，比如多少記錄讀入、多少記錄寫出。你還可以查看多少時(shí)間花在了每個(gè)處理程序上，它占用了多少內(nèi)存。這些度量指標(biāo)以及更多的指標(biāo)還可以通過(guò)Java管理擴(kuò)展(JMX)來(lái)加以訪問(wèn)。

我們將數(shù)據(jù)送入到HDFDS后，立即就能開始查詢Impala，并運(yùn)行分析、機(jī)器學(xué)習(xí)或可視化。

如今，由于用戶改動(dòng)和更新系統(tǒng)，或者甚至更換平臺(tái)，物聯(lián)網(wǎng)設(shè)備、傳感器日志、Web點(diǎn)擊流及其他重要數(shù)據(jù)源在不斷變化。數(shù)據(jù)內(nèi)容、結(jié)構(gòu)、行為和含義的這些變化是不可預(yù)測(cè)、未宣布、沒(méi)完沒(méi)了的，它們會(huì)給數(shù)據(jù)處理和分析系統(tǒng)及其運(yùn)營(yíng)帶來(lái)重大危害。StreamSets Data Collector有助于管理數(shù)據(jù)基礎(chǔ)設(shè)施不斷出現(xiàn)的變化，馴服數(shù)據(jù)漂移，并確保數(shù)據(jù)處理系統(tǒng)的完整性。

原文標(biāo)題：Tame unruly big data flows with StreamSets，作者：Arvind Pabhakar

【51CTO譯稿，合作站點(diǎn)轉(zhuǎn)載請(qǐng)注明原文譯者和出處為51CTO.com】