createNearestJoinEngine

首发版本：3.00.4，3.00.3.1

语法

createNearestJoinEngine(name, leftTable, rightTable, outputTable, kNearest, metrics, matchingColumn, [timeColumn], [useSystemTime=false], [garbageSize = 5000], [maxDelayedTime], [nullFill], [cachedTableCapacity=1024], [snapshotDir], [snapshotIntervalInMsgCount])

详情

创建流计算最近邻关联引擎。引擎接收的左表和右表以 matchingColumn 指定的连接列分组，并在每个组内进行匹配。对于每个分组中每条左表数据，实时关联右表中时间戳不晚于该记录的最近 k 条记录，并基于这些记录计算并输出结果。

注：
不支持乱序处理机制，用户需要自行保证输入的数据有序。

参数

name 表示引擎的名称，作为其在一个数据节点/计算节点上的唯一标识。可包含字母，数字和下划线，但必须以字母开头。

leftTable 表对象。可以不包含数据，但结构必须与输入的流数据表相同。

rightTable 表对象。可以不包含数据，但结构必须与输入的流数据表相同。

outputTable 计算结果的输出表。引擎会将计算结果注入该表。输出表的各列的顺序如下：

  1. 时间列。其中：

    • 若 useSystemTime = true，为 TIMESTAMP 类型；

    • 若 useSystemTime = false，数据类型与 timeColumn 列一致。

  2. 连接列。与 matchingColumn 中的列以及其顺序一致，可为多列。

  3. 计算结果列。可为多列。

kNearest 一个正整数，表示记录的条数。对于每条左表记录，关联右表中时间戳不大于该记录时间戳的最近 k 条记录。

metrics 以元代码的格式表示计算指标，支持输入元组。有关元代码的更多信息可参考 元编程。

  • 计算指标可以是一个或多个表达式、系统内置或用户自定义函数。

  • metrics 内支持调用具有多个返回值的函数，且必须指定列名，例如 <func(price) as `col1`col2>。

    若在 metrics 指定了 leftTable 和 rightTable 中具有相同名称的列，默认取左表的列，可以通过 "tableName.colName" 指定该列来自哪个表。

  • 在需要输出 array vector 类型结果时，需要显式指定toArray函数转换类型，例如：<toArray(price)>，其中 price 为右表中的普通列。

  • 支持使用 toColumnarTuple 函数，将非聚合计算结果转换为列式元组（columnar tuple）。
注意：

  • metrics 中使用的列名大小写不敏感，不要求与输入表的列名大小写保持一致。

  • 当以下函数只计算 rightTable 中的数据列时，最近邻关联引擎对它们进行了优化：sum, sum2, avg, std, var, corr, covar, wavg, wsum, beta, max, min, last, first, med, percentile。

matchingColumn 表示连接列的字符串标量/向量/tuple，支持 Integral, Temporal 或 Literal（UUID 除外）类型。matchingColumn 指定规则为：
  1. 只有一个连接列：当左表和右表的连接列名相同时，matchingColumn 是一个字符串标量，否则是一个长度为 2 的 tuple，例如：左表连接列名为 sym，右表连接列名为 sym1，则 matchingColumn = [[`sym],[`sym1]]。
  2. 有多个连接列：当左表和右表的连接列名相同时，matchingColumn 是一个字符串向量，否则是一个长度为 2 的 tuple，例如：左表连接列名为 timestamp, sym，右表连接列名为 timestamp, sym1，则 matchingColumn = [[`timestamp, `sym], [`timestamp,`sym1]]。

timeColumn 可选参数，当 useSystemTime = false 时，指定要连接的两个表中时间列的名称。leftTable 和 rightTable 时间列名称可以不同，但数据类型需保持一致。当 leftTable 和 rightTable 时间列名称不同时，timeColumn 为一个长度为2的字符串向量。

useSystemTime 可选参数，表示 outputTable 中第一列（时间列）为系统当前时间（useSystemTime = true）或左表的时间列（useSystemTime = false）。

