DStream::snapshotJoinEngine

DStream::snapshotJoinEngine(rightStream, metrics, matchingColumn, [timeColumn], [keepLeftDuplicates=false], [keepRightDuplicates=false], [isInnerJoin=true])

创建流计算快照连接引擎。参考：createSnapshotJoinEngine。

返回值：一个 DStream 对象。

rightStream DStream 对象，表示输入的右表数据源。

timeColumn: 字符串标量或向量，可选参数。用于指定 leftTable 和 rightTable 中时间列的名称。两表的时间列名称可以不同，但数据类型需保持一致。当 leftTable 和 rightTable 时间列名称相同时，timeColumn 是字符串标量，否则，timeColumn 是长度为2的字符串向量。

if (!existsCatalog("orca")) {
	createCatalog("orca")
}
go
use catalog orca


// 如已存在流图，则先销毁该流图
// dropStreamGraph('joinEngine')

g = createStreamGraph('joinEngine')
r = g.source("right", 1024:0, `timestamp`sym2`id`price`qty, [TIMESTAMP, SYMBOL, INT, DOUBLE, DOUBLE])
g.source("left", 1024:0, `timestamp`sym1`id`price`val, [TIMESTAMP, SYMBOL, INT, DOUBLE, DOUBLE])
    .snapshotJoinEngine(r, metrics=[<val*10>, <qty>], matchingColumn = [["id","sym1"],["id","sym2"]], 
timeColumn = `timestamp, isInnerJoin=true, keepLeftDuplicates=true,keepRightDuplicates=true)
    .sink("output")
g.submit()
go

timestamp = 2024.10.10T15:12:01.507+1..10
sym = take(["a","b","c","d"],10)
id = [1,1,2,1,5,2,4,4,1,4]
price = [2.53,7.61,8.07,7.87,7.29,9.39,5.98,9.49,9.20,9.17]
val = [101,108,101,109,104,100,108,100,107,104]
tmp1 = table(timestamp as timestamp,sym as sym1,id as id,price as price,val as val)
appendOrcaStreamTable("left", tmp1)

id = [1,2,4,3,5,5,4,2,5,5]
price =  [1.08,9.08,9.97,7.60,1.91,6.77,7.81,8.81,0.61,5.92]
qty =  [208,200,203,202,204,201,206,207,205,205]
tmp2 = table(timestamp as timestamp,sym as sym2,id as id,price as price,qty as qty)
appendOrcaStreamTable("right", tmp2)

select * from orca_table.output