@Rays 2017-03-20T14:29:09.000000Z 字数 6878 阅读 2764

使用WSO2复杂事件处理器处理用户轨迹流

未分类

摘要： 如何从不准确性数据流中抽取有用信息，这一直是物联网应用流数据处理中的一个重要问题。本文给出了一种使用卡尔曼滤波器平滑iBeacon传感器网络中采集的用户轨迹信息的方法，并展示了该方法的有效性。这一方法已经构建到WSO2的复杂事件处理器中。

作者： Ramindu De Silva, Miyuru Dayarathna

审校： Sergio De Simone

正文：

本文要点

卡尔曼滤波可用于对混杂了噪声的用户轨迹做平滑。

从传感器数据构建的轨迹与实际值的偏差很大，需要使用轨迹平滑和重新排列技术做修正。

在真实的应用中，无序事件处理器需要与卡尔曼滤波器结合，以确保输入卡尔曼滤波的事件流是按时间排序的。

轨迹尾部的多个波动会导致最终估计值出现显著错误。

用户的位置信息和移动速度可用于卡尔曼滤波算法，能提高算法的准确性。

序言

如何从不确定性数据流（即一系列携带了不准确信息的数据项）中抽取有用信息，这一直是流数据处理领域的一个重要问题，该问题具有广泛的应用场景。在本文中，我们将展示一个从交通和物流领域不确定性数据流中合成有用信息的应用。由于各种因素（例如探测器功能异常、环境噪声等），探测器网络所采集的探测器数据大部分是不准确的，因此对此类数据进行滤波处理是非常重要的。确切地说，我们给出了一种在WSO2的复杂事件处理（CEP，complex event processing）中使用卡尔曼滤波器的方法，用于平滑iBeacon探测器网络采集的用户轨迹信息。通过对比原始用户轨迹和卡尔曼滤波后的结果，我们展示了该方法的有效性。本文中使用了物联网（Internet of Things，IoT）技术中的一个关键实例应用作为用例。本文提出的方法已实现为一个复杂事件处理中间件，作为WSO2综合企业中间件平台的组成部分，随WSO2 CEP一起发布。

问题描述

下面定义的应用场景将在本文中贯穿始终。一个用户携带了iBeacon传感器（例如智能手机），行走在一个iBeacon场中。iBeacon传感器接收来自各个iBeacon传输的信号。用户在某一时刻的大概位置是由iBeacon传感器检测到的iBeacon位置做三角定位得到（参见图1）。但是这样的三角定位所获取的位置信息 $P(x,y)$ 只是一个粗略的近似。导致位置不确定性的原因很多，例如不同的iBeacon发出的信号被具有相似信号强度的iBeacon传感器在不同时刻接收、iBeacon传感器采样频率低、事件无序到达等。如果iBeacon传感器以低采样频率运行，将无法准确地捕获用户在轨迹上的突然改变。

图1：一位携带iBeacon传感器的用户步行通过一个iBeacon场。从iBeacon发出的信号被传感器接受，用于构建大致的定位信息。

如果绘制用户的三角定位路径，所创建的大致轨迹会与用户的原始轨迹会有明显的偏差。需要使用轨迹平衡和重新排列技术对用户轨迹进行修正。

解决方法概述

我们使用了两种技术解决上述问题。第一，我们使用了一种无序处理组件，在输入事件流中引入次序。第二，我们实现了一种卡尔曼滤波器组件，可以在给定前期的状态和当前的三角定位的情况下预测用户的位置。下面我们将分别介绍无序事件处理和卡尔曼滤波器的实现机制。

无序处理

流数据处理的应用一般存在无序事件到达，数据流中元组间的无序问题（即缺少次序）是由于网络延迟、操作符并行、异步流归并等因素导致。实现无序处理（也称为事件重新排列）目前有四种主要的技术，分别为“基于缓冲区”、“基于标点”、“基于推测”和“基于近似”。本文所提出的重排序解决方法基于k-slack算法，主要思想是在将输入流中的元组提供给查询操作符之前，使用缓冲区对元组按时间戳递增的次序排序。需要指出的是，即使使用了缓冲区做事件排序，输出中依然存在着无序的事件。导致原因有多种，例如显著延迟的事件。

重排序的数据流进而被发送给位置平均模块，获取平均后的位置。

位置平均