garbageSize 可选参数，是正整数，默认值是5,000（单位为行）。随着订阅的流数据不断积累进入 window join 引擎，存放在内存中的数据会越来越多，这时需要清理不再需要的历史数据。当左/右两表各个分组内的数据行数超过 garbageSize 值时，系统会清理本次计算不需要的历史数据。

maxDelayedTime 可选参数，是正整数，用于触发引擎中长时间未输出的分组数据进行计算。 具体来说，若(某个分组中的左表数据时间戳) + (maxDelayedTime) < (右表最新收到的任意一个分组数据的时间戳)，则这条数据会触发关联计算。

注：

指定该参数时，必须同时指定 timeColumn，且两者的单位需一致。默认值为3秒，根据 timeColumn 的精度换算。例如，若 timeColumn 的精度是毫秒，则默认值为3000毫秒。

注：
请合理设置 maxDelayedTime 参数。当左表数据较少而右表数据频率较高时，若 maxDelayedTime 设置过小，系统可能在未触发某些分组计算之前，就清理掉右表中过早的数据。

nullFill 和输出表列字段等长且类型一一对应的元组，用于填充以下列中的空值：输出表中包含的左表列、右表列、右表列被聚合计算后的计算结果列。

cachedTableCapacity 可选参数，为正整数，表示引擎为每个不同的分组分别创建的左缓存表和右缓存表的初始容量（以数据条数计）。默认值为 1024。

snapshotDir 可选参数，字符串，表示保存引擎快照的文件目录。

  • 指定的目录必须存在，否则系统会提示异常。

  • 创建流数据引擎时，如果指定了 snapshotDir，会检查该目录下是否存在快照。如果存在，会加载该快照，恢复引擎的状态。

  • 多个引擎可以指定同一个目录存储快照，用引擎的名称来区分快照文件。

  • 一个引擎的快照可能会使用三个文件名：

  • 临时存储快照信息：文件名为 <engineName>.tmp；

  • 快照生成并刷到磁盘：文件保存为 <engineName>.snapshot；

  • 存在同名快照：旧快照自动重命名为 <engineName>.old。

snapshotIntervalInMsgCount 可选参数，正整数，表示每隔多少条数据保存一次流数据引擎快照。

返回值

一个表。

例子

例1. 本例简单展示如何通过 createNearestJoinEngine 创建一个 nearest join 引擎，按股票代码（sym）分组，将左表的价格数据与右表时间戳不晚于左表记录的最近 8 条数值数据进行关联计算。

share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE[], DOUBLE]) as output

nullFill= [2012.01.01T00:00:00.000, `NONE, 0.0, [], 0.0]

njEngine=createNearestJoinEngine(name="test1", leftTable=leftTable, rightTable=rightTable, outputTable=output,  kNearest=8, metrics=<[price,toArray(val),sum(val)]>, matchingColumn=`sym, timeColumn=`time, useSystemTime=false,nullFill=nullFill)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{njEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{njEngine, false}, msgAsTable=true)


n=10

tp2=table(take(2012.01.01T00:00:00.000+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(double(1..n),2*n) as val)
tp2.sortBy!(`time)
rightTable.append!(tp2)


