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

許森

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

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

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

0 引言

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

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

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

1 概述

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

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

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

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

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

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

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

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

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

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

為方便查詢將車牌號+車牌顏色+車牌類型+通過時間作為Rowkey[5]，利用HBase的主鍵索引功能進行快速查詢。卡口表邏輯視圖設(shè)計如表1。

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

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

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

為達到2.1rowkey的設(shè)計要求，就需要對卡口數(shù)據(jù)進行分組。Flink keyby是一個基于key的分組轉(zhuǎn)換算子，可以根據(jù)指定的key對數(shù)據(jù)流中的元素進行重新劃分。這里設(shè)置keyby為車牌號，F(xiàn)link會在處理實時卡口數(shù)據(jù)時將車牌號相同的數(shù)據(jù)分到同一組。卡口原始數(shù)據(jù)的字段都是單獨的，在設(shè)計process時，把車牌號+車牌顏色+車牌類型+通過時間合成一個字段，最后把處理好的結(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è)計與二級索引使批量獲取記錄集合中的元素挨在一起（應(yīng)該在同一個Region下），可以在遍歷結(jié)果時獲得很好的性能。Hadoop平臺上的編程模式是將程序拆分成Map過程和Reduce過程，Map過程將HBase中的行數(shù)據(jù)進行處理，為Reduce過程提供輸入。Reduce過程收集map過程的輸出，對其進行匯總與計算，并將結(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的車輛的所有通過實踐（PassDate）進行排序，排序后的時間為（st1，…sty），對應(yīng)的交叉口為（pc1，…pcy），定義PC（sti）為在時間sti到達的路口。定義RD（xi，…xj）為移除了（xi，…xj）中相鄰重復(fù)點的向量（保留第一個重復(fù)點）。cnx在下的旅行時間TTx如下計算：

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

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

3.3 實驗

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

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

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

在MapReduce處理卡口數(shù)據(jù)后，結(jié)果被存儲于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ù)相似度進行樣本分組的一種方法。K-Means的處理過程包含輸入格式化，隨機生成Cluster，聚類計算，結(jié)果輸出等過程。算法過程如下：

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

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

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

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

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

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

4.3 實驗

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

5 結(jié)語

該文針對傳統(tǒng)存儲與處理大數(shù)據(jù)的局限性，提出了基于Flink和Hadoop的卡口數(shù)據(jù)挖掘分析方法，通過Flink框架來過濾原始卡口數(shù)據(jù)流，提高用于分析的數(shù)據(jù)集質(zhì)量，利用MapReduce和K-Means聚類算法計算指定時間內(nèi)所有車輛的路段旅行時間，并利用實驗驗證了方法的有效性。該值可用于區(qū)域交通方案的評估分析，如綠波效果的評估等，為用戶提供快速的輔助分析決策，克服了傳統(tǒng)人工實地跑車的方式進行統(tǒng)計，對實現(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]戴傳友. 基于云計算的交通卡口數(shù)據(jù)分析[D].廣州：華南理工大學(xué)， 2016.