createNearestJoinEngine

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

Creates a nearest-neighbor join streaming engine. The engine groups the left and right input tables by the columns specified in matchingColumn, and for each row in the left table, it retrieves the k nearest records from the right table (with timestamps not later than the current row) within each group, and computes the output based on these records in real time.

Return value: A table.

name is a string indicating the name of the nearest-neighbor join streaming engine. It is the unique identifier of the engine on a data/compute node. It can contain letters, numbers and underscores and must start with a letter.

leftTable and rightTable are table objects whose schema must be the same as the stream table to which the engine subscribes.

outputTable is a table to which the engine inserts calculation result. It can be an in-memory table or a DFS table. Before calling a function, an empty table with specified column names must be created.

The columns of outputTable are in the following order:

(1) The first column must be a temporal column.

• if useSystemTime = true, the data type must be TIMESTAMP;

• if useSystemTime = false, it has the same data type as timeColumn.

(2) Then followed by one or more columns on which the tables are joined, arranged in the same order as specified in matchingColumn.

(3) Further followed by one or more columns which are the calculation results of metrics.

kNearest is a positive integer. For each row in the left table, the engine retrieves the k nearest rows from the right table whose timestamps are less than or equal to the current row's timestamp.

metrics is metacode (which can be a tuple) specifying the calculation formulas. For more information about metacode, please refer to Metaprogramming.

• metrics can use one or more expressions, built-in or user-defined functions.

• metrics can be functions that return multiple values and the columns in the output table to hold the return values must be specified. For example, <func(price) as `col1`col2>.

• The columns in right table can be converted into array vectors using toArray, e.g., <toArray(price)>.

• The column names specified in metrics are not case-sensitive and can be inconsistent with the column names of the input tables.

If you want to specify a column that exists in both the left and the right tables, use the format tableName.colName. By default, the column from the left table is used.

The following functions are optimized in the engine when they are applied only to the columns from the right table: sum, sum2, avg, std, var, corr, covar, wavg, wsum, beta, max, min, last, first, med, percentile.

timeColumn (optional) When useSystemTime = false, it must be specified to indicate the name(s) of the time column in the left table and the right table. The time columns must have the same data type. If the names of the time column in the left table and the right table are the same, timeColumn is a string. Otherwise, it is a vector of 2 strings indicating the time column in each table.

garbageSize (optional) is a positive integer with the default value of 5,000 (rows). As the subscribed data is ingested into the engine, it continues to take up the memory. Within the left/right table, the records are grouped by matchingColumn values; When the number of records in a group exceeds garbageSize, the system will remove those already been calculated from memory.

maxDelayedTime (optional) is a positive integer. maxDelayedTime only takes effect when timeColumn is specified and the two arguments must have the same time precision. Use maxDelayedTime to trigger windows which remain uncalculated long past its end. The default maxDelayedTime is 3 seconds. For more information about this parameter, see "Window triggering rules" in the Details section.

nullFill (optional) is a tuple of the same size as the number of output columns. It is used to fill in the null values in the output table. The data type of each element corresponds to each output column.

cachedTableCapacity (optional) is a positive integer indicating the initial capacity (in terms of the number of rows) of the left and right cache tables for each group. The default value is 1024.

To enable snapshot in the streaming engines, specify parameters snapshotDir and snapshotIntervalInMsgCount.

snapshotDir (optional) is a string indicating the directory where the streaming engine snapshot is saved. The directory must already exist, otherwise an exception is thrown. If the snapshotDir is specified, the system checks whether a snapshot already exists in the directory when creating a streaming engine. If it exists, the snapshot will be loaded to restore the engine state. Multiple streaming engines can share a directory where the snapshot files are named as the engine names.

snapshotIntervalInMsgCount (optional) is a positive integer indicating the number of messages to receive before the next snapshot is saved.

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