基于Flink和Hadoop的卡口數(shù)據(jù)分析

許森

摘要 文章針對傳統(tǒng)方式存儲(chǔ)與處理大數(shù)據(jù)的局限性，提出了基于Flink和Hadoop的卡口數(shù)據(jù)分析方法，通過Flink對原始卡口數(shù)據(jù)流進(jìn)行高效的初步處理和異常數(shù)據(jù)過濾，得到高質(zhì)量原始卡口數(shù)據(jù)，采用HBase讀寫海量卡口數(shù)據(jù)，利用MapReduce計(jì)算指定時(shí)間內(nèi)所有車輛的路段旅行時(shí)間，并利用Mahout中的K-Means進(jìn)行聚類分析獲取有效且符合實(shí)際的結(jié)果值，通過實(shí)驗(yàn)驗(yàn)證，證明方法的有效性和合理性。

關(guān)鍵詞 大數(shù)據(jù)；卡口分析；Flink；Hadoop；分布式框架

中圖分類號 TP311.13文獻(xiàn)標(biāo)識碼 A文章編號 2096-8949（2023）09-0004-03

0 引言

隨著物聯(lián)網(wǎng)和云計(jì)算的快速崛起，全球數(shù)據(jù)量大幅度地增加，標(biāo)志著進(jìn)入了大數(shù)據(jù)時(shí)代。海量的數(shù)據(jù)必然無法用人腦來推算和估測，或者用單臺(tái)的計(jì)算機(jī)進(jìn)行處理，必須采用分布式計(jì)算架構(gòu)，依托云計(jì)算的分布式處理、分布式數(shù)據(jù)庫、云存儲(chǔ)和虛擬化技術(shù)。

傳統(tǒng)的流處理框架基于微批量處理，時(shí)效性不能做到真正的毫秒級別，并且只支持處理時(shí)間，即數(shù)據(jù)到達(dá)系統(tǒng)的時(shí)間，而不是數(shù)據(jù)本身的時(shí)間戳，這在面對海量實(shí)時(shí)流數(shù)據(jù)時(shí)會(huì)導(dǎo)致數(shù)據(jù)的亂序和不一致的問題。傳統(tǒng)關(guān)系型數(shù)據(jù)庫存儲(chǔ)與分析大數(shù)據(jù)時(shí)的開銷嚴(yán)重影響了數(shù)據(jù)庫性能。

該文針對海量交通卡口數(shù)據(jù)的過濾存儲(chǔ)與基于卡口數(shù)據(jù)的精準(zhǔn)數(shù)據(jù)分析挖掘進(jìn)行了大量的研究，利用Flink[1]對海量的初始數(shù)據(jù)進(jìn)行初步過濾和計(jì)算，利用HBase[2]存儲(chǔ)卡口數(shù)據(jù)，在Hadoop[3]平臺(tái)上利用MapReduce[4]和Mahout進(jìn)行卡口數(shù)據(jù)的數(shù)據(jù)挖掘分析，獲取路段的有效旅行時(shí)間，克服了傳統(tǒng)手段難過濾海量數(shù)據(jù)，傳統(tǒng)數(shù)據(jù)庫空間擴(kuò)展性差、查詢時(shí)間過長等問題。

1 概述

Flink是一種分布式計(jì)算框架，可以對無界有界的數(shù)據(jù)流進(jìn)行有狀態(tài)計(jì)算。所以，它可以處理實(shí)時(shí)的流數(shù)據(jù)，也可以處理離線的批數(shù)據(jù)，而且可以保證數(shù)據(jù)的正確性和一致性。在面對城市海量卡口數(shù)據(jù)的時(shí)候能夠精準(zhǔn)有效地進(jìn)行實(shí)時(shí)過濾，篩掉部分無效初始數(shù)據(jù)的同時(shí)可以進(jìn)行字段的整合計(jì)算等，為后面大數(shù)據(jù)平臺(tái)進(jìn)行復(fù)雜計(jì)算和分析提供更加有序且有效的數(shù)據(jù)。

