時空大數據分析在人群聚集統計中的應用

鄭曉東 鄭業爽 宋思琪

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

關鍵詞： 人群聚集； 時空大數據； kmeans算法； 鄧恩指數； 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 引言

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

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

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

1 Hadoop框架的分析模型業務架構

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

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

2 時空大數據的導入和清洗

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

本模型主要基于以下原始業務信息數據來完成分析：通信信息數據，基站經緯度數據和商業區數據。原始通信信息示例數據如圖2所示。

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

基站經緯度數據包含字段longitude經度，latitude緯度，laci基站信息。商業區數據包含字段longitude經度，latitude緯度，area_name商業區名稱。

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

3 時空大數據的數據建模分析處理

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

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

⑴ 將統計好的數據轉成list列表并遍歷這個列表。

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

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

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

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

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

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

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

key：（imsi-cluster）

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

cluster_latitude，nid）

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

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

4 MapReduce數據處理具體實現

MapReduce是基于Hadoop的數據分析和處理的核心組件，它可以將用戶編寫的業務邏輯代碼和自身組件整合成一個完整的分布式運算處理程序，然后并發運行在Hadoop集群上。人群密度應用場景的MapReduce實現使用圖4所示的商業區數據和圖5所示的清洗后時空數據。首先將兩份數據以經緯度為連結字段做連表操作，忽略imsi字段，拼接經度和緯度作為鍵，統計本時間段內本經緯度點一共出現了多少人。MapReduce編程模型只能包含一個map處理功能和一個reduce處理功能，如果業務邏輯比較復雜不能一次性處理完，只能編寫多個MapReduce程序，并以串行的方式來執行它們。比如將人群聚集密度從高到低排序，需要編寫一個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有兩個字段屬性，經緯度和匯總數量。MyKeyPair按第二個字段匯總數量作為排序標準，利用Mapper的自動排序功能，即可實現匯總數量由大到小降序排列。MyKeyPair主要代碼如下：

public class MyKeyPair implements

WritableComparable {

String first;

int second;

@Override

public int compareTo（MyKeyPair o） {

return o.second-second;

}

……

}

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

5 Hive數據處理具體實現

由于MapReduce在處理連表操作和數據匯總的時候，代碼相對繁瑣，后續處理我們使用Hive來實現。Hive操作接口采用類似SQL的語法，避免編寫MapReduce代碼。由于Hive的執行延遲比較高，Hive優勢在于處理大數據，對于處理小數據沒有優勢，因此 Hive適用于數據分析且對實時性要求不高的場合。我們將圖4和圖7所展示的數據存入Hive兩張新建的外部文本文件表中。圖4所示數據表名和字段命名為region（la string， lo string， lname string），圖7所示數據表名和字段命名為lac_cnt（lac string， cnt int），則統計每個區域人群密度的匯總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語句以經緯度點為連表條件，得到每個經緯度點的人群數量。再以區域名稱為分組字段，通過sum（）聚合函數統計出每個區域的人群數量。

為了處理圖6駐留或途經分析結果數據，和前面一樣，我們將其存入Hive新建的外部文本文件表中，數據表名和字段命名為stay_pass（imsi，ts，lac_id，cell_id，tag）。需要關聯的基站對應區域數據表名和字段命名為area_lac（aname， lac）。則統計每個地區人群駐留或途經的匯總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表上的分組操作去除同一個人的重復數據。然后和area_lac做左外連表操作，并以區域字段和駐留或途經狀態字段分組，通過count（）聚合函數得到每個區域的留或途經人數。注意Hive外聯結操作不會生成空值數據，如果要保證結果數據的完整性需要使用Python或Scala腳本處理結果數據并補齊缺失的空值和0值數據。

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

6 結束語

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

參考文獻（References）：

[1] 馮明翔，羅小芹. 基于時空大數據的公共交通關鍵指標計算方法[C]//中國城市規劃學會城市交通規劃學術委員會.綠色.智慧.融合——2021/2022年中國城市交通規劃年會論文集，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海量數據遷移系統開發及應用[J].計算機工程與應用，2019，55（13）：66-71

[7] 劉莉萍，章新友，牛曉錄，等.基于Spark的并行關聯規則挖掘算法研究綜述[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-），男，湖北宜昌人，碩士，三峽大學科技學院專任教師，主要研究方向：智能化信息處理、計算機網絡與應用。