鄭曉東 鄭業(yè)爽 宋思琪

摘? 要： 分析移動通信時空大數(shù)據(jù)可以得到城市居民的出行情況和活動規(guī)律，從而為城市交通措施優(yōu)化提供數(shù)據(jù)支持和科學依據(jù)。該模型首先清洗原始通信數(shù)據(jù)，并對已清洗的數(shù)據(jù)做必要轉換。然后使用kmeans聚集算法和鄧恩指數(shù)來計算最佳聚類區(qū)域，并結合實際邏輯判斷，標識出人群的駐留或途經(jīng)狀態(tài)。最后利用Hadoop中的MapReduce和Hive組件對數(shù)據(jù)進行分析匯總，針對應用場景得到相應的人群聚集模型并以可視化的方式呈現(xiàn)出來。

關鍵詞： 人群聚集； 時空大數(shù)據(jù)； kmeans算法； 鄧恩指數(shù)； MapReduce； Hive

中圖分類號：TP181；TP311? ? ? ? ? 文獻標識碼：A? ? ?文章編號：1006-8228（2023）04-67-05

Abstract： By analyzing the spatiotemporal big data of mobile communication， we can get the travel situation and activity rules of urban residents， thus providing data support and scientific basis for the optimization of urban traffic measures. In this model， the original communication data is first cleaned， and the necessary conversion is made on the cleaned data. Then kmeans clustering algorithm and Dunn index are used to calculate the best clustering region and combine with practical logic judgment to identify the resident or passing state of the crowd. Finally， the MapReduce and Hive components in Hadoop are used to analyze and summarize the data， and the corresponding crowd gathering model is obtained according to the application scenario and presented in a visual way.

Key words： crowd gathering; spatiotemporal big data; kmeans algorithm; Dunn's index; MapReduce; Hive

0 引言

城市交通出行日益擁堵，節(jié)假日人群階段性的大規(guī)模聚集越來越常見。如何優(yōu)化交通，避免交通堵塞，決策者需要詳實的人們出行和聚集數(shù)據(jù)作為制定政策的依據(jù)[1]。通過搭建Hadoop時空大數(shù)據(jù)存儲分析處理平臺，利用智慧城市建設中搜集的海量時空大數(shù)據(jù)，加以分析處理并以可視化的結果展示，可以使Hadoop相關技術在存儲、管理、挖掘時空大數(shù)據(jù)等方面的優(yōu)勢很好地用于解決此類問題[2]。

Hadoop生態(tài)系統(tǒng)中最為主要的技術包括HDFS、MapReduce和Hive。HDFS實現(xiàn)的海量大數(shù)據(jù)分布式集群存儲，大幅提高了計算機的數(shù)據(jù)存儲能力和數(shù)據(jù)操作便捷性。MapReduce的分布式并行編程架構，使得無須昂貴的服務器，僅利用廉價的桌面計算機集群就可批處理HDFS上的非實時海量數(shù)據(jù)[3]。Hive是數(shù)據(jù)倉庫工具，它把數(shù)據(jù)文件以數(shù)據(jù)表方式來解釋，提供類似傳統(tǒng)ER數(shù)據(jù)庫的結構化查詢語言HiveSQL，即HQL供用戶使用，并在底層將HQL執(zhí)行語句轉換成對應功能一致的MapReduce任務來執(zhí)行。Hive典型的功能是頻繁分析數(shù)據(jù)并生成數(shù)據(jù)報表。

本文詳細論述Hadoop框架中的HDFS和Yarn安裝和配置，并通過MapReduce和Hive編程演示，結合kmeans聚集算法[4]和鄧恩指數(shù)[5]，來論述如何清洗并處理時空大數(shù)據(jù)，最終得到可視化展示結果。

1 Hadoop框架的分析模型業(yè)務架構

