随着大数据时代的到来，传统的静态数据的采集、处理和存储技术变得不再适合流式数据计算。面对流式数据的相应特点，相应的计算和应用的问题随之而来，这些问题也获得众多学者和开发人员的关注。

实时准确地发现物流热点区域，可以帮助决策用户获取实时的决策信息。同时，对物流热点区域的历史演化分析可以帮助企业更加准确地掌握热点动态。本节介绍如何使用CluStream算法对交通运输物流热点区域进行分析，并使用Web-GIS在地图上实现展示聚类结果的热力图。使用Apache Kafka和Apache Storm进行数据收集和分析，使用Apache HBase存储最终结果，最终使用百度地图生成热力图。系统结构见图4.4。

图4.4　基于分布式流式数据算法的应用

1.流式数据的收集

编程模拟两种类型流式数据：一种为运营车辆产生的流式数据，相应数据属性包括：车牌号、经纬度、海拔高度、方向、速度、油耗、灯光等；另一种为客户产生的流式数据，数据属性包括：经纬度、货物类型等。利用Apache Kafka的TOPIC配置项将两类流式数据进行区分。

2.流式数据的预处理

流式数据预处理的过程包括数据采集、数据格式的转换和数据属性的选取，如图4.5所示。

图4.5　数据的预处理

在数据预处理的过程中，按照通信协议对数据进行格式转换，最后选择需要进行分析的数据属性，减少后续计算的数据量。

3.流式数据的分析

CluStream算法提出了适用于流式数据计算的定义：

定义2　快照是在规定时间点需要被存储的微簇。

定义3　快照的阶将时间轴划分成不同粒度的时刻，其范围取值为1到logα（T），α为整数且α≥1，T为从数据流开始到现在的时间。

定义4　倒金字塔时间衰减结构是一种微簇信息的存储策略，即微簇信息是否、何时被存储，它保证了只需使用较少的空间就可以存储快照。当满足下面条件时，不同阶对应的快照会被系统存储：

条件1　在时间间隔αi（α为整数且α≥1）内出现的第i阶快照，当其时间值的平方能被αi整除时，系统会对每一张快照进行存储。

条件2　在任意时刻，第i阶对应的快照只存储其中最后αl+1个快照。

根据快照的定义和存储条件我们可以得到：

推论1　对于一个数据流，在T时刻，快照的最大阶数为logα（T）。

推论2　对于一个数据流，在T时刻，最多存储的快照数量为（αl+1）logα（T）。

当T=55，α=2，l=2时，则在第0阶和第1阶需要被存储的快照对应的时刻如图4.6所示，两个时间轴为同一轴。

图4.6　第0阶和第1阶被存储的快照对应的时刻

第0阶需要被存储的快照对应的时刻为55、54、53、52、51，时间间隔为20=1；第0阶需要被存储的快照对应的时刻为54、52、50、48、46，时间间隔为21=2。同时从图4.6中可以看到，第0阶和第1阶需要被存储的快照对应的时刻的时间间隔为1和2，因此不同阶需要被存储的快照对应的时刻的时间间隔不同，即粒度不同。以此类推，并且每个时刻的快照只需要存储一次，T=55时刻，应存储的快照时刻如表4.3所示。

表4.3　T=55时的快照存储表

1）在线部分

算法的在线部分不需要用户输入查询时间窗，它的目标是为了让离线部分更高效地使用其统计信息。相应的过程如下：

（1）初始化微簇。

假设算法在任意时刻总共维护q个微簇，M1，…，Mq表示某一时刻的全部q个微簇，在数据流的开始时刻，在磁盘上存储初始的InitNumber个数据点，使用离线的标准K-means聚类算法形成q个初始微簇，其中q＜InitNumber。(www.daowen.com)

（2）在线处理。

对于新到达的点Xi，Xi会被某个簇吸收或者形成新簇。首先，计算Xi到M1，…，Mq微簇中心的距离，其中dist（Mj，Xik）表示点Xi到微簇Mj中心的距离。找到离Xi最近的微簇Mp，并将点Xi置于微簇Mp中，如图4.7所示。

