streamFilter

streamFilter(name, dummyTable, filter, [msgSchema], [timeColumn], [conditionColumn])

Create an engine that splits the ingested stream for different handlers. Return a table object.

The engine works as follows:

1. Deserialize the ingested data.

   Note that this step only takes place when the ingested data is from a heterogeneous stream table, i.e., the output of heterogeneous replay.

2. Split the ingested stream based on the conditions as specified by filter.

3. Ingest the split streams to the handlers as specified by filter in the order of their timestamps.

name is a STRING scalar indicating the name of the stream filter engine. It must begin with a letter and may contain letters, numbers and underscores.

dummyTable is a table. It has the same schema as the stream table that the stream filter subscribes to. The table can be empty.

The following parameters are only required when processing data from a standard stream table:

timeColumn (optional) is a STRING indicating the name of the temporal column in dummyTable. If unspecified, it takes the name of the first column in the dummyTable.

conditionColumn (optional) is a STRING indicating a column (must be STRING or SYMBOL type) in dummyTable. If this parameter is unspecified, the "condition" key of filter takes no effect.

(1) Processing a heterogeneous stream table:

Replay the DFS tables "orders" and "trades" to simulate an asof join of the two streams.

If we simply perform an N-to-N replay on the two tables, it is not guaranteed that the records will be ingested to the left and right tables of the asof join engine in chronological order: It may happen that a record with a larger timestamp arrives in the left table before a record with a smaller timestamp arrives in the right table. For more information, see replay.

Therefore, we replay the two tables into one heterogeneous stream table to make sure all records are ordered by timestamp. The stream filter subscribes to the heterogenous stream table, splits the ingested data into two streams and distributes them to the left and right tables of the asof join engine (createAsofJoinEngine). In this way, we can make sure that the data is ingested to the left and right tables of the asof join engine in the order of the timestamps.

//create the "orders" table
n=1000
sym = take(take("IBM",n).join(take("GS",n)), n*2*3)
date=take(2022.01.04..2022.01.06, n*2*3).sort!()
timestamp1=take(2022.01.04 09:30:00.000+rand(1000,n),n) join take(2022.01.04 09:31:00.000+rand(1000,n),n)
timestamp2=take(2022.01.05 09:30:00.000+rand(1000,n),n) join take(2022.01.05 09:31:00.000+rand(1000,n),n)
timestamp3=take(2022.01.06 09:30:00.000+rand(1000,n),n) join take(2022.01.06 09:31:00.000+rand(1000,n),n)
timestamp=timestamp1 join timestamp2 join timestamp3
volume = rand(100, n*2*3)
t=table(sym,date,timestamp,volume)

if(existsDatabase("dfs://test_order")){
dropDatabase("dfs://test_order")
}
db1_or=database("",RANGE, 2022.01.04..2022.01.07)
db2_or=database("",VALUE,`IBM`GS)
db_or=database("dfs://test_order",COMPO,[db1_or, db2_or])
orders=db_or.createPartitionedTable(t,`orders,`date`sym)
orders.append!(t);
select count(*) from orders
// output
6000

//create the "trades" table
n=2000
sym = take(take("IBM",n).join(take("GS",n)), n*2*3)
date=take(2022.01.04..2022.01.06, n*2*3).sort!()
timestamp1=take(2022.01.04 09:30:00.000+rand(1000,n),n) join take(2022.01.04 09:31:00.000+rand(1000,n),n)
timestamp2=take(2022.01.05 09:30:00.000+rand(1000,n),n) join take(2022.01.05 09:31:00.000+rand(1000,n),n)
timestamp3=take(2022.01.06 09:30:00.000+rand(1000,n),n) join take(2022.01.06 09:31:00.000+rand(1000,n),n)
timestamp=timestamp1 join timestamp2 join timestamp3
volume = rand(100, n*2*3)
price = rand(50.0, n*3) join  rand(20.0, n*3)

t=table(sym,date,timestamp,volume,price)

if(existsDatabase("dfs://test_trades")){
dropDatabase("dfs://test_trades")
}
db1=database("",RANGE, 2022.01.04..2022.01.07)
db2=database("",VALUE,`IBM`GS)
db=database("dfs://test_trades",COMPO,[db1, db2])
trades=db.createPartitionedTable(t,`trades,`date`sym)
trades.append!(t);
select count(*) from trades
// output
12000

//generate the heterogeneous data sources and create a table as the the outputTable of replay()
ds_or = replayDS(sqlObj=<select * from loadTable(db_or, `orders)>, dateColumn=`date, timeColumn=`timestamp)
ds = replayDS(sqlObj=<select * from loadTable(db, `trades)>, dateColumn=`date, timeColumn=`timestamp)
input_dict=dict(["orders","trades"], [ds_or, ds])
share streamTable(100:0,`timestamp`sym`blob`volume, [TIMESTAMP,SYMBOL, BLOB, INT]) as opt