一个iBeacon传感器会在一个点上探测到多个iBeacon。当它从某个iBeacon附近离开时，会停止接受iBeacon的ping信号。平均方法将在考虑所有这些可用信息的情况下对用户做三角定位。

包含平均位置的数据流会提供给卡尔曼滤波器用于进行轨迹平滑。接下来，我们将对卡尔曼滤波做简要介绍，进而说明我们的方法是如何应用卡尔曼滤波的。

基于卡尔曼滤波的轨迹平滑

如何在混杂有随机噪声的条件下检测给定类型的信号，这是一个已经得到广泛研究的问题，最早可回溯到上世纪五十年代。卡尔曼滤波器是一种基于线性二次问题的估计方法，它提供了一系列方程式，可以以最小化均方误差的方式高效计算（递归）处理状态估计值。滤波器对受白噪声干扰的线性动态系统的瞬时状态做出估计。在不知道系统当前准确行为的情况下，滤波器也可以估计过去、当前和未来的状态。卡尔曼滤波器的典型应用场景包括噪声数据平滑和提供感兴趣参数的估计值。卡尔曼滤波器具有大量的应用，例如：自动和辅助驾驶、笔记本电脑触控板输出的平滑处理、全球定位系统（GPS）接收器等。

由于大量的轨迹数据的出现，轨迹平滑已成为当前研究所关注的一个重要问题。在对可视动作进行跟踪和预测时，使用卡尔曼滤波器是一种首选解决方法。因为它只保持前一个状态，因此使用的内存资源非常少。它使用递归运算，速度很快，适合解决实时问题，新的输入值可以马上得到处理。

我们提出的方法

我们将在本节中给出使用卡尔曼滤波器平滑轨迹的具体实现细节。目前有两种用户位置估计技术。如果用户是静止的，我们可以只使用用户轨迹的位置信息。如果用户是移动的，我们可以在卡尔曼滤波器算法中使用用户的位置信息和速度信息。在本文中，我们所面对的问题是移动用户轨迹的平滑，因此采用的是后一种技术。此外，第二种技术也更为精确，因为它在位置估计过程中还使用了实验者的前期速度作为参数。但是为了更清楚地说明我们的方法，下面将会对两种技术都给出解释。表1中列出了本文所使用的参数定义。中间的一列显示了所使用的矩阵维度，我们在卡尔曼滤波器实现中实际使用的矩阵维度显示在括号中。

表1：本文使用的参数定义。

变量命名	维度（实际值）	变量描述
$Z^k$	$N×1$ （2×1）	在时间 $k$ 处的测量值。
$X^k$	$N×1$ （2×1）	对测量值使用卡尔曼滤波器所得到的估计值。
$R^k$	$M×M$ （2×2）	不确定性的协方差（即传感器噪声的协方差） $R^k$ 。在我们的方法中并未使用。
$A^k$	$N×N$ （2×2）	转移矩阵（状态增益）。
$P^k$	$N×N$ （2×2）	估计值 $X^k$ 的协方差矩阵。
$Q^k$	$N×N$ （2×2）	使用协方差 $Q^k$ 将 $X^{k−1}$ 中各点移动到高斯块中（在我们的方法中并未使用）。
$H$	$N×N$ （2×2）	由于读取的单位和尺度可能并非同我们所跟踪的是一致的，使用这个矩阵目的在于将其读取值缩放到测量的规模。
$K$	$N×M$ （2×2）	卡尔曼增益矩阵。

静态卡尔曼滤波器模型

基本的卡尔曼滤波器是基于静态模型实现的，我们将位置作为系统中的唯一变量。我们假定在这样的场景中的测量变量有一个不变值，我们需要估计这个不变值。静态模型的类似应用还包括电压测量、静止物体位置测定等。如图2所示，通过使用卡尔曼滤波器，我们基于观测读数获得了“可能的真实读数”。这意味着卡尔曼滤波器的输出更接近用户的实际位置，虽然并不保证该值会是准确的用户位置。

图2：卡尔曼滤波器的期望输出。我们需要从被探测器噪声干扰的观测读数中找到可能的真实读数。

卡尔曼滤波器静态模型的主要方程为公式(1)。

$X’^k = X^{k-1} + K^{k-1} ( Z^k - H*X^{k-1}) \tag{1}$

预测

