轨迹数据处理“小钢炮”，Lindorm时空引擎Ganos实测

• 在线查询，如针对历史轨迹的时空范围查询（过去24小时车辆在某个区域的行驶轨迹）、周边查询（附近2公里内的出租车）等，一般要求在毫秒或秒级。

• 实时计算，如实时的电子围栏判断（判断车辆是否驶出目标范围），时空统计（实时热力图，计算某个区域实时车辆数），一般要求在毫秒级。

• 离线计算，针对大规模历史轨迹数据做挖掘，比如根据轨迹挖掘出迁徙模式。

CREATE TABLE foil_2010 (
  medallion VARCHAR, 
  pickup_datetime TIMESTAMP, 
  geom GEOMETRY(POINT), 
  primary key(z-order(geom), medallion)
) WITH (COMPRESSION='ZSTD');
复制

String schemaDescription = "medallion:String,pickup_datetime:Date,*geom:Point:srid=4326";
SimpleFeatureType simpleFeatureType = SimpleFeatureTypes.createType("foil_2010", schemaDescription);
simpleFeatureType.getUserData().put("geomesa.table.compression.enabled", true);
simpleFeatureType.getUserData().put("geomesa.table.compression.type", "zstd");
simpleFeatureType.getUserData().put("geomesa.indices.enabled", "z2");
复制

db.createCollection( "foil_2010", {storageEngine:{wiredTiger:{configString:'block_compressor=zstd'}}} );
//创建空间索引
db.foil_2010.createIndex({geom:"2dsphere"});
sh.enableSharding("test");
sh.shardCollection("test.foil_2010",{"_id":"hashed"});
......
collection.bulkWrite(rows);
复制

SELECT medallion 
FROM foil_2010 
WHERE ST_Contains(ST_GeomFromText('POLYGON ((...))'), geom);
复制

ecqlPredicate = "CONTAINS(POLYGON ((...)) , geom)";
query  = new Query("foil_2010", ECQL.toFilter(ecqlPredicate));
result = datastore.getFeatureSource("foil_2010").getFeatures(query);
iterator = result.features();
while (iterator.hasNext()) {
    iterator.next();
}
复制

Bson filter = Filters.geoWithin("geom", new Polygon(new PolygonCoordinates(polygonCoords)));
MongoCursor<Document> cursor = collection.find(filter).iterator();
while (cursor.hasNext()) {
    cursor.next();
}
复制

SELECT medallion 
FROM foil_2010 
WHERE ST_Contains(ST_GeomFromText('POLYGON ((...))'), geom) 
AND pickup_datetime >= 'xxxx-xx-xx xx:xx:xx' 
AND pickup_datetime <= 'xxxx-xx-xx xx:xx:xx';
复制

ecqlPredicate = "CONTAINS(POLYGON ((...)) , geom) AND pickup_datetime >= xxx AND pickup_datetime <= xxx";
query = new Query("foil_2010", ECQL.toFilter(ecqlPredicate));
result = datastore.getFeatureSource("foil_2010").getFeatures(query);
iterator = result.features();
while (iterator.hasNext()) {
    iterator.next();
}
复制

Bson filter = Filters.and(Filters.geoWithin("geom", new Polygon(new PolygonCoordinates(polygonCoords))), Filters.and(Filters.gte("pickup_datetime",startTime), Filters.lte("pickup_datetime", endTime));
MongoCursor<Document> cursor = collection.find(filter).iterator();
while (cursor.hasNext()) {
    cursor.next();
}
复制

# lindorm-cli
create table bj_busline (route_id int,
                         poly geometry(polygon), 
                         primary key(route_id));
复制

CREATE External Table dimTable
WITH (
  table_type = 'lindorm.table', 
  table_name = 'bj_busline', 
  endpoint = 'lindorm-1:30020', 
  output_batchsize=500,
  cache_type = 'LRU', 
  cache_ttl = 1800000,
  ganos_index_type='RTREE',
  ganos_index_polygon_col_name='poly'
);
复制

CREATE STREAM input_stream (
  p_id int, 
  p_location geometry(point), 
  time bigint) 
WITH (stream_topic='input', value_format='JSON', key_value = 'p_id');


CREATE STREAM output_stream (
  p_id int, 
  p_location geometry(point), 
  route_id varchar,
  time bigint) 
WITH (stream_topic='output', value_format='JSON', key_value = 'p_id');
复制

CREATE CQ busRouteJoin INSERT INTO output_stream
SELECT
l.p_id AS p_id, 
l.p_location AS p_location,
r.route_name AS route,
l.time AS time
FROM input_stream l LEFT JOIN dimTable r ON ST_Contains(r.poly, l.p_location);
复制

SELECT medallion
FROM foil_2010
WHERE ST_DWithinSphere(ST_GeomFromText('POINT (120.177206 30.273576)'), geom, 300e0);
复制

Bson filter = Filters.nearSphere("geom", new Point(new Position(120.177206, 30.273576)), 300.0, 0.0);
MongoCursor<Document> cursor = collection.find(filter).iterator();
while (cursor.hasNext()) {
    cursor.next();
}
复制