tp1=table(take(2012.01.01T00:00:00.003+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(NULL join rand(10.0, n-1),2*n) as price)
tp1.sortBy!(`time)
leftTable.append!(tp1)


tp2=table(take(2012.01.01T00:00:00.010+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(double(1..n),2*n) as val)
tp2.sortBy!(`time)
rightTable.append!(tp2)

select * from output
time sym factor1 factor2 factor3
2012.01.01 00:00:00.003 A 0 [1, 2, 3, 4] 10
2012.01.01 00:00:00.004 A 8.049739237951693 [1, 2, 3, 4, 5] 15
2012.01.01 00:00:00.005 A 6.31845193685475 [1, 2, 3, 4, 5, 6] 21
2012.01.01 00:00:00.006 A 0.01247286192106635 [1, 2, 3, 4, 5, 6, 7] 28
2012.01.01 00:00:00.007 A 8.373015887228414 [1, 2, 3, 4, 5, 6, 7, 8] 36
2012.01.01 00:00:00.008 A 4.636610761119452 [2, 3, 4, 5, 6, 7, 8, 9] 44
2012.01.01 00:00:00.003 B 8.049739237951693 [2, 3, 4, 5] 14
2012.01.01 00:00:00.004 B 6.31845193685475 [2, 3, 4, 5, 6] 20
2012.01.01 00:00:00.005 B 0.01247286192106635 [2, 3, 4, 5, 6, 7] 27
2012.01.01 00:00:00.006 B 8.373015887228414 [2, 3, 4, 5, 6, 7, 8] 35
2012.01.01 00:00:00.007 B 4.636610761119452 [2, 3, 4, 5, 6, 7, 8, 9] 44
2012.01.01 00:00:00.008 B 7.700075873220435 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.009 B 0.5831421500989946 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.009 A 7.700075873220435 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.010 A 0.5831421500989946 [4, 5, 6, 7, 8, 9, 10, 1] 50
2012.01.01 00:00:00.011 A 5.117162734418752 [5, 6, 7, 8, 9, 10, 1, 2] 48
2012.01.01 00:00:00.012 A 8.823084861596655 [6, 7, 8, 9, 10, 1, 2, 3] 46
2012.01.01 00:00:00.010 B 5.117162734418752 [5, 6, 7, 8, 9, 10, 1, 2] 48
2012.01.01 00:00:00.011 B 8.823084861596655 [6, 7, 8, 9, 10, 1, 2, 3] 46
2012.01.01 00:00:00.013 B 0 [8, 9, 10, 1, 2, 3, 4, 5] 42

例2. 本例展示matchingColumn 为多个连接列的情况。左右表的连接列名不同，左表使用 market 和 sym，右表使用 mkt 和 code。

share streamTable(1:0, `time`market`symbol`price, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as leftTable2
share streamTable(1:0, `time`mkt`code`val, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as rightTable2
share table(100:0, `time`market`symbol`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE, DOUBLE[], DOUBLE]) as output2

nullFill2 = [2012.01.01T00:00:00.000, `NONE, `NONE, 0.0, [], 0.0]

// matchingColumn=[[`market`,symbol], [`mkt`,code]], 使用 tuple 指定左右表列名的映射关系
njEngine2=createNearestJoinEngine(name="test2", leftTable=leftTable2, rightTable=rightTable2, outputTable=output2, kNearest=5, metrics=<[price, toArray(val), sum(val)]>, matchingColumn=[[`market,`symbol], [`mkt,`code]], timeColumn=`time, useSystemTime=false, nullFill=nullFill2)

subscribeTable(tableName="leftTable2", actionName="joinLeft2", offset=0, handler=appendForJoin{njEngine2, true}, msgAsTable=true)
subscribeTable(tableName="rightTable2", actionName="joinRight2", offset=0, handler=appendForJoin{njEngine2, false}, msgAsTable=true)

n=8

// 右表数据：SSE-A, SSE-B, SZSE-A, SZSE-B
tp_right=table(take(2012.01.01T00:00:00.000+0..10, 4*n) as time, 
               (take(`SSE, n) join take(`SSE, n) join take(`SZSE, n) join take(`SZSE, n)) as mkt, 
               (take(`A, n) join take(`B, n) join take(`A, n) join take(`B, n)) as code, 
               take(double(1..n), 4*n) as val)
tp_right.sortBy!(`time)
rightTable2.append!(tp_right)

// 左表数据
tp_left=table(take(2012.01.01T00:00:00.003+0..10, 4*n) as time, 
              (take(`SSE, n) join take(`SSE, n) join take(`SZSE, n) join take(`SZSE, n)) as market, 
              (take(`A, n) join take(`B, n) join take(`A, n) join take(`B, n)) as symbol, 
              take(NULL join rand(10.0, n-1), 4*n) as price)
tp_left.sortBy!(`time)
leftTable2.append!(tp_left)

select * from output2
time market symbol factor1 factor2 factor3
2012.01.01 00:00:00.003 SSE A 0 [1, 2, 3, 4] 10
2012.01.01 00:00:00.004 SSE A 5.665292291824075 [1, 2, 3, 4, 5] 15
2012.01.01 00:00:00.005 SSE A 7.102908472787335 [2, 3, 4, 5, 6] 20
2012.01.01 00:00:00.006 SSE A 1.4781711766212646 [3, 4, 5, 6, 7] 25
2012.01.01 00:00:00.003 SSE B 1.4781711766212646 [4, 5, 6, 7] 22
2012.01.01 00:00:00.004 SSE B 8.480999725517119 [4, 5, 6, 7, 8] 30
2012.01.01 00:00:00.005 SSE B 4.88365191847171 [4, 5, 6, 7, 8] 30
2012.01.01 00:00:00.006 SSE B 8.075179240683834 [4, 5, 6, 7, 8] 30
2012.01.01 00:00:00.007 SSE B 5.3205939278287335 [4, 5, 6, 7, 8] 30
2012.01.01 00:00:00.003 SZSE A 8.075179240683834 [7, 8] 15
2012.01.01 00:00:00.004 SZSE A 5.3205939278287335 [7, 8] 15
2012.01.01 00:00:00.008 SZSE A 0 [8, 1, 2, 3, 4] 18
2012.01.01 00:00:00.009 SZSE A 5.665292291824075 [1, 2, 3, 4, 5] 15
2012.01.01 00:00:00.005 SZSE B 0 [1, 2, 3, 4] 10
2012.01.01 00:00:00.006 SZSE B 5.665292291824075 [1, 2, 3, 4, 5] 15
2012.01.01 00:00:00.007 SZSE B 7.102908472787335 [2, 3, 4, 5, 6] 20
2012.01.01 00:00:00.008 SZSE B 1.4781711766212646 [3, 4, 5, 6, 7] 25

例3. 设置 maxDelayedTime=5（5 毫秒），左表股票 A 数据时间戳过了 5ms 后，右表依然没有股票 A 的新数据到达，但是右表收到了股票 B 的新数据，且时间戳满足 (左表 A 的时间戳) + 5 < (右表 B 的最新时间戳)，则引擎会强制触发股票 A 的计算并输出。

// 创建表结构
share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE[], DOUBLE]) as output3

nullFill= [2012.01.01T00:00:00.000, `NONE, 0.0, [], 0.0]

// 创建引擎，设置 maxDelayedTime=5
njEngine=createNearestJoinEngine(name="testMaxDelay", leftTable=leftTable, rightTable=rightTable, outputTable=output3, kNearest=8, metrics=<[price,toArray(val),sum(val)]>, matchingColumn=`sym, timeColumn=`time, useSystemTime=false, nullFill=nullFill, maxDelayedTime=5)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{njEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{njEngine, false}, msgAsTable=true)

n=10

// 注入数据
// 右表数据，A 和 B 股票的时间戳都是 2012.01.01 00:00:00.000-009
tp2_1 = table(take(2012.01.01T00:00:00.000 + 0..(n-1), n) join take(2012.01.01T00:00:00.000 + 0..(n-1), n) as time, take(`A, n) join take(`B, n) as sym, double(1..20) as val)
rightTable.append!(tp2_1)
// 左表数据，A 和 B 股票的时间戳都是 2012.01.01 00:00:00.005-014
tp1 = table(take(2012.01.01T00:00:00.005 + 0..(n-1), n) join take(2012.01.01T00:00:00.005 + 0..(n-1), n) as time, take(`A, n) join take(`B, n) as sym, rand(10.0, 20) as price)
leftTable.append!(tp1)
// 此时右表只注入 B 的数据（时间 2012.01.01 00:00:00.010-020），强制触发 A 的计算
tp2_2 = table(2012.01.01T00:00:00.010 + 0..10 as time, take(`B, 11) as sym, double(21..31) as val)
rightTable.append!(tp2_2)

// 查看结果
select * from output3
time sym factor1 factor2 factor3
2012.01.01 00:00:00.005 A 5.603691575956459 [1, 2, 3, 4, 5, 6] 21
2012.01.01 00:00:00.006 A 0.5876812223782021 [1, 2, 3, 4, 5, 6, 7] 28
2012.01.01 00:00:00.007 A 9.89037483462374 [1, 2, 3, 4, 5, 6, 7, 8] 36
2012.01.01 00:00:00.008 A 5.416240437286031 [2, 3, 4, 5, 6, 7, 8, 9] 44
2012.01.01 00:00:00.005 B 8.404138905090312 [11, 12, 13, 14, 15, 16] 81
2012.01.01 00:00:00.006 B 5.313030796896906 [11, 12, 13, 14, 15, 16, 17] 98
2012.01.01 00:00:00.007 B 8.364397254391665 [11, 12, 13, 14, 15, 16, 17, 18] 116
2012.01.01 00:00:00.008 B 6.625538486201674 [12, 13, 14, 15, 16, 17, 18, 19] 124
2012.01.01 00:00:00.009 A 7.032353791182943 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.010 A 2.569216928111673 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.011 A 0.9054661195039437 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.012 A 1.8481626598742238 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.013 A 0.7850572887773293 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.014 A 3.8230257723068415 [3, 4, 5, 6, 7, 8, 9, 10] 52
2012.01.01 00:00:00.009 B 4.547090470918242 [13, 14, 15, 16, 17, 18, 19, 20] 132
2012.01.01 00:00:00.010 B 9.568013637226079 [14, 15, 16, 17, 18, 19, 20, 21] 140
2012.01.01 00:00:00.011 B 9.821790172676973 [15, 16, 17, 18, 19, 20, 21, 22] 148
2012.01.01 00:00:00.012 B 6.679888851167608 [16, 17, 18, 19, 20, 21, 22, 23] 156
2012.01.01 00:00:00.013 B 4.947056149818715 [17, 18, 19, 20, 21, 22, 23, 24] 164
2012.01.01 00:00:00.014 B 0.8401612892840433 [18, 19, 20, 21, 22, 23, 24, 25] 172

其结果与下面不设置 maxDelayedTime 的结果表相比，出现了后者没等到右表数据的 A 股票 009 到 014 的结果：

time sym factor1 factor2 factor3
2012.01.01 00:00:00.005 A 0.7138895151936192 [1, 2, 3, 4, 5, 6] 21
2012.01.01 00:00:00.006 A 1.2329890698271313 [1, 2, 3, 4, 5, 6, 7] 28
2012.01.01 00:00:00.007 A 2.8890747187866537 [1, 2, 3, 4, 5, 6, 7, 8] 36
2012.01.01 00:00:00.008 A 9.563884956194993 [2, 3, 4, 5, 6, 7, 8, 9] 44
2012.01.01 00:00:00.005 B 3.0864263426248044 [11, 12, 13, 14, 15, 16] 81
2012.01.01 00:00:00.006 B 6.196044850208807 [11, 12, 13, 14, 15, 16, 17] 98
2012.01.01 00:00:00.007 B 5.7922327415533905 [11, 12, 13, 14, 15, 16, 17, 18] 116
2012.01.01 00:00:00.008 B 3.3132631851623913 [12, 13, 14, 15, 16, 17, 18, 19] 124
2012.01.01 00:00:00.009 B 6.309944234394921 [13, 14, 15, 16, 17, 18, 19, 20] 132
2012.01.01 00:00:00.010 B 4.219283364764248 [14, 15, 16, 17, 18, 19, 20, 21] 140
2012.01.01 00:00:00.011 B 5.666794845756794 [15, 16, 17, 18, 19, 20, 21, 22] 148
2012.01.01 00:00:00.012 B 0.6544164360161909 [16, 17, 18, 19, 20, 21, 22, 23] 156
2012.01.01 00:00:00.013 B 0.15740090984790608 [17, 18, 19, 20, 21, 22, 23, 24] 164
2012.01.01 00:00:00.014 B 1.398327357926948 [18, 19, 20, 21, 22, 23, 24, 25] 172