以下情况，Xi不属于任何已存在的微簇：

图4.7　点属于最近的微簇

情况1　Xi在Mp的边界外，如图4.8所示。

图4.8　点落在最近微簇边界外

情况2　由于数据流的不断演化，Xi是一个新簇。

针对情况1，算法为Xi创建一个独有ID的新簇，因此，同时要减少一个已经存在的微簇。为此，算法需要删除一个最早的簇或者合并两个最早的簇。可以有两种方法来确定需要删除的或合并的微簇。

第一种方法，在一个给定的微簇M中，μM和σM分别表示点到达时间的平均值和标准差。计算公式如下：

其中，Ti为微簇中点到达的时间戳，为平均时间，则根据微簇的定义可以使用微簇存储的指标来获得时间戳的平均值和标准差，假设时间戳是正态分布的，则微簇M的时间戳分布为N（μM，σM 2），可以选取δ=μM+int×σM值最小的微簇进行删除。其中int值可以为（1，2，3）。当int=3时，根据正态分布的特征，可以说明，微簇中99%左右的点是在δ之前出现的，δ值越小，说明该微簇越旧。

第二种方法是为每个微簇保存最后m个数据点的到达时间戳，假设，在微簇M的点中找到到达时间的第m/（2×n）百分位，用其估计每一个簇中最后m个到达的数据点的平均时间戳δ，删除带平均时间戳最小的或小于用户定义的阈值的微簇。

以上两种方法都可以用于选取较旧的微簇进行删除，第一种方法与第二种方法相比，不需要记录每个微簇的最后m个数据点的时间戳，节省了存储空间，但是其需要建立在正态分布假设基础之上，对于其他形式的时间分布特征并不适用。除了使用上述方法外，还需要为新到达的点Xi生成一个具有唯一ID的微簇，新微簇的半径为到达最近微簇中心点的距离，如图4.9所示。

图4.9　删除最早的簇

特殊情况：如果是通过指定阈值而不是选取δ最小值来删除微簇，则有可能出现多个符合条件的待删除微簇（如图4.10所示），此时算法需要选取并合并某两个靠得最近的微簇。此时用它们原来的ID一起来标志这个新的微簇。

在线部分的算法流程如图4.11所示。

2）离线部分

在算法的离线部分，用户可以在不同时间幅度内发现簇，进行高层聚类分析。离线部分所用的数据是在线部分形成的微簇的统计信息，因此可以满足内存有限的需求。离线部分的计算将整个微簇抽象成一个伪数据点，因此传统的K-means算法需要进行一下修改：在初始阶段算法将微簇的中心抽象成种子。同时，种子被选中概率与微簇中点的个数成正比，不再按照随机的方式进行；在划分阶段，一个种子到一个伪数据点的距离就等于它到伪数据点所代表微簇的中心的距离。离线部分的算法流程图如图4.12所示。

图4.10　合并两个微簇

图4.11　在线部分的算法流程

基于Storm的CluStream算法热点区域分析由三个Bolt组件组成，分别是KafkaSpout、CFBolt和ClusteringBolt。如图4.13所示，KafkaSpout读取Kafka消息队列中的数据，并将数据发射给CFBolt组件；CFBolt接收到数据后，首先会选取数据中的经度和纬度属性，然后对数据进行格式转换，转换完毕后进行CluStream在线部分计算，更新微簇信息，然后根据时间衰减结构将需要存储的快照进行缓存，最后会将微簇信息发射给ClusteringBolt组件；ClusteringBolt调用CluStream算法对微簇信息进行聚类分析并将结果保存到数据库进行存储。

图4.12　离线部分的算法流程图

图4.13　基于Storm的CluStream算法热点区域分析Topo图

最后，系统从数据库读取数据并将聚类得到的结果与热力图进行映射，直观地反映出交通运输物流活动中的热点区域，区域中心点采用类的中心点，热力值为类中点的个数。实验得到的某一时刻实时热力图如图4.14所示。