预测过程将对下一个时间 $k$ 处的状态做预测。

$X^k = A*X^{k-1} \tag{2}$

$P^{k} = A*P^{k-1}AT \tag{3}$

其中是 $X^k$ 是随机分布的中心，表示最可能的状态，不确定度表示为方差 $P^k$ 。

更新

更新过程对卡尔曼增益矩阵进行更新。

$S = H*P^k*H^T+R \tag{4}$

$K’k = P^k*H^T*S-1 \tag{5}$

$X’^k = X^k + K’^k (Z^k - H*X^k) \tag{6}$

$P’z^k = P^k - K’^k*H*P^k \tag{7}$

其中，

$A = 1$

$X = 前期预测值$ 。前期预测值假定为最初测量值。

$P = [1000\ 0]$

$H = 1$

$R =噪声的标准偏差$ 。本文中我们使用 $噪声的标准偏差 = 0.01$

公式(5)中的 $K’^k$ 对应于更新的卡尔曼滤波器增益矩阵，该矩阵使用前期估计值的协方差矩阵计算得到。这样我们就可以计算最优估计值 $X’^k$ （即可能的真实读数），及最优估计值的协方差矩阵 $P’^k$ 。在下一轮预测或更新计算中，这些值再反馈输入给卡尔曼滤波器，直至达到所需的循环次数。注意表1中虽然列出了 $R^k$ （测量噪声的协方差矩阵）和 $Q^k$ （过程噪声的协方差矩阵）这两个矩阵，但是在我们当前实现的卡尔曼滤波器方法中并未使用它们。

动态卡尔曼滤波模型

在本文的场景中，我们假定测量变量的值会随时间发生改变。移动对象的定位就是一个很好的例子。动态卡尔曼滤波器模型的主函数与上面的公式(1)一样，但是矩阵中使用的参数做了如下更改：

$A = [1\ 时间差; 0\ 1]$

一开始，我们假定时间差为0，

$X = [前期预测值;\ 变化率]$

并假定上一个预测值为最初测量值。

$P = [1000\ 0; 0\ 1000]$

$H = [1\ 0; 0\ 1]$

$R = [噪声的标准偏差\ 0;\ 0\ 噪声的标准偏差]$

$噪声的标准偏差 = 0.01$

系统的实现

我们使用WSO2 CEP完全实现了上面提出的用户轨迹平滑方法，并采用真实的轨迹数据集做了应用测试。注意，本文给出的方法也可以在其它一些最新的事件处理系统中实现。系统高层架构如图3所示，图中描述了方法的整体实现。从iBeacon传感器读取的位置事件被WSO2的CEP接收器所接收。在WSO2 CEP的组成中，有一个称为Siddhi的事件处理库。虽然大部分通用事件处理函数（例如投影、聚合、窗口、连接等）可以使用Siddhi的SiddhiQL实现，但是一些更为特定的处理需要使用定制的扩展程序去编程实现。我们使用一种称为“reorder”的扩展程序，reorder可以处理数据流中存在的无序情况。在图3中，使用了Siddhi的扩展程序reorder对事件流做了重排序。

图3：整体系统结构。

此后，我们计算平均定位并发送给卡尔曼滤波器做轨迹信息平滑。最终，平滑后的轨迹信息发布在一个称为“Geo Dashboard”的外部应用上。借助于地理空间技术，这样的轨迹信息可用在计算特定地区的等待时间上。这里，等待时间是指穿行特定区域所需花费的时间。列表1中给出的Siddhi查询显示了该流水线的实现方式。

--通过event接收者检索事件。
.
.
--将beacon的ping数据流转换为通用格式。
.
.
@info( name = 'reorderPersonLocatorStream')
from beaconPingStream#reorder:kslack(timestamp, 120000l, -1l, true) 
select recordLocator, uuid, timestamp, beaconProximity, fenceAirportCode, 
       fenceIdentifier, locationEventTriggerType, locationEventType,
       mileagePlusNumber 
insert into sortedEditedPingStream;
.
.
@info( name = 'averagingBeaconLocationsPerPerson')
from sortedEditedPingStream#custom:averageLocationCalculate(recordLocator,
                                    latitude,longitude, beaconProximity, uuid,
                                    floor, false, airport)