時空大數(shù)據(jù)Hadoop[6]分析模型業(yè)務架構如圖1所示。先使用load命令將本地csv數(shù)據(jù)文件導入到MySQL中，再使用Sqoop將MySQL中的數(shù)據(jù)，導入到hdfs上存儲。然后使用SparkSQL按照設計規(guī)則清洗數(shù)據(jù)，并將結果數(shù)據(jù)存儲在hdfs上。如果需要實時數(shù)據(jù)處理，可以使用kafka模擬真實業(yè)務場景的實時流數(shù)據(jù)，再使用Spark streaming做實時運算處理，并將結果數(shù)據(jù)保存到hdfs上。其中數(shù)據(jù)分析和處理需要理解清洗后的真實數(shù)據(jù)，找出數(shù)據(jù)存在的特征并建立算法模型。例如為了實現(xiàn)途經(jīng)或駐留應用場景分析，需要分析清洗后存儲在hdfs上的數(shù)據(jù)，確定合適的算法模型，再通過聚類算法與邏輯相結合的方式實現(xiàn)途經(jīng)或駐留的識別，將結果存儲在Hive中[7]。統(tǒng)計和數(shù)據(jù)可視化是結合業(yè)務場景需求，將統(tǒng)計出的不同維度的結果數(shù)據(jù)以可視化方式展示給最終用戶。

此系統(tǒng)架構主要分為兩部分，業(yè)務程序部分和數(shù)據(jù)存儲部分。業(yè)務程序部分主要是Hadoop的Spark離線處理和實時處理。Spark離線處理主要對之前1～2天的數(shù)據(jù)進行處理，并輸出途經(jīng)或駐留、人群密度以及其他業(yè)務場景。Spark實時處理主要對當前數(shù)據(jù)進行處理，輸出人群密度熱力結果數(shù)據(jù)。數(shù)據(jù)存儲部分主要有兩種存儲方式：MySQL和HDFS&Hive。原始數(shù)據(jù)和靜態(tài)數(shù)據(jù)存儲于關系型數(shù)據(jù)庫MySQL中。經(jīng)Sqoop導入后的數(shù)據(jù)都存儲在hdfs中，包括清洗數(shù)據(jù)和結果數(shù)據(jù)，結果數(shù)據(jù)存儲在Hive中，便于組合分析和展示。

2 時空大數(shù)據(jù)的導入和清洗

交通時空大數(shù)據(jù)分析系統(tǒng)需要使用移動通信數(shù)據(jù)來處理和分析[8]。在可能出現(xiàn)人口大量聚集的城鎮(zhèn)區(qū)域內，衡量人群聚集達到何種密度，采取何種舉措來合理疏散人群，既緩解交通壓力又不影響商業(yè)運行和區(qū)域價值體現(xiàn)，需要有科學客觀的數(shù)據(jù)依據(jù)。通過和某市電信運營商合作，由其提供本地信令數(shù)據(jù)，并將我們實現(xiàn)的分析模型部署到該運營商的機房內，共同完成人群出行分析并將最終分析結果提供給某城市交通優(yōu)化決策者。

本模型主要基于以下原始業(yè)務信息數(shù)據(jù)來完成分析：通信信息數(shù)據(jù)，基站經(jīng)緯度數(shù)據(jù)和商業(yè)區(qū)數(shù)據(jù)。原始通信信息示例數(shù)據(jù)如圖2所示。

逗號分隔的原始業(yè)務信息數(shù)據(jù)包含10個字段，分別是ts_end信息記錄結束毫秒時間戳，imsi用戶識別碼惟一id，lac_id基站位置區(qū)編碼，cell_id扇區(qū)編號，phone電話號，ts_start信息記錄開始毫秒時間戳，tmp0一級行政區(qū)編號，tmp1二級行政區(qū)編號，nid信息類別1，npid信息類別2?；窘?jīng)緯度示例數(shù)據(jù)和商業(yè)區(qū)示例數(shù)據(jù)分別如圖3和圖4所示。

基站經(jīng)緯度數(shù)據(jù)包含字段longitude經(jīng)度，latitude緯度，laci基站信息。商業(yè)區(qū)數(shù)據(jù)包含字段longitude經(jīng)度，latitude緯度，area_name商業(yè)區(qū)名稱。