HBase是Apache Hadoop的數(shù)據(jù)庫，能夠?qū)Υ髷?shù)據(jù)提供隨機(jī)、實(shí)時(shí)的讀寫訪問功能，具有開源、分布式、可擴(kuò)展及面向?qū)ο蟮奶攸c(diǎn)。城市交通網(wǎng)中具有上百個(gè)路口，各個(gè)方向的卡口在機(jī)動(dòng)車通過時(shí)，都會(huì)向中心傳輸大量數(shù)據(jù)，傳統(tǒng)的數(shù)據(jù)庫擴(kuò)展性差，因此該文采取HBase作為數(shù)據(jù)庫。

MapReduce計(jì)算框架簡單，可以通過它輕易地編寫出同時(shí)在多臺(tái)主機(jī)上運(yùn)行的程序，還可以在任何安裝Hadoop的集群中運(yùn)行同樣的程序，不論這個(gè)集群有多少臺(tái)主機(jī)，因此該文采用MapReduce作為卡口數(shù)據(jù)的處理框架。

Mahout作為一個(gè)機(jī)器學(xué)習(xí)軟件庫，充分利用了Hadoop平臺(tái)，它主要關(guān)注于推薦引擎、聚類和分類。該文需要對卡口數(shù)據(jù)進(jìn)行聚類分析，因此使用Mahout作為數(shù)據(jù)挖掘的工具。

Hive作為構(gòu)建在Hadoop上的數(shù)據(jù)倉庫，可以用來進(jìn)行數(shù)據(jù)提取轉(zhuǎn)化加載（ETL），這是一種可以存儲(chǔ)、查詢和分析存儲(chǔ)在Hadoop中的大規(guī)模數(shù)據(jù)的機(jī)制，該文將計(jì)算結(jié)果存儲(chǔ)于HDFS中，利用Hive進(jìn)行數(shù)據(jù)讀取和展示。

該文通過Flink過濾數(shù)據(jù)，處理HBase中的卡口數(shù)據(jù)表，通過MapReduce過程和K-Means聚類分析，輸出指定路徑上的有效旅行時(shí)間。整個(gè)過程如圖1。

2 卡口數(shù)據(jù)過濾和存儲(chǔ)

2.1 卡口存儲(chǔ)數(shù)據(jù)表設(shè)計(jì)

卡口數(shù)據(jù)數(shù)量龐大，將其存儲(chǔ)在非關(guān)系型數(shù)據(jù)庫HBase中。HBase中rowkey的設(shè)計(jì)應(yīng)該便于數(shù)據(jù)的快速查詢。

旅行時(shí)間查詢是統(tǒng)計(jì)指定時(shí)間段it內(nèi)所有車輛通過指定路徑（p1，…pn）需要的時(shí)間（其中pi為交叉口i），由于行健的設(shè)計(jì)規(guī)則，行健的第一部分不能為到達(dá)時(shí)間，那么為了增加按時(shí)間查詢旅行時(shí)間的查詢速度，需要利用二級索引。

為方便查詢將車牌號+車牌顏色+車牌類型+通過時(shí)間作為Rowkey[5]，利用HBase的主鍵索引功能進(jìn)行快速查詢?？诒磉壿嬕晥D設(shè)計(jì)如表1。

索引數(shù)據(jù)的存儲(chǔ)利用協(xié)處理器，在添加一條卡口數(shù)據(jù)的之前，向卡口表中寫入一行包含查詢條件的索引數(shù)據(jù)，索引數(shù)據(jù)表的設(shè)計(jì)如表2。

2.2 Flink過濾卡口數(shù)據(jù)設(shè)計(jì)

卡口海量初始數(shù)據(jù)流不可避免地會(huì)有明顯異常的數(shù)據(jù)，剔除這些異常數(shù)據(jù)能夠明顯提升需要計(jì)算的數(shù)據(jù)集質(zhì)量。filter算子是Flink的過濾篩選器，將符合條件的數(shù)據(jù)輸出到下一步進(jìn)行計(jì)算。這里設(shè)置過濾條件為卡口編號、車牌號、車牌顏色、車牌類型和通過時(shí)間等字段不允許為空，不符合條件的數(shù)據(jù)在輸入到Flink作業(yè)的時(shí)候就會(huì)被剔除，有效提升數(shù)據(jù)質(zhì)量。

