createLookupJoinEngine

createLookupJoinEngine(name, leftTable, rightTable, outputTable, metrics, matchingColumn, [rightTimeColumn], [checkTimes], [outputElapsedMicroseconds=false], [keepDuplicates=false], [isInnerJoin], [snapshotDir], [snapshotIntervalInMsgCount])

创建流数据表的 lookup join 引擎，该引擎以 matchingColumn 作为连接列，将左表（必须为流数据表）与右表进行左连接或内连接。lookup join 引擎常用于右表更新不频繁的场景（如保存了日频指标的维度表）。若右表为非流数据表，则需设置定时刷新该表数据。

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

leftTable 可选参数, 表对象。可以不包含数据，但结构必须与订阅的流数据表相同。

rightTable可选参数, 表对象，可以是内存表、流数据表、维度表或 SQL 查询元代码（查询语句必须返回表 ）。请注意，如果 rightTable 没有被订阅，但 rightTable 会定期更新，则必须设置 checkTimes 来定时刷新右表数据。

outputTable 必选参数，为计算结果的输出表。在使用 createLookupJoinEngine 函数之前，需要将输出表预先设立为一个空表，并指定各列列名以及数据类型。

输出表的各列的顺序如下：

1. 连接列。与 matchingColumn 中的列以及其顺序一致，可为多列。
2. 计算结果列。可为多列。
3. 耗时列。如果指定 outputElapsedMicroseconds = true，则指定一个 LONG 类型的列用于记录单次响应计算耗时（单位：微秒）。
4. batchSize 列。如果指定 outputElapsedMicroseconds = true, 则指定一个INT类型的列，记录单次响应的数据条数。

rightTimeColumn 可选参数, 是字符串标量，表示右表的时间列名称。若设置该参数，右表会根据指定的时间列的时间戳保留最新的数据（若有多行，则取其中最后一行）。若不指定该参数，则根据数据注入系统的时间保留最新数据。

outputElapsedMicroseconds 可选参数, 布尔值，表示是否输出单次响应计算的耗时（从触发计算的数据注入引擎到计算完成的耗时），默认为 false。指定参数 outputElapsedMicroseconds 后，在定义 outputTable 时需要在计算结果列后增加一个 LONG 类型的列和 INT 类型的列，详见 outputTable 参数说明。

keepDuplicates 可选参数, 布尔值，表示是否保留右表各分组内的所有数据。默认值为 false，即在关联时只取右表各分组内的最新一条数据。当设置为 true 时，在关联时则使用右表各分组内的所有数据，此时的连接类型为内连接，即只有左右两表匹配的记录被计算输出。

isInnerJoin：布尔值，可选参数，表示是否进行内连接。该参数只在 keepDuplicates 为 false 时可进行配置。该参数默认值为 false，即进行左连接，无论在右表中是否找到匹配的记录，都会输出结果（右表未匹配的记录输出空值）。设置为 true 时左右两表进行内连接。

若要开启快照机制 (snapshot)，必须指定 snapshotDir 与 snapshotIntervalInMsgCount。

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

• 指定的目录必须存在，否则系统会提示异常。
• 创建流数据引擎时，如果指定了 snapshotDir，会检查该目录下是否存在快照。如果存在，会加载该快照，恢复引擎的状态。
• 多个引擎可以指定同一个目录存储快照，用引擎的名称来区分快照文件。
• 一个引擎的快照可能会使用三个文件名：
• 临时存储快照信息：文件名为 <engineName>.tmp；
• 快照生成并刷到磁盘：文件保存为 <engineName>.snapshot；
• 存在同名快照：旧快照自动重命名为 <engineName>.old。

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

例1.