原始數(shù)據(jù)清洗主要是針對通信信息數(shù)據(jù)。我們需要抽取ts_end，imsi，longitude，latitude四個字段，并去除imsi字段中包含特殊字符如‘#，*的數(shù)據(jù)條目。業(yè)務信息殘缺的記錄條目，即imsi、lac_id、cell_id字段值為空的記錄條目需要刪除。干擾數(shù)據(jù)條目即不是2022.06.03當天的數(shù)據(jù)需要刪除。通信信息數(shù)據(jù)和基站經(jīng)緯度數(shù)據(jù)關聯(lián)后經(jīng)緯度字段為空的數(shù)據(jù)條目需要刪除。ts_end時間戳格式需要轉換為年月日時分秒的形式，如‘20220603000035。以人為單位按時間正序排序后，以分鐘為粒度單位刪除重復數(shù)據(jù)，并刪除同一imsi所關聯(lián)數(shù)據(jù)條目小于10的所有該imsi關聯(lián)數(shù)據(jù)。清洗后的實時流數(shù)據(jù)如圖5所示。

3 時空大數(shù)據(jù)的數(shù)據(jù)建模分析處理

原始數(shù)據(jù)清洗和轉化后，可以通過k-means聚類算法[9]和鄧恩指數(shù)[10]對數(shù)據(jù)加以標識，以備后期數(shù)據(jù)抽取和展現(xiàn)的處理。但在正式對其分析和處理前，需要識別途經(jīng)和駐留。途經(jīng)和駐留識別結果數(shù)據(jù)的實現(xiàn)分三階段，第一階段是數(shù)據(jù)預處理，算法主要識別出駐留區(qū)域，將干擾性強的途經(jīng)點先標注出來并輸出。第二階段讀取清洗后的數(shù)據(jù)和第一階段輸出數(shù)據(jù)，將第一階段標注的途經(jīng)點暫時去除。然后以某人一天的數(shù)據(jù)為單位，利用無監(jiān)督聚類的方法，設置多個聚類數(shù)目，通過聚類評分指標評價出效果最好的位置點，生成候選駐留點。這個階段使用k-means聚類算法，先識別出人的駐留區(qū)域，通過邏輯算法再識別駐留區(qū)域內不符合條件的途經(jīng)點，最后將途經(jīng)和駐留的數(shù)據(jù)信息標簽化。第三階段通過邏輯算法，將不符合邏輯的候選駐留點判定為途經(jīng)點。最后整合各階段標注點，輸出途經(jīng)和駐留識別結果數(shù)據(jù)。

第一階段，即數(shù)據(jù)預處理階段，讀取清洗后的數(shù)據(jù)，抽取timestamp時間戳，imsi，經(jīng)度，緯度四個字段創(chuàng)建臨時表，然后獲取每個imsi所出現(xiàn)經(jīng)緯度點次數(shù)小于4的數(shù)據(jù)，即數(shù)據(jù)中出現(xiàn)頻率三次以內的點，按照時間排序。這些數(shù)據(jù)條目在時間順序上可能獨立一行，也可能幾行連續(xù)。上述過程需要通過Python腳本判斷當前點的前后數(shù)據(jù)是否連續(xù)，并計算連續(xù)段的速度。處理的步驟如下。

⑴ 將統(tǒng)計好的數(shù)據(jù)轉成list列表并遍歷這個列表。

⑵ 如果當前點為起點數(shù)據(jù)，判斷該點與下一個點的順序是否匹配。如果是則算作連續(xù)點，如果不是則算作單點。

⑶ 如果當前點為尾點數(shù)據(jù)，判斷該點與上一個點的順序是否匹配。如果是則算作連續(xù)點，如果不是則算作單點。

⑷ 如果當前點為中間點數(shù)據(jù)，判斷該點是否與上一個點和下一個點的順序都不匹配，如果是則算作單點，連續(xù)段結束。如果與上一個點匹配但與下一個點不匹配，則算作連續(xù)點，連續(xù)段結束。如果都不是，則判斷為連續(xù)點，并準備循環(huán)處理下一個點的數(shù)據(jù)。

