【原创】StreamInsight查询系列(二十一)——查询模式之使用地理数据

上篇文章介绍了查询模式中如何检测间隙事件,这篇博文将介绍StreamInsight中如何使用地理数据。

测试数据准备

为了方便测试查询,我们首先准备一个静态的测试数据源:

// 创建包含地理位置的引用数据
var assetData = new []
{
    new { AssetId = 1, AssetName = "My Stuff",          Latitude = 47.64339, Longitude = -122.12840 },
    new { AssetId = 2, AssetName = "My Other Stuff", Latitude = 47.64827, Longitude = -122.13171 } 
};

var now = DateTime.Parse("09/13/2011 11:13:00 PM");

var liveData = new []
{
    new { Timestamp = now.AddMinutes(1),  AssetId = 1, State = "Green", Id = 1 },
    new { Timestamp = now.AddMinutes(2),  AssetId = 2, State = "Red", Id = 2 },
    new { Timestamp = now.AddMinutes(3),  AssetId = 1, State = "Green", Id = 3 },
    new { Timestamp = now.AddMinutes(4),  AssetId = 2, State = "Green", Id = 4  }    
};

其中assetData用作引用数据,而liveData为假定的实时数据。接下去我们将上述数据源转变为点类型复杂事件流:

// 将asset数据转变为引用事件流
var assetStream = assetData.ToPointStream(Application, t =>
    PointEvent.CreateInsert(now.ToLocalTime(), t),
    AdvanceTimeSettings.IncreasingStartTime);

// 延伸引用事件流中事件生命周期至7天
var assetRef = assetStream
    .AlterEventDuration(e => TimeSpan.FromDays(7));

// 将实时数据转变为实时事件流
var liveStream = liveData.ToPointStream(Application, t =>
    PointEvent.CreateInsert(t.Timestamp.ToLocalTime(), t),
    AdvanceTimeSettings.IncreasingStartTime);

使用地理数据

问题1:怎样在StreamInsight查询中使用SQL Server地理库(SQL Server Geography Library)?

要想在Streaminsight查询中使用SQL Server地理库,首先需要引用Microsoft.SqlServer.Types.dll 程序集。这个程序集最早出现在Sql Server 2008中,读者可以在GAC或C:\Program Files (x86)\Microsoft SQL Server\100\SDK\Assemblies\处找到它(注意前提是你已经安装了SQL Server或至少安装了SQL Server管理服务器)。如果实在找不到该程序集,读者可以考虑从NuGet中下载安装这个包。引用完这个程序集后,你就可以正常使用 Microsoft.SqlServer.Types 命名空间下的类了(SqlGeography, SqlGeometry等)。

下面谈谈在LINQPad中如何使用SQL Server地理库:

  1. 右键单击查询窗口,选择“Query Properties”;
  2. 在“Additional References”中点击“Add…”添加SQL地理库程序集Microsoft.SqlServer.Types.dll,如下:
    【原创】StreamInsight查询系列(二十一)——查询模式之使用地理数据
  3. 切换到“Additional Namespace Imports”,添加命名空间Microsoft.SqlServer.Types,如下:
    【原创】StreamInsight查询系列(二十一)——查询模式之使用地理数据
  4. 在LINQPad中执行下述语句:
    const int SRID = 4326;
    SqlGeometry.Point(0, 0, 4326).STDistance(SqlGeometry.Point(1, 1, SRID)).Dump();
    
    可以在结果窗口中看到1.4142135623731,这就意味着可以使用SQL Server地理库了。(注:读者可能好奇SRID是什么?它其实是空间引用标识符(Spatial Reference Identifier)的英文缩写,SQL Server 使用等于4326 的默认 SRID,以映射到WGS 84 空间引用系统,更多内容请见这里。)

问题2:给定位置信息(GPS坐标等),怎样检测5分钟的时间窗口内事件A和事件B是否相聚5英里以内?

为了简化起见,我们假定需要计算出5分钟窗口内所有相聚5英里的事件对。那么一个可行的解决方案是:

  1. 使用AlterEventDuration延伸每个事件生命周期为5分钟;
  2. 与原有输入事件流进行交叉联接
    要注意的是,我们还需要从联接结果中过滤掉自配对以及超过5英里的配对

下面是具体的查询实现:

首先将实时事件流和引用事件流进行联接,已得到完整的状态和位置信息

// 将两个事件流进行联接得到状态和位置信息
var joinedStream = from e1 in liveStream
                   join e2 in assetRef on e1.AssetId equals e2.AssetId
                   select new
                   {
                       AssetId = e1.AssetId,
                       Id = e1.Id,
                       State = e1.State,
                       Latitude = e2.Latitude,
                       Longitude = e2.Longitude
                   };

其次,在5分钟时间窗口内进行交叉联接并过滤结果

 

// 在5分钟的事件窗口内进行交叉联接
var crossJoin = from left in joinedStream.AlterEventDuration(e => TimeSpan.FromMinutes(5))
                from right in joinedStream 
                where 
                    left.AssetId != right.AssetId && // 保证不同的资源类型
                    left.State != right.State  && // 保证不同的状态值
                    // 查找彼此相聚5英里以内的两个事件
                    GetDistance(left.Latitude, left.Longitude, right.Latitude, right.Longitude) <= 5.0
                select new
                {
                    AssetA = left.AssetId,
                    AssetB = right.AssetId,
                    EventIdA = left.Id,
                    EventIdB = right.Id,
                    StateA = left.State,
                    StateB = right.State,
                    Distance = GetDistance(left.Latitude, left.Longitude, right.Latitude, right.Longitude)
               };

  计算两点之间距离函数GetDistance,可以使用SQL Server地理库完成,如下:

 

const int SRID = 4326;
const double MetersPerMile = 1609.344;

/// <summary>
/// 使用SqlGeography类型计算两点之间距离(单位为 英里)
/// </summary>
public static double GetDistance(
    double xLatitude,
    double xLongitude,
    double yLatitude,
    double yLongitude)
{
    return SqlGeography.Point(xLatitude, xLongitude, SRID)
        .STDistance(SqlGeography.Point(yLatitude, yLongitude, SRID))
        .Value / MetersPerMile;
}

最终的结果如下:

【原创】StreamInsight查询系列(二十一)——查询模式之使用地理数据

下一篇将介绍StreamInsight查询模式中的持续更新。

转载于:https://www.cnblogs.com/StreamInsight/archive/2011/09/14/StreamInsight-Query-Series-Part21-Query-Patterns-Working-with-Geography-Data.html