createNarrowReactiveStateEngine

Syntax

createNarrowReactiveStateEngine(name, metrics, metricNames, dummyTable, outputTable, keyColumn, [filter], [snapshotDir], [snapshotIntervalInMsgCount], [keepOrder], [keyPurgeFilter], [keyPurgeFreqInSecond=0], [raftGroup], [outputElapsedMicroseconds=false], [keyCapacity=1024], [parallelism=1])

Details

Creates a reactive state engine that returns a table in narrow format. Its calculation rules and method are the same as those of createReactiveStateEngine, but its output schema differs: It outputs results of multiple factors to rows in a single result column, whereas createReactiveStateEngine outputs the results to separate columns.

Note:
Out-of-order processing is not supported. You must ensure that the data is ordered.

Parameters

Except as otherwise noted on this page, most parameters of createNarrowReactiveStateEngine have the same meanings as the corresponding parameters of createReactiveStateEngine. Only parameters specific to this function and unsupported features are described below.

metrics is metacode or a tuple of metacode containing columns from the input table (excluding keyColumn, optional) or factors (formulas for calculation, required).

metricNames is a STRING scalar or vector, indicating the name for each factor specified in metrics. The number and order of names must align to that of factors specified in metrics.

outputTable is the output table for the results. It can be an in-memory table or a DFS table. Create an empty table and specify the column names and types before calling the function.

The output columns are in the following order:

(1) The first few columns must be in the same order as that of keyColumn.

(2) If the outputElapsedMicroseconds is set to true, specify two more columns: a LONG column and an INT column.

(3) The references to columns from the input table specified in metrics.

(4) A single column containing metricNames.

(5) Then followed by one result column.

Note:

High availability is not supported for this engine; therefore, the parameter raftGroup is not supported.

Starting from version 3.00.4, specify snapshotDir and snapshotIntervalInMsgCount to enable snapshots for this engine.

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 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.

The file extension of a snapshot can be:
  • <engineName>.tmp: temporary snapshot
  • <engineName>.snapshot: a snapshot that is generated and flushed to disk
  • <engineName>.old: if a snapshot with the same name already exists, the previous snapshot is renamed to <engineName>.old.

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

Returns

A table object.

Examples

Calculate the cumulative volume and the moving average and output the results of both factors to a single column.

dummy = streamTable(1:0, ["securityID1","securityID2","securityID3","createTime","updateTime","upToDatePrice","qty","value"], [STRING,STRING,STRING,TIMESTAMP,TIMESTAMP,DOUBLE,DOUBLE,INT]) 
share streamTable(1:0,["securityID1","securityID2","securityID3","createTime","updateTime","metricNames","factorValue"], [STRING,STRING,STRING, TIMESTAMP,TIMESTAMP,STRING,DOUBLE]) as outputTable
// Define two factors: cumulative volume and the moving average
factor = [<createTime>, <updateTime>,<cumsum(qty)>,<cumavg(upToDatePrice)>]
Narrowtest = createNarrowReactiveStateEngine(name="narrowtest1",metrics=factor,metricNames=["factor1","factor2"],dummyTable=dummy,outputTable=outputTable,keyColumn=["securityID1","securityID2","securityID3"])

num = 5
tmp = table(take("A" + lpad(string(1..4),4,"0"),num) as securityID1,take("CC.HH" + lpad(string(21..34),4,"0"),num) as securityID2,take("FFICE" + lpad(string(13..34),4,"0"),num) as securityID3, 2023.09.01 00:00:00+(1..num) as createTime, 2023.09.01 00:00:00+(1..num) as updateTime,100.0+(1..num) as upToDatePrice, 130.0+(1..num) as qty,take(1..3,num) as value)
Narrowtest.append!(tmp)

select * from outputTable

/*
securityID1	securityID2	securityID3	createTime	updateTime	metricNames	factorValue
A0001	CC.HH0021	FFICE0013	2023.09.01T00:00:01.000	2023.09.01T00:00:01.000	factor1	131
A0001	CC.HH0021	FFICE0013	2023.09.01T00:00:01.000	2023.09.01T00:00:01.000	factor2	101
A0002	CC.HH0022	FFICE0014	2023.09.01T00:00:02.000	2023.09.01T00:00:02.000	factor1	132
A0002	CC.HH0022	FFICE0014	2023.09.01T00:00:02.000	2023.09.01T00:00:02.000	factor2	102
A0003	CC.HH0023	FFICE0015	2023.09.01T00:00:03.000	2023.09.01T00:00:03.000	factor1	133
A0003	CC.HH0023	FFICE0015	2023.09.01T00:00:03.000	2023.09.01T00:00:03.000	factor2	103
A0004	CC.HH0024	FFICE0016	2023.09.01T00:00:04.000	2023.09.01T00:00:04.000	factor1	134
A0004	CC.HH0024	FFICE0016	2023.09.01T00:00:04.000	2023.09.01T00:00:04.000	factor2	104
A0001	CC.HH0025	FFICE0017	2023.09.01T00:00:05.000	2023.09.01T00:00:05.000	factor1	135
A0001	CC.HH0025	FFICE0017	2023.09.01T00:00:05.000	2023.09.01T00:00:05.000	factor2	105
*/

Related functions: addReactiveMetrics, getReactiveMetrics