第二階段，即算法建模階段，首先讀取清洗后的數(shù)據(jù)和第一階段處理后的數(shù)據(jù)，去除第一階段判定為途經(jīng)點的數(shù)據(jù)。這里需要將nid信息類別字段加上，方便第三階段的業(yè)務邏輯處理。然后將數(shù)據(jù)以人為單位進行分割，并將數(shù)據(jù)轉換成 kmeans算法需要的向量格式。之后遍歷估算的最佳聚類數(shù)目，一般為3到9，每次計算當前數(shù)據(jù)單元的kmeans的聚類效果。算法實現(xiàn)為遍歷數(shù)據(jù)列表，包含字段類簇、經(jīng)度、緯度，并計算任意兩個經(jīng)緯度點之間的間距。相同類簇的在一個list列表，不同類簇在另一個list列表。在遍歷聚類數(shù)目的同時計算其鄧恩指數(shù)DVI，鄧恩指數(shù)DVI的計算公式如下：

即：DVI=任意兩個簇元素的最短類間距離除以任意簇中的最大類內距離。注意需要去掉同一個簇內兩點最大間距為零的點。接著找出鄧恩指數(shù)最大的類簇，通過聚類評分指標即鄧恩指數(shù)確定好最佳聚類后重新計算。之后將類簇的類別和每個類簇的中心點經(jīng)緯度插進計算好的數(shù)據(jù)中，格式為：ims，類簇，數(shù)據(jù)點的經(jīng)度，數(shù)據(jù)點的緯度，類簇中心點的經(jīng)度，類簇中心點的緯度。最后與第一階段處理后的數(shù)據(jù)表關聯(lián)以獲取數(shù)據(jù)的時間，并保存計算結果。

第三階段，為業(yè)務邏輯實現(xiàn)階段。讀取第二階段算法建模得到的計算結果，按照imsi分組和時間字段排序。然后計算類簇中數(shù)據(jù)點的前后時間跨度和數(shù)據(jù)點距離類簇中心點的間距。處理方式如下：

① 將處理好的數(shù)據(jù)按照鍵值對分組，鍵key和值value分別為：

key：（imsi-cluster）

value：（time，longitude，latitude，cluster，cluster_longitude，

cluster_latitude，nid）

② 遍歷分組后的數(shù)據(jù)點前后時間差，按照首點，尾點，中間連續(xù)點的規(guī)則來處理。如果是首點，該點與上一個點時間跨度為0， 該點與下一個點時間跨度為下一個點時間減去當前點時間；如果是尾點，該點與上一個點時間跨度為當前點時間減去上一個點時間，該點與下一個點時間跨度為0；如果是中間點，該點與上一個點時間跨度為當前點時間減去上一個點時間，該點與下一個點時間跨度為下一個點時間減去當前點時間。

在前面工作的基礎上，根據(jù)數(shù)據(jù)中當前點與中心點的間距和時間跨度，來識別駐留分段和途經(jīng)點。并加標識字段，0為駐留點，1為出行點。在第三階段的最后，要把識別出的途經(jīng)點和駐留點數(shù)據(jù)轉成DataFrame臨時表格。然后讀取第一階段途經(jīng)點的識別數(shù)據(jù)，并整合所有的駐留和途經(jīng)數(shù)據(jù)，按照新的數(shù)據(jù)格式輸出。新的數(shù)據(jù)格式包含imsi用戶惟一標識id，ts時間戳，lac_id，cell_id，tag途經(jīng)或駐留標簽。其中tag字段值0表示駐留，1表示途經(jīng)。結果示例數(shù)據(jù)如圖6所示。

4 MapReduce數(shù)據(jù)處理具體實現(xiàn)

MapReduce是基于Hadoop的數(shù)據(jù)分析和處理的核心組件，它可以將用戶編寫的業(yè)務邏輯代碼和自身組件整合成一個完整的分布式運算處理程序，然后并發(fā)運行在Hadoop集群上。人群密度應用場景的MapReduce實現(xiàn)使用圖4所示的商業(yè)區(qū)數(shù)據(jù)和圖5所示的清洗后時空數(shù)據(jù)。首先將兩份數(shù)據(jù)以經(jīng)緯度為連結字段做連表操作，忽略imsi字段，拼接經(jīng)度和緯度作為鍵，統(tǒng)計本時間段內本經(jīng)緯度點一共出現(xiàn)了多少人。MapReduce編程模型只能包含一個map處理功能和一個reduce處理功能，如果業(yè)務邏輯比較復雜不能一次性處理完，只能編寫多個MapReduce程序，并以串行的方式來執(zhí)行它們。比如將人群聚集密度從高到低排序，需要編寫一個MiddleMapper，其map方法具體代碼如下：