login(`admin, `123456)
share streamTable(1000:0, `timestamps`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as trades
share streamTable(1000:0, `timestamps`sym`val`id, [TIMESTAMP, SYMBOL, DOUBLE, INT]) as prices
share table(100:0, `sym`factor1`factor2`factor3, [SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

LjEngine = createLookupJoinEngine(name="test1", leftTable=trades, rightTable=prices, outputTable=output, metrics=<[price,val,price*val]>, matchingColumn=`sym)
subscribeTable(tableName="trades", actionName="append_leftTable", offset=0, handler=appendForJoin{LjEngine, true}, msgAsTable=true)
subscribeTable(tableName="prices", actionName="append_rightTable", offset=0, handler=appendForJoin{LjEngine, false}, msgAsTable=true)

n = 15
tem1 = table( (2018.10.08T01:01:01.001 + 1..12) join (2018.10.08T01:01:01.001 + 1..3)as timestamps,take(`A`B`C, n) as sym,take(1..15,n) as val,1..15 as id)
prices.append!(tem1)
sleep(2000)
n  = 10
tem2 = table( 2019.10.08T01:01:01.001 + 1..n as timestamps,take(`A`B`C, n) as sym,take(0.1+10..20,n) as price)
trades.append!(tem2)
sleep(100)
select * from output

例2.

share streamTable(1000:0, `timestamps`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as trades
share streamTable(1000:0, `timestamps`sym`val`id, [TIMESTAMP, SYMBOL, DOUBLE, INT]) as prices
share table(100:0, `sym`factor1`factor2`factor3, [SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output
LjEngine = createLookupJoinEngine(name="test1", leftTable=trades, rightTable=prices, outputTable=output, metrics=<[price,val,price*val]>, matchingColumn=`sym, rightTimeColumn=`timestamps)
subscribeTable(tableName="trades", actionName="append_leftTable", offset=0, handler=appendForJoin{LjEngine, true}, msgAsTable=true)
subscribeTable(tableName="prices", actionName="append_rightTable", offset=0, handler=appendForJoin{LjEngine, false}, msgAsTable=true)

n = 15
tem1 = table( (2018.10.08T01:01:01.001 + 1..12) join (2018.10.08T01:01:01.001 + 1..3)as timestamps,take(`A`B`C, n) as sym,take(1..15,n) as val,1..15 as id)
prices.append!(tem1)
sleep(2000)
n  = 10
tem2 = table( 2019.10.08T01:01:01.001 + 1..n as timestamps,take(`A`B`C, n) as sym,take(0.1+10..20,n) as price)
trades.append!(tem2)
sleep(100)
select * from output

例3. 右表是内存表，需设置 checkTimes

share streamTable(1000:0, `timestamps`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as trades
share table(100:0, `sym`factor1`factor2`factor3, [SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output
share table(1000:0, `timestamps`sym`val`id, [TIMESTAMP, SYMBOL, DOUBLE, INT]) as prices
LjEngine = createLookupJoinEngine(name="test1", leftTable=trades, rightTable=prices, outputTable=output, metrics=<[price,val,price*val]>, matchingColumn=`sym, rightTimeColumn=`timestamps, checkTimes=1s)
subscribeTable(tableName="trades", actionName="append_leftTable", offset=0, handler=appendForJoin{LjEngine, true}, msgAsTable=true)

n = 15
tem1 = table( (2018.10.08T01:01:01.001 + 1..12) join (2018.10.08T01:01:01.001 + 1..3)as timestamps,take(`A`B`C, n) as sym,take(1..15,n) as val,1..15 as id)
prices.append!(tem1)
sleep(2000)
n  = 10
tem2 = table( 2019.10.08T01:01:01.001 + 1..n as timestamps,take(`A`B`C, n) as sym,take(0.1+10..20,n) as price)
trades.append!(tem2)
sleep(100)
select * from output

例4. 左表为一个实时的交易表与右表（相对稳定的维度表）做连接。

share streamTable(1000:0, `time`volume`id, [TIMESTAMP, INT,INT]) as trades
dbPath="dfs://testlj"
if(existsDatabase(dbPath)){
   dropDatabase(dbPath)
}
rt=table(1000:0, `time`price`id, [TIMESTAMP, DOUBLE, INT])
db=database(dbPath, VALUE, `A`B`C)
prices=db.createTable(rt,`rightTable)
share table(10000:0, `id`volume`price`prod, [INT,INT,DOUBLE,DOUBLE]) as outputTable

tradesLookupJoin = createLookupJoinEngine(name="streamLookup1", leftTable=trades, rightTable=prices, outputTable=outputTable, metrics=<[volume,price,volume*price]>, matchingColumn=`id, rightTimeColumn=`time,checkTimes=1s)
subscribeTable(tableName="trades", actionName="append_trades", offset=0, handler=appendForJoin{tradesLookupJoin, true}, msgAsTable=true)

def writeData(t,n){
    timev = 2021.10.08T01:01:01.001 + timestamp(1..n)
    volumev = take(1..n, n)
    id = take(1 2 3, n)
    insert into t values(timev, volumev, id)
}
writeData(rt, 10)
prices.append!(rt)
sleep(2000)
writeData(trades, 6)
sleep(2)

select * from outputTable

例5. 通过 rightTable 可以对分布式分区表中的字段进行关联，此时 rightTable 是一个 SQL 查询元代码：

share streamTable(1000:0, `time`volume`id, [TIMESTAMP, INT,INT]) as trades
dbPath="dfs://lookupjoinDB"
if(existsDatabase(dbPath)){
   dropDatabase(dbPath)
}
rt=table(1000:0, `time`price`id, [TIMESTAMP, DOUBLE, INT])
db=database(dbPath, HASH, [INT,5])
prices=db.createPartitionedTable(rt,`rightTable, `id)
share table(10000:0, `id`volume`price`prod, [INT,INT,DOUBLE,DOUBLE]) as outputTable

tradesLookupJoin = createLookupJoinEngine(name="streamLookup1", leftTable=trades, rightTable=<select * from loadTable(dbPath, `rightTable)>, outputTable=outputTable, metrics=<[volume,price,volume*price]>, matchingColumn=`id, rightTimeColumn=`time,checkTimes=1s)
subscribeTable(tableName="trades", actionName="append_trades", offset=0, handler=appendForJoin{tradesLookupJoin, true}, msgAsTable=true)

def writeData(t,n){
    timev = 2021.10.08T01:01:01.001 + timestamp(1..n)
    volumev = take(1..n, n)
    id = take(1 2 3, n)
    insert into t values(timev, volumev, id)
}
writeData(rt, 10)
prices.append!(rt)
sleep(2000)
writeData(trades, 6)
sleep(2)

select * from outputTable

例6. 设置 isInnerJoin 为 true，当在右表中没有匹配的记录时不输出结果。

login(`admin, `123456)
share streamTable(1000:0, `timestamps`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as trades
share streamTable(1000:0, `timestamps`sym`val`id, [TIMESTAMP, SYMBOL, DOUBLE, INT]) as prices
share table(100:0, `sym`factor1`factor2`factor3, [SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

LjEngine = createLookupJoinEngine(name="test1", leftTable=trades, rightTable=prices, outputTable=output, metrics=<[price,val,price*val]>, matchingColumn=`sym, isInnerJoin=true)

n = 15
tem1 = table( (2018.10.08T01:01:01.001 + 1..12) join (2018.10.08T01:01:01.001 + 1..3)as timestamps,take(`A`B`C, n) as sym,take(1..15,n) as val,1..15 as id)
appendForJoin(LjEngine, false,tem1)
sleep(2000)

// 左表中存在 2 条右表中无匹配的记录
n  = 10
tem2 = table( 2019.10.08T01:01:01.001 + 1..n as timestamps,take(`A`B`C`d, n) as sym,take(0.1+10..20,n) as price)
appendForJoin(LjEngine, true,tem2)
sleep(100)
select count(*) from output // 8