軌跡資料處理“小鋼炮”，Lindorm時空引擎Ganos實測

語言: CN / TW / HK

時間 2022-09-14 18:10:51 阿里雲資料庫

引言

車聯網、共享出行、智慧物流等領域的快速發展產生了大量的時空軌跡資料。這些軌跡資料來源源不斷的產生，要求儲存系統具備較高的寫入能力、可擴充套件能力以及較低的儲存成本。同時針對這些軌跡資料又產生了各類時空查詢，按照實時性的要求分為三類場景：

線上查詢，如針對歷史軌跡的時空範圍查詢（過去24小時車輛在某個區域的行駛軌跡）、周邊查詢（附近2公里內的計程車）等，一般要求在毫秒或秒級。
實時計算，如實時的電子圍欄判斷（判斷車輛是否駛出目標範圍），時空統計（實時熱力圖，計算某個區域實時車輛數），一般要求在毫秒級。
離線計算，針對大規模歷史軌跡資料做挖掘，比如根據軌跡挖掘出遷徙模式。

一直以來，各類NoSQL資料庫對時空資料的高併發寫入、線上查詢等支援並不完善，每個NoSQL資料庫基本上只能用於某種特定場景。比如：基於Hadoop平臺的方案一般會在UDF層提供時空運算元，但缺少時空索引，無法將時空運算元下推到儲存層，基本上只能用作離線查詢；Apache Sedona（原GeoSpark）內建了時空索引，但一般用於時空資料探勘等場景，不適合實時線上查詢； MongoDB內建了2dsphere空間索引，但由於寫入速度、擴充套件性的瓶頸，普遍只將其用作LBS等場景； GeoMesa作為一款中介軟體，可以藉助HBase等儲存具備較高的寫入能力，而且支援空間填充曲線作為時空索引，但不支援二級索引，當客戶從多維度查詢時需要建立多張表，資料存在冗餘，儲存成本非常高。

在實時計算的場景裡，目前也缺少一個完備的支援時空資料的流引擎，導致很多客戶會引入一個通用的流引擎或者直接用資料庫來近似代替。

此外，這些資料庫均不支援SQL介面，各個資料庫的資料型別、訪問介面均不一致，業務系統在接入不同的資料庫時要進行大量的適配改造。

Lindorm作為一款阿里雲推出的雲原生超融合 多模資料庫 ，包含了流引擎、寬表引擎、物件引擎、搜尋引擎等，那麼自然少不了對時空資料的支援。最新發布的Lindorm已經深度融合了達摩院空天資料庫引擎 Ganos （下文統稱為 Lindorm Ganos ），可以一站式的解決海量軌跡場景的儲存和各類查詢需求，彌補了各類NoSQL在時空方面的不足。

標準化 ：採用SQL介面和Geometry型別，使用者可以像使用PostGIS一樣來使用Lindorm Ganos。
高效能 ：一方面繼承了Lindorm在寫入、擴充套件性、成本等基礎能力的優勢；另一方面提供了時空主鍵索引、時空二級索引來應對多維度查詢，在高效查詢的同時無需為時空場景專門冗餘一份資料。
全面性 ：支援Lindorm寬表引擎和流引擎，一套系統裡既可以電子圍欄這樣的實時計算場景，也可以支援大規模歷史資料的查詢和統計，降低了解決方案的複雜性。

本文在Lindorm Ganos中對上述常用的時空場景進行測試，用過程和實際的資料展示Lindorm Ganos的具備的能力和特性。

1.時空範圍查詢

時空範圍查詢是時空領域的基礎查詢能力，這裡所說的時空範圍查詢是一個統稱，具體又可以分為：

根據空間範圍查詢
根據時間範圍查詢
根據空間範圍 + 時間範圍查詢

Lindorm Ganos提供了原生的時空資料型別、時空運算元、時空主鍵索引、時空二級索引，避免為每一種查詢冗餘儲存一份資料。

查詢給定紅色範圍內的軌跡點

1.1 測試資料

本節以紐約出租車資料為例，來展示Ganos在空間範圍查詢、空間範圍+時間範圍查詢軌跡的步驟和效果，並與相近的幾個系統進行對比。

軌跡資料

紐約2010年計程車軌跡資料，作為基礎資料（資料下載：http://databank.illinois.edu/datasets/IDB-9610843）。取medallion、pickup_datetime、pickup_longitude、pickup_latitude 四個欄位的資料，排除髒資料，共158618394 行，寫入資料原始大小7.6 GB。

查詢範圍資料

紐約行政區劃資料（資料下載：http://data.cityofnewyork.us/City-Government/Community-Districts/yfnk-k7r4），用來作為查詢時的空間範圍，共163個地理圍欄範圍。

1.2 測試環境

Lindorm Ganos測試叢集：Lindorm三節點16核32GB，效能型雲端儲存
開源GeoMesa 3.0.0版本，底層儲存為HBase測試叢集：3節點獨享 16核32GB，SSD雲盤
MongoDB測試叢集：3個Mongos為16核32GB（通用型）；3個shard為16核64G（通用型）

1.3 測試內容

針對海量軌跡的寫入用時
建立時空索引後的空間佔用
時空範圍查詢用時

建立表和索引

Lindorm Ganos：使用SQL語法，壓縮方式為ZSTD，同時為geom列建立空間索引。

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

GeoMesa(HBase)：使用GeoMesa客戶端，壓縮方式為ZSTD，同時為geom列建立z2空間索引。限於篇幅，部分程式碼如下：

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");

MongoDB：壓縮方式為ZSTD，開啟sharding方式批量寫入，同時為geom列建立2dsphere索引。限於篇幅，部分程式碼如下：

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);

空間範圍查詢

Lindorm Ganos：使用SQL語法。

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