//subscribe to the output table of replay to ingest the data to the stream filter
share streamTable(100:0,`timestamp`sym`blob`volume, [TIMESTAMP,SYMBOL, BLOB, INT]) as streamFilterOpt
share streamTable(100:0, `sym`date`timestamp`volume, [SYMBOL, DATE, TIMESTAMP, INT] ) as streamOrders
share streamTable(100:0, `sym`date`timestamp`volume`price, [SYMBOL, DATE, TIMESTAMP, INT, DOUBLE] ) as streamTrades
streamOpt=table(100:0, `timestamp`sym`volume`price`result, [TIMESTAMP, SYMBOL, INT, DOUBLE, DOUBLE])

filter1=dict(STRING,ANY)
filter1['condition']=`orders
filter1['timeRange']=09:30:00.000:09:30:00.005

filter2=dict(STRING,ANY)
filter2['condition']=`trades
filter2['timeRange']=09:30:00.000:09:30:00.005

ajEngine=createAsofJoinEngine(name="ajEngine", leftTable=streamOrders, rightTable=streamTrades, outputTable=streamOpt, metrics=<[volume,price,price*volume]>, matchingColumn=`sym, useSystemTime=true)
filter1['handler']=getLeftStream(ajEngine)
filter2['handler']=getRightStream(ajEngine)
schema=dict(["orders","trades"], [streamOrders, streamTrades])

engine=streamFilter(name=`streamFilter,dummyTable=streamFilterOpt, filter=[filter1,filter2],msgSchema=schema)
subscribeTable(tableName="opt", actionName="sub1", offset=0, handler=engine, msgAsTable=true)

//replay the heterogeneous data sources and output to the table "opt"
replay(inputTables=input_dict,outputTables=opt, timeColumn=`timestamp)

select count(*) from streamOpt
// output
20

//drop the subscription
unsubscribeTable(tableName="opt", actionName="sub1")
dropStreamEngine(`streamFilter)
dropStreamEngine(`ajEngine)

(2) Processing standard stream table:

In this example, data from the standard stream table "trades" is ingested to the stream filter, where the records are filtered and assigned to handlers for further processing.

n=20
sym = symbol(take(`A`B`C,n))
name = string(rand(1..10,n))
date = temporalAdd(2012.12.06,0..(n-1),'d')
time = temporalAdd(09:30:00.000,0..(n-1),'ms')
vol = 100+take(1..8,20)
t = table(date,time,sym,name,vol)

share streamTable(100:0,`date`time`sym`name`vol,[DATE,TIME,SYMBOL,STRING,INT]) as st1
share streamTable(100:0,`date`time`sym`name`vol,[DATE,TIME,SYMBOL,STRING,INT]) as st2
share streamTable(100:0,`date`time`sym`name`vol,[DATE,TIME,SYMBOL,STRING,INT]) as st3


share streamTable(100:0,`time`sym`sum_vol,[TIME,SYMBOL,INT]) as output1
share streamTable(100:0,`time`avg_vol,[TIME,INT]) as output2

// create 2 streaming engines has the handlers of the stream filter
engine1=createTimeSeriesEngine(name="timeEngine", windowSize=3, step=3, metrics=<[sum(vol)]>, dummyTable=st3, outputTable=output1, timeColumn=`time, useSystemTime=false, keyColumn=`sym, garbageSize=50)
engine2=createReactiveStateEngine(name="reactiveEngine", metrics=<[mavg(vol, 3)]>, dummyTable=st1, outputTable=output2, keyColumn=`sym)

//share "trades" as the stream table to be subscribed by the stream filter
share streamTable(100:0,`date`time`sym`name`vol,[DATE,TIME,SYMBOL,STRING,INT]) as trades

//set the first filter and ingest the result to engine2
filter1 = dict(STRING,ANY)
filter1['condition']=`A
filter1['handler']=engine2
filter1['timeRange']=(09:30:00.001:09:30:00.010,09:29:00.000:09:30:00.000)

//set the second filter and ingest the result to st2
filter2 = dict(STRING,ANY)
filter2['handler']=st2
filter2['timeRange']=09:30:00.002:09:30:00.005

//set the first filter and ingest the result to engine1
filter3 = dict(STRING,ANY)
filter3['condition']=`C`A
filter3['handler']=engine1

///The stream filter subscribes to the stream table "trades" and distributes the ingested data based on the specified conditions
streamFilter2=streamFilter(name="streamFilterDemo",dummyTable=trades,filter=[filter1,filter2,filter3], timeColumn=`time, conditionColumn=`sym)
subscribeTable(tableName="trades", actionName="sub1", offset=0, handler=streamFilter2, msgAsTable=true)
trades.append!(t)

select * from output1

select * from output2

select * from st2

"condition" can also be specified as Boolean expressions to support more complex filter logic. In the example above, replace the value of "condition" in filter2 with a Boolean expression to filter the data based on the columns "vol" and "date".

filter2 = dict(STRING,ANY)
filter2['condition'] = <sym==`A and 101<vol<105 and date<2012.12.15>
filter2['handler'] = st2
filter2['timeRange'] = 09:30:00.002:09:30:00.010

select * from st2