protected void map（LongWritable key， Text value，

Context context） throws IOException，

InterruptedException {

String line=value.toString（）;

String[] arr=line.split（"＼t"）;

if（arr.length==2） {

MyKeyPair myKey=new MyKeyPair（）;

myKey.first=arr[0];

myKey.second=Integer.parseInt（arr[1]）;

context.write（myKey， new IntWritable

（Integer.parseInt（arr[1]）））;

}

}

此處MyKeyPair有兩個字段屬性，經(jīng)緯度和匯總數(shù)量。MyKeyPair按第二個字段匯總數(shù)量作為排序標準，利用Mapper的自動排序功能，即可實現(xiàn)匯總數(shù)量由大到小降序排列。MyKeyPair主要代碼如下：

public class MyKeyPair implements

WritableComparable {

String first;

int second;

@Override

public int compareTo（MyKeyPair o） {

return o.second-second;

}

……

}

對應的MiddleReducer的reduce具體代碼功能比較簡單，此處不再展示。最終結果如圖7所示，數(shù)據(jù)包含兩個字段，第一個字段為經(jīng)緯度，第二個字段為此經(jīng)緯度人群密度數(shù)量。

5 Hive數(shù)據(jù)處理具體實現(xiàn)

由于MapReduce在處理連表操作和數(shù)據(jù)匯總的時候，代碼相對繁瑣，后續(xù)處理我們使用Hive來實現(xiàn)。Hive操作接口采用類似SQL的語法，避免編寫MapReduce代碼。由于Hive的執(zhí)行延遲比較高，Hive優(yōu)勢在于處理大數(shù)據(jù)，對于處理小數(shù)據(jù)沒有優(yōu)勢，因此 Hive適用于數(shù)據(jù)分析且對實時性要求不高的場合。我們將圖4和圖7所展示的數(shù)據(jù)存入Hive兩張新建的外部文本文件表中。圖4所示數(shù)據(jù)表名和字段命名為region（la string， lo string， lname string），圖7所示數(shù)據(jù)表名和字段命名為lac_cnt（lac string， cnt int），則統(tǒng)計每個區(qū)域人群密度的匯總HQL語句如下：

select r.lname， sum（l.cnt）

from lac_cnt l

inner join region r on l.lac=concat（r.la，'-'，r.lo）

group by r.lname;

該HQL語句以經(jīng)緯度點為連表條件，得到每個經(jīng)緯度點的人群數(shù)量。再以區(qū)域名稱為分組字段，通過sum（）聚合函數(shù)統(tǒng)計出每個區(qū)域的人群數(shù)量。

為了處理圖6駐留或途經(jīng)分析結果數(shù)據(jù)，和前面一樣，我們將其存入Hive新建的外部文本文件表中，數(shù)據(jù)表名和字段命名為stay_pass（imsi，ts，lac_id，cell_id，tag）。需要關聯(lián)的基站對應區(qū)域數(shù)據(jù)表名和字段命名為area_lac（aname， lac）。則統(tǒng)計每個地區(qū)人群駐留或途經(jīng)的匯總HQL語句如下：

select a.aname， s.tag， count（s.tag）

from area_lac a

left outer join （select imsi， lac_id， cell_id， tag from

stay_pass group by imsi， lac_id， cell_id， tag） as s

on a.lac=concat（s.lac_id，'-'，s.cell_id）

group by a.aname， s.tag;

該HQL語句先做stay_pass表上的分組操作去除同一個人的重復數(shù)據(jù)。然后和area_lac做左外連表操作，并以區(qū)域字段和駐留或途經(jīng)狀態(tài)字段分組，通過count（）聚合函數(shù)得到每個區(qū)域的留或途經(jīng)人數(shù)。注意Hive外聯(lián)結操作不會生成空值數(shù)據(jù)，如果要保證結果數(shù)據(jù)的完整性需要使用Python或Scala腳本處理結果數(shù)據(jù)并補齊缺失的空值和0值數(shù)據(jù)。