為達(dá)到2.1rowkey的設(shè)計(jì)要求，就需要對卡口數(shù)據(jù)進(jìn)行分組。Flink keyby是一個(gè)基于key的分組轉(zhuǎn)換算子，可以根據(jù)指定的key對數(shù)據(jù)流中的元素進(jìn)行重新劃分。這里設(shè)置keyby為車牌號，F(xiàn)link會(huì)在處理實(shí)時(shí)卡口數(shù)據(jù)時(shí)將車牌號相同的數(shù)據(jù)分到同一組?？谠紨?shù)據(jù)的字段都是單獨(dú)的，在設(shè)計(jì)process時(shí)，把車牌號+車牌顏色+車牌類型+通過時(shí)間合成一個(gè)字段，最后把處理好的結(jié)果通過Flink sink直接寫入到HBase。

3 卡口數(shù)據(jù)處理-MapReduce

MapReduce數(shù)據(jù)處理模型非常簡單，Map和Reduce過程的輸入和輸出是鍵值對（key，value），即遵循如下常規(guī)模式：Map：（k1，v1）→（k2，v2） Reduce（k3，list（v3））→（k4，v4）

該文MapReduce的輸入是HBase的查詢結(jié)果，通過巧妙的Rowkey設(shè)計(jì)與二級索引使批量獲取記錄集合中的元素挨在一起（應(yīng)該在同一個(gè)Region下），可以在遍歷結(jié)果時(shí)獲得很好的性能。Hadoop平臺(tái)上的編程模式是將程序拆分成Map過程和Reduce過程，Map過程將HBase中的行數(shù)據(jù)進(jìn)行處理，為Reduce過程提供輸入。Reduce過程收集map過程的輸出，對其進(jìn)行匯總與計(jì)算，并將結(jié)果寫入HDFS[6]。

3.1 Map過程

Map過程利用Scan類讀取HBase中的數(shù)據(jù)，從HBase表中讀取的作業(yè)以[rowkey：result]格式接收（key，value）鍵值對，即CarNum+TypeColor+CarType+PassDate作為key，列族cf1作為value輸入。經(jīng)過程序的簡單處理，將CarNum+TypeColor+CarType作為key，PointCode+PassDate作為value輸出。

3.2 Reduce過程

經(jīng)過Shuffle過程的處理，Reduce過程將Map過程的輸出作為輸入，對車牌號為cnx的車輛的所有通過實(shí)踐（PassDate）進(jìn)行排序，排序后的時(shí)間為（st1，…sty），對應(yīng)的交叉口為（pc1，…pcy），定義PC（sti）為在時(shí)間sti到達(dá)的路口。定義RD（xi，…xj）為移除了（xi，…xj）中相鄰重復(fù)點(diǎn)的向量（保留第一個(gè)重復(fù)點(diǎn)）。cnx在下的旅行時(shí)間TTx如下計(jì)算：

TTx={stj?sti|RD（PC（sti），…，PC（stj））=}

在Reduce過程完成后，按照CarNum+StartTime作為key，TTx作為value輸出，將數(shù)據(jù)持久化存儲(chǔ)于HDFS中。

3.3 實(shí)驗(yàn)

在包含1 000 000條卡口數(shù)據(jù)的數(shù)據(jù)集上進(jìn)行該次實(shí)驗(yàn)，取兩個(gè)相鄰路口組成，根據(jù)路段距離，將旅行時(shí)間上限閾值設(shè)置為2 min，下限閾值設(shè)置為0.5 min，統(tǒng)計(jì)的時(shí)間為00：00：00—00：30：00，在以上輸入?yún)?shù)下運(yùn)行MapReduce過程，計(jì)算結(jié)果如圖2中橫坐標(biāo)為車輛在路線的起始時(shí)間，縱軸為完成路線的總時(shí)間（min），例如在00：10：00有5輛車通過卡口，這5輛車通過路線的時(shí)間分別為（0.87，0.88，0.9，0.88，0.87），表示為圖2的3個(gè)較大的圓點(diǎn)。

4 卡口數(shù)據(jù)處理-K-Means

4.1 利用Hive讀取數(shù)據(jù)

在MapReduce處理卡口數(shù)據(jù)后，結(jié)果被存儲(chǔ)于HDFS中，可以通過建立外部表的方式讀取HDFS上的數(shù)據(jù)?？梢酝ㄟ^如下方式建立Hive外部表：

Create external table TGS_DATA_HIS_TRAVELTIME

（data string，time string）

row format delimited

fields terminated by ‘@

location‘/user/TGS_DATA_HIS_TRAVELTIME_RESULT

