share streamTable(1000:0, `time`a, [TIMESTAMP, DOUBLE]) as table1

share streamTable(1000:0, `time`a, [TIMESTAMP, DOUBLE]) as table2

def append_after_filtering(inputTable, msg){

t = select * from msg where a>100

insert into inputTable values(t.time,t.a)

}

subscribeTable(tableName="table1", actionName="test", offset=-1, handler=append_after_filtering{table2}, msgAsTable=true)

订阅了table1，想要将消息过滤后保存至table2，但table2中无数据。

默认排序时间排序

1 个回答

Hao Jia 2022-03-25 13:41

在定义handler函数的时候，对于inputTable需要用mutable关键字进行修饰，表示可变参数。

share streamTable(1000:0, `time`a, [TIMESTAMP, DOUBLE]) as table1

share streamTable(1000:0, `time`a, [TIMESTAMP, DOUBLE]) as table2

def append_after_filtering(mutable inputTable, msg){

t = select * from msg where a>100

insert into inputTable values(t.time,t.a)

}

subscribeTable(tableName="table1", actionName="test", offset=-1, handler=append_after_filtering{table2}, msgAsTable=true)

相似问题