StreamGraph::source

StreamGraph::source(name, capacity:size, colNames, colTypes, [asyncWrite=true], [compress=true], [cacheSize], [retentionMinutes=1440], [flushMode=0], [preCache], [cachePurgeTimeColumn], [cachePurgeInterval], [cacheRetentionTime])

Creates a persisted shared stream table. For details, see enableTableShareAndPersistence.

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.

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.

asyncWrite (optional) is a Boolean value indicating whether persistence is enabled in asynchronous mode. The default value is true, meaning asynchronous persistence is enabled. In this case, once data is written into memory, the write is deemed complete. The data stored in memory is then persisted to disk by another thread.

compress (optional) is a Boolean value indicating whether to save a table to disk in compression mode. The default value is true.

cacheSize (optional) is an integer used to determine the maximum number of records to retain in memory. If set to 0 or not specified, all records will be retained. Any positive integer smaller than 1000 will automatically be adjusted to 1000.

retentionMinutes (optional) is an integer indicating for how long (in minutes) a log file larger than 1GB will be kept after last update. The default value is 1440, which means the log file is kept for 1440 minutes, i.e., 1 day.

flushMode (optional) is an integer indicating whether to enable synchronous disk flush. It can be 0 or 1. The persistence process first writes data from memory to the page cache, then flushes the cached data to disk. If flushMode is 0 (default), asynchronous disk flushing is enabled. In this case, once data is written from memory to the page cache, the flush is deemed complete and the next batch of data can be written to the table. If flushMode is set to 1, the current batch of data must be flushed to disk before the next batch can be written.

preCache (optional) is an integer indicating the number of records to be loaded into memory from the persisted stream table on disk when DolphinDB restarts. If it is not specified, all records are loaded into memory when DolphinDB restarts.

cachePurgeTimeColumn (optional) is a STRING scalar indicating the time column in the stream table.

cachePurgeInterval (optional) is a DURATION scalar indicating the interval to trigger cache purge.

cacheRetentionTime (optional) is a DURATION scalar indicating the retention time of cached data. Note: Since version 3.00.2/2.00.14, cacheRetentionTime must be smaller than cachePurgeInterval.

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

g = createStreamGraph("indicators")
g.source("trade", 1:0, `time`symbol`price`volume, [DATETIME,SYMBOL,DOUBLE,LONG])