MapReduce處理后的結(jié)果數(shù)據(jù)按行寫入HDFS目錄/user/TGS_DATA_HIS_TRAVELTIME_RESULT中，鍵值對以@分隔。利用select * from TGS_DATA_HIS_TRAVELTIME可以獲取結(jié)果數(shù)據(jù)。

4.2 K-Means過程

K-Means方法作為聚類分析的一種，是在沒有給定劃分類別的情況下，根據(jù)數(shù)據(jù)相似度進(jìn)行樣本分組的一種方法。K-Means的處理過程包含輸入格式化，隨機(jī)生成Cluster，聚類計(jì)算，結(jié)果輸出等過程。算法過程如下：

（1）讀取MapReduce過程輸出的數(shù)據(jù)，利用一個(gè)Map過程將HDFS上的數(shù)據(jù)轉(zhuǎn)換成SequenceFile文件，數(shù)據(jù)以向量形式保存。

（2）隨后采用蓄水池抽樣的方法在上面的向量里隨機(jī)抽樣k個(gè)序列族Cluster。

（3）分別計(jì)算各向量到各個(gè)聚類中心的距離，將向量分配到距離最近的聚類中。

（4）完成所有向量的分配后，重新計(jì)算K個(gè)聚類的中心。

（5）不斷迭代，當(dāng)滿足迭代結(jié)束條件后，聚類完成。

采用包含向量個(gè)數(shù)最多的簇的中心點(diǎn)作為旅行時(shí)間的有效值。

4.3 實(shí)驗(yàn)

實(shí)驗(yàn)中設(shè)k=3，最大迭代次數(shù)為3，收斂閾值設(shè)為1，K-Means過程處理卡口數(shù)據(jù)后的結(jié)果如圖3，圖 中橫坐標(biāo)為車輛在路線的起始時(shí)間，縱軸為完成路線的總時(shí)間（min）。圖中各點(diǎn)的集合為含有最多向量的簇集，該簇集的中心點(diǎn)為0.84 m，取該值表示的50 s作為路線旅行時(shí)間的有效值，該值與多次人工實(shí)地跑車的有效結(jié)果時(shí)間在評價(jià)應(yīng)用中的誤差可以忽略，可知使用該文的方法可以準(zhǔn)確地對路線的旅行時(shí)間進(jìn)行分析計(jì)算。

5 結(jié)語

該文針對傳統(tǒng)存儲(chǔ)與處理大數(shù)據(jù)的局限性，提出了基于Flink和Hadoop的卡口數(shù)據(jù)挖掘分析方法，通過Flink框架來過濾原始卡口數(shù)據(jù)流，提高用于分析的數(shù)據(jù)集質(zhì)量，利用MapReduce和K-Means聚類算法計(jì)算指定時(shí)間內(nèi)所有車輛的路段旅行時(shí)間，并利用實(shí)驗(yàn)驗(yàn)證了方法的有效性。該值可用于區(qū)域交通方案的評估分析，如綠波效果的評估等，為用戶提供快速的輔助分析決策，克服了傳統(tǒng)人工實(shí)地跑車的方式進(jìn)行統(tǒng)計(jì)，對實(shí)現(xiàn)智能交通管理具有重要的意義。

參考文獻(xiàn)

[1]Garcia-Gil D， Ramfrez-Gallego S， Garcia S， et al. A Comparison on Scalability Forbatch Big Data Processing on Apache Spark and Apache Flink[J]. Big DataAnalytics，2017（1）： 20-21.

[2]Chang F， Dean J， Ghemawat S， et al. Bigtable： A distributed Storage System for Structured Data[J]. ACM Transactions on Computer Systems （TOCS）， 2008（2）： 1-26.

[3]Borthakur D， Gray J， Sarma J S， et al. Apache Hadoop Goes Realtime at Facebook[C].Proceedings of the 2011 ACM SIGMOD International Conference on Management of Data. ACM， 2011： 1071-1080.

[4]Mahout A. Scalable Machine Learning and Data Mining[J]. 2013-4-24. http： //mahout. Apache.org， 2012.

[5]Ghemawat S， Gobioff H， Leung S T. The Google File System[C].ACM SIGOPS Operating Systems Review. ACM， 2003（5）： 29-43.

[6]戴傳友. 基于云計(jì)算的交通卡口數(shù)據(jù)分析[D].廣州：華南理工大學(xué)， 2016.