GeoMesa(HBase)：使用GeoTools的ECQL語法，並通過GeoTools的介面迭代獲取資料。

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();
}

MongoDB：新增Filter，並通過MongoDB介面迭代獲取資料。

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

時空範圍查詢

Lindorm Ganos：使用SQL語法。

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';

GeoMesa(HBase)：使用GeoTools的ECQL語法，並通過GeoTools的介面迭代獲取資料。

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();
}

MongoDB：新增多個Filter，並通過MongoDB介面迭代獲取資料。

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 Ganos使用的SQL語法都是最簡潔的，使用非常方便。

1.4 測試結果

寫入用時

Lindorm Ganos繼承了Lindorm高效的寫入能力，寫入耗時為GeoMesa（HBase）的 1/2 ，為MongoDB的 1/5 。

資料庫	用時
Lindorm Ganos	7 min
GeoMesa(HBase)	13 min
MongoDB	34 min

儲存空間佔用

在建立空間索引的情況下，Lindorm Ganos佔用的儲存空間更少，為GeoMesa(HBase)的 80% ，為MongoDB的 47% 。

資料庫	表大小	空間索引大小
Lindorm Ganos	4.7 GB	無額外空間，主鍵作索引
GeoMesa(HBase)	5.9 GB	無額外空間，主鍵作索引
MongoDB	8.2 GB	1.6 GB

空間範圍查詢用時

空間範圍查詢場景下，隨著返回結果的增加，幾個系統的查詢用時也在增加。Lindorm Ganos在大部分的查詢中，查詢效能都是大幅領先GeoMesa(HBase)和MongoDB，耗時分別為GeoMesa(HBase)的 1/3 ，MongoDB的 1/2 。

時空範圍查詢用時

時間+空間範圍查詢場景下，Lindorm Ganos在大部分的查詢中，查詢效能都是領先GeoMesa(HBase)和MongoDB，耗時分別是二者的 1/3 和 1/2 左右，個別查詢耗時與二者持平。

2.電子圍欄

實時監控車輛的執行軌跡，判斷車輛是否偏離設定的路線也是一個強需求，這類查詢一般稱為電子圍欄或地理圍欄判斷。與時空範圍查詢相反，電子圍欄是給定一個位置點，來判斷該點是否在某個/些範圍內。這類查詢對實時性的要求比較高，且併發也較大，為了應對這種場景，Ganos結合Lindorm流引擎，可以以流計算的方式來處理。

本節以實時電子圍欄為例，來展示Ganos在這個場景中的應用。除電子圍欄外，Lindorm 流引擎也支援車輛出入圍欄告警等用法。

實時監控車輛是否偏離預定線路

2.1 測試資料

使用北京市公交車線路資料，資料集內共包含1543條線路，資料型別為LineString。我們先通過 ST_Buffer 緩衝區計算函式將線路擴充套件為寬度 10 米的Polygon，這些道路的Polygon作為電子圍欄。

2.2 測試環境

Lindorm Ganos 寬表：4核 16GB 2節點
Lindorm Streams 引擎：32核 64GB 2節點，Topic 分別為 4/8/12/16 partition，1 Producer 1 Consumer

2.3 測試內容

判斷實時寫入的軌跡點在哪個線路上，即需要為每個Point計算出當前所在的Polygon物件，同時統計不同吞吐量下的系統延遲。

寫入資料

將線路的route_id以及對應的Polygon儲存到 Lindorm 寬表，結構如下：

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

建立流引擎對映表

在流引擎中建立一張對映表，指向Lindorm路線表bj_busline，這樣就可以在流引擎中使用寬表中的資料。同時為對映表建立時空索引來加速實時計算，至此，地理圍欄資料存在於流引擎中。

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');

計算鏈路

基於 Lindorm 流引擎連續查詢（Continuous Query，簡稱 CQ）建立計算鏈路，該CQ會從寫入流中取出資料，並通過Join計算之後，將結果寫入到 output_stream 中：

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);

此處 p_id 為上報點位 id，p_location 為點位具體資訊，route 為匹配到的線路名稱，time 為時間戳。

當軌跡點流入後，會與地理圍欄比對，並輸出每個所在的線路名稱，若點並不在某條線路上，則對應的 route 結果為 null。

p_id	p_location	route	time
0	Point(116.4132, 40.0568)	984路區間	2022-09-07 15:18:22
1	Point(116.4133, 40.0569)	null	2022-09-07 15:18:25

2.4 測試結果

下面是不同流入速度和 partition 下，對 1543 個地理圍欄判斷從生產到消費延遲。可以看到隨著點流入速度的提高，系統的延遲也明顯增大，這是因為在上述的計算鏈路裡，每流入1個點則流出1543條判斷結果，所以計算強度是千倍的提升。同時，在同一流入速度下，隨著partition的增加，系統的延遲會有降低。

3.周邊查詢

周邊查詢一般不加時間條件，比如需要查詢周邊300米內有哪些餐館，周圍2公里有哪些計程車。與時空範圍查詢的區別點在於：

周邊查詢一般只會指定中心點以及查詢距離，距離的度量單位一般為米或千米，而時空範圍查詢一般是含有一系列座標點的Polygon物件。
計算距離時，一般是指地球橢球體上的距離，而不是平面座標系的距離。

周邊查詢示意圖

3.1 測試內容

利用紐約出租車資料，查詢所在位置點周邊範圍內的車輛。由於GeoMesa不支援球面距離計算DWithin，我們只與MongoDB的同功能進行對比，比較不同查詢半徑和返回結果數量級的查詢效能。

資料集、建表和寫入的流程與第1節相同，此處不再重複。

查詢語句示例

Lindorm Ganos ：使用SQL語法

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

MongoDB

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();
}