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

本文要点

• 卡尔曼滤波可用于对混杂了噪声的用户轨迹做平滑。
• 从传感器数据构建的轨迹与实际值的偏差很大，需要使用轨迹平滑和重新排列技术做修正。
• 在真实的应用中，无序事件处理器需要与卡尔曼滤波器结合，以确保输入卡尔曼滤波的事件流是按时间排序的。
• 轨迹尾部的多个波动会导致最终估计值出现显著错误。
• 用户的位置信息和移动速度可用于卡尔曼滤波算法，能提高算法的准确性。

序言

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

问题描述

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

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

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

解决方法概述

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

无序处理

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

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

位置平均

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

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

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

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

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

我们提出的方法

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

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

静态卡尔曼滤波器模型

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

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

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

动态卡尔曼滤波模型

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

系统的实现

我们使用 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

感谢薛命灯对本文的审校。

给InfoQ 中文站投稿或者参与内容翻译工作，请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博（ @InfoQ ， @丁晓昀），微信（微信号： InfoQChina ）关注我们。