select *
insert into totalAveragedPingStream;
.
.
@info( name = 'smoothingLatitudeAndLatitude')
from totalAveragedPingStream#kf:kalmanFilter(averagedLatitude, averagedLongitude, 
                                 timestamp, recordLocator, floor)
select kalmonFilteredStream.recordLocator as id, 
       kalmonFilteredStream.timestamp as timeStamp, kalmanLatitude as latitude, 
       kalmanLongitude as longitude, "CUSTOMER" as type, floor, 
       beaconType, mileagePlusNumber, beaconName, airport, terminal
insert into personLocationStream;
.
.
--发布用户位置到Geo Dashboard。
.
.
@info( name = 'calculateSecurityAreaWaitTime')
from personLocationStream#custom:updateSecurityWaitTime(latitude, longitude, 
                                     floor, id, timeStamp, true, airportCode, 
                                     terminal)
select id, timeStamp, latitude, longitude, type, speed, heading, floor, geoFence,  
       averageWaitingTime, mileagePlusNumber, geoFenceLat, geoFenceLong, 
       airportCode
insert into securityWaitTimeStream;
.
.
--将更新后的安全等待时间发布给Geo Dashboard。
.

列表1：我们方法的一个Siddhi代码段例子，显示了轨迹平滑过程的执行机制。

实验

我们对一个用户的轨迹数据集执行了我们的卡尔曼滤波器方法，数据集中轨迹数据是从真实的传感器网络环境中获取的，其中的传感器网络由约20个iBeacon传感器组成，安装在一个公开地点，延展为480米。数据集中包含了大约9300个事件。数据集被重排序并使用了上述的卡尔曼滤波器。图4显示了我们绘制的原始轨迹和卡尔曼滤波后的轨迹。从图中可以观察到，原始探测器读数相对而言不够平滑，路径（显示为蓝色）具有相当数量的波动，而由卡尔曼滤波器计算得到的路径（显示为绿色）更为平滑。

图4：一个使用卡尔曼滤波器技术平滑的用户轨迹的例子。其中红色点显示了用户的三角定位信息，绿色十字显示了使用卡尔曼滤波器平滑并重新排列后的轨迹。

图5显示了对Geo Dashboard的几个截图，截图展示了我们的轨迹平滑技术的有效性。

图5：一个例子，展示了我们的方法在用户轨迹数据流处理上的有效性。显示在Geo Dashboard中的两个轨迹是一个用户在机场的移动情况。(a)截图仅记录了原始轨迹信息和三角定位信息；(b)截图是使用我们的卡尔曼滤波器方法处理后的轨迹。

在轨迹终止处的多个波动导致了一些在最终估计值中可观察到的错误。如果在轨迹开始时有多处波动，卡尔曼滤波器可以成功地平滑路径。但是如果在轨迹终止时存在多个波动，即使最终事件给出了轨迹的正确位置信息，滤波器也会将这些看成是噪声数据，使得对最后位置的平滑导致了估计值的错误。

总结

很多现实问题需要使用噪声测量去估计一个时变系统的状态。在本文中，我们给出一个实例，即如何使用卡尔曼滤波器修正混杂了噪声的用户轨迹。在此过程中，我们首先重排序可能存在无序抵达的事件，然后使用卡尔曼滤波器估计真实的用户轨迹。我们实现了两种卡尔曼滤波器方法，分别称作静态模型和动态模型。通过同处绘制从输入流接收到的原始轨迹和修正后轨迹，我们验证了修正的质量。在实现中，我们将卡尔曼滤波器和无序处理实现为一个WSO2 Siddhi扩展程序，并分别以Apache 2.0开源许可在【1】和【2】处发布。

本文作者简介

Ramindu De Silva是一位软件工程师，就职于WSO2。他主要从事WSO2的复杂事件处理器实现。他是斯里兰卡Sri Jayewardenepura大学的计算机科学专业硕士，具有马来西亚APIIT的信息及通信技术（ICT）学位。

Miyuru Dayarathna是WSO2的高级技术主管。他是一名计算机科学家，在流计算、图数据管理和挖掘、云计算、性能工程和物联网等研究领域颇有贡献。他还是斯里兰卡Moratuwa大学计算机科学与工程系的顾问，在知名国际杂志和会议上发表了一些技术论文。

查看英文原文：Processing Streaming Human Trajectories with WSO2 CEP