前述結果可以保存到文件或者Hive數(shù)據(jù)表庫中，讀取并展示的過程同普通Web項目類似。通過SpringMVC讀取文件或數(shù)據(jù)庫中的結果數(shù)據(jù)，轉換成Echarts可以直接使用的數(shù)據(jù)形式后，返回給Axios發(fā)起的數(shù)據(jù)請求。Axios收到數(shù)據(jù)后，交給集成在VUE中的Echarts組件直接顯示即可。為了獲取實時更新的數(shù)據(jù)，可以設置crontab定時任務定期生成Hive結果數(shù)據(jù)。然后用WebSocket推送數(shù)據(jù)到前端頁面，展示實時更新的數(shù)據(jù)。

6 結束語

本文通過具體的Haddop應用和編程實踐操作，演示了時空大數(shù)據(jù)的導入存儲、處理轉換、分析以及結果展示，表明HDFS+MapReduce+Hive在集群計算平臺上提供時空大數(shù)據(jù)的高階應用可以有很強的實用性。結合Python、Scala、SpringMVC、VUE、Axios、Echarts等其他應用項目開發(fā)技術，文本完整的演示了大數(shù)據(jù)應用框架項目的開發(fā)流程。項目最終展示結果可為區(qū)域人群交通優(yōu)化政策的制定和執(zhí)行提供科學客觀的數(shù)據(jù)支持。另外，該項目的應用場景還可以進一步擴充，包括通勤線路分析、出行目的分析，出行方式分析等等。隨著智慧城市建設的不斷深入，利用種類越來越多、體積越來越大、覆蓋面越來越廣的海量時空數(shù)據(jù)來高效管理智慧城市有著非常廣闊的應用前景。

參考文獻（References）：

[1] 馮明翔，羅小芹. 基于時空大數(shù)據(jù)的公共交通關鍵指標計算方法[C]//中國城市規(guī)劃學會城市交通規(guī)劃學術委員會.綠色.智慧.融合——2021/2022年中國城市交通規(guī)劃年會論文集，2022：11

[2] 馮興杰，王文超.Hadoop與Spark應用場景研究[J].計算機應用研究，2018，35（9）：2561-2566

[3] Bawankule， Kamalakant L， Dewang， et al. Historical data based approach to mitigate stragglers from the Reduce phase of MapReduce in a heterogeneous Hadoop cluster[J]. Cluster Computing，2022：1-19

[4] Awad F H， Hamad M M. Improved k-Means Clustering Algorithm for Big Data Based on Distributed Smartphone Neural Engine Processor[J]. Electronics，2022，11（6）：883-883

[5] Ibrahim O A， Keller J M， Bezdek J C，et al.Evaluating Evolving Structure in Streaming Data With Modified Dunn's Indices[J]. IEEE Transactions on Emerging Topics in Computational Intelligence，2019，5（2）：1-12

[6] 尹喬，魏占辰，黃秋蘭，等.Hadoop海量數(shù)據(jù)遷移系統(tǒng)開發(fā)及應用[J].計算機工程與應用，2019，55（13）：66-71

[7] 劉莉萍，章新友，牛曉錄，等.基于Spark的并行關聯(lián)規(guī)則挖掘算法研究綜述[J].計算機工程與應用，2019， 55（9）：1-9

[8] Wei J S. Geographic spatiotemporal big data correlation analysis via the Hilbert-Huang transformation[J]. Journal of Computer and System Sciences，2017，89：130-141

[9] Nigro L. Performance of Parallel K-Means Algorithms in Java[J]. Algorithms，2022，15（4）：117-117

[10] Vara N. Application of k-means clustering algorithm to improve effectiveness of the results recommended by journal recommender system[J]. Scientometrics，2022，127（6）：3237-3252

*基金項目：2021年度湖北省教育廳科學研究計劃指導性項目（B2021423）； 宜昌市2019年應用基礎研究項目（A19-302-14）

作者簡介：鄭曉東（1982-），男，湖北宜昌人，碩士，三峽大學科技學院專任教師，主要研究方向：智能化信息處理、計算機網(wǎng)絡與應用。