許森

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

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

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

0 引言

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

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

該文針對海量交通卡口數(shù)據(jù)的過濾存儲與基于卡口數(shù)據(jù)的精準數(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ù)的時候能夠精準有效地進行實時過濾，篩掉部分無效初始數(shù)據(jù)的同時可以進行字段的整合計算等，為后面大數(shù)據(jù)平臺進行復雜計算和分析提供更加有序且有效的數(shù)據(jù)。

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

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

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

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

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

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

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

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

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

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

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

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

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

為達到2.1rowkey的設計要求，就需要對卡口數(shù)據(jù)進行分組。Flink keyby是一個基于key的分組轉換算子，可以根據(jù)指定的key對數(shù)據(jù)流中的元素進行重新劃分。這里設置keyby為車牌號，F(xiàn)link會在處理實時卡口數(shù)據(jù)時將車牌號相同的數(shù)據(jù)分到同一組?？谠紨?shù)據(jù)的字段都是單獨的，在設計process時，把車牌號+車牌顏色+車牌類型+通過時間合成一個字段，最后把處理好的結果通過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的查詢結果，通過巧妙的Rowkey設計與二級索引使批量獲取記錄集合中的元素挨在一起（應該在同一個Region下），可以在遍歷結果時獲得很好的性能。Hadoop平臺上的編程模式是將程序拆分成Map過程和Reduce過程，Map過程將HBase中的行數(shù)據(jù)進行處理，為Reduce過程提供輸入。Reduce過程收集map過程的輸出，對其進行匯總與計算，并將結果寫入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），對應的交叉口為（pc1，…pcy），定義PC（sti）為在時間sti到達的路口。定義RD（xi，…xj）為移除了（xi，…xj）中相鄰重復點的向量（保留第一個重復點）。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ù)路段距離，將旅行時間上限閾值設置為2 min，下限閾值設置為0.5 min，統(tǒng)計的時間為00：00：00—00：30：00，在以上輸入?yún)?shù)下運行MapReduce過程，計算結果如圖2中橫坐標為車輛在路線的起始時間，縱軸為完成路線的總時間（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ù)后，結果被存儲于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處理后的結果數(shù)據(jù)按行寫入HDFS目錄/user/TGS_DATA_HIS_TRAVELTIME_RESULT中，鍵值對以@分隔。利用select * from TGS_DATA_HIS_TRAVELTIME可以獲取結果數(shù)據(jù)。

4.2 K-Means過程

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

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

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

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

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

（5）不斷迭代，當滿足迭代結束條件后，聚類完成。

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

4.3 實驗

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

5 結語

該文針對傳統(tǒng)存儲與處理大數(shù)據(jù)的局限性，提出了基于Flink和Hadoop的卡口數(shù)據(jù)挖掘分析方法，通過Flink框架來過濾原始卡口數(shù)據(jù)流，提高用于分析的數(shù)據(jù)集質量，利用MapReduce和K-Means聚類算法計算指定時間內所有車輛的路段旅行時間，并利用實驗驗證了方法的有效性。該值可用于區(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].廣州：華南理工大學， 2016.