StreamGraph::haKeyedSource

StreamGraph::haKeyedSource(name, keyColumn, capacity:size, colNames, colTypes, raftGroup, cacheLimit, [retentionMinutes=1440])

Creates a high-availability keyed stream table.

Return value: A DStream object.

name is a string representing the name of the Orca stream table. You can provide either the fully qualified name (FQN), such as "trading.orca_table.factors", or just the table name, like "factors". If only the name is given, the system will automatically complete it using the current catalog.

keyColumn (optional) is a string indicating the name of the primary key column.

capacity is a positive integer indicating the amount of memory (in terms of the number of rows) allocated to the table. When the number of rows exceeds capacity, the system will first allocate memory of 1.2~2 times of capacity, copy the data to the new memory space, and release the original memory. For large tables, these steps may use significant amount of memory.

size is an integer no less than 0 indicating the initial size (in terms of the number of rows) of the table. If size=0, create an empty table; If size>0, the initialized values are:

• false for Boolean type;

• 0 for numeric, temporal, IPADDR, COMPLEX, and POINT types;

• Null value for Literal, INT128 types.

colNames is a STRING vector of column names.

colTypes is a STRING vector of data types. It can use either the reserved words for data types or corresponding strings.

raftGroup is an integer greater than 1 indicating Raft group ID.

cacheLimit is an integer representing the maximum number of rows of the high-availability stream table to be kept in memory. If cacheLimit>1000, it is automatically adjusted to 1000.

retentionMinutes (optional) is an integer indicating for how long (in terms of minutes) a log file larger than 1GB will be kept after last update. The default value is 1440, which means the log file only keeps data in the past 24 hours.

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

g = createStreamGraph("indicators")
g.haKeyedSource("ha_keyedTable",`symbol, 1:0, `time`symbol`price`volume, [DATETIME,SYMBOL,DOUBLE,LONG], 3, 50000)