# createStreamDispatchEngine {#createstreamdispatchengine}

**Parent topic:**[Functions](../../Functions/category.md)

## Syntax {#syntax}

`createStreamDispatchEngine(name, dummyTable, keyColumn, outputTable, [dispatchType='hash'], [hashByBatch=false], [outputLock=true], [queueDepth=4096], [outputElapsedTime=false], [mode='buffer'])`

## Details {#details}

`createStreamDispatchEngine` function creates a stream dispatch engine that distributes incoming data streams to specified output tables for computational load balancing. The output tables can be in-memory tables, DFS tables, or other streaming engines. The function returns a table object.

Key characteristics of the stream dispatch engine:

-   Supports multithreaded input and output of streams.

-   Used only for data dispatching, not for metrics computing.


Typical usage:

The stream dispatch engine can distribute market data to one or more computational streaming engines that calculate factors. This achieves optimal performance by balancing the computational load.

## Arguments {#arguments}

**name** is a string indicating the name of the engine. It is the only identifier of an engine. It can contain letters, numbers and underscores but must start with a letter.

**dummyTable** is a table whose schema must be the same as the stream table to which the engine subscribes. Whether *dummyTable* contains data does not matter.

**keyColumn** is a string. If provided, the ingested data will be distributed to output tables based on the values in this column. Unique values in *keyColumn* are treated as keys.

**outputTable** is one or more tables that the engine outputs data to. When *outputElapsedTime* = false, *outputTable* must have the same schema as *dummyTable*; when *outputElapsedTime* = true, *outputTable* should have two additional columns - a LONG column and an INT column \(see *outputElapsedTime*\).

Up to 100 tables can be specified for *outputTable*. The engine starts a thread for each output table to process the distributed data. To specify multiple output tables, pass a tuple, embedding sub-tuples if needed. For examples:

-   To distribute evenly to 4 tables, specify `outputTable=[table1, table2, table3, table4]`

-   To distribute evenly to 2 replicated table sets, specify `[[table1_1, table1_2], [table2_1, table2_2]]`. This maintains 2 replicas of the ingested data - replica 1 distributed across table1\_1 and table1\_2, and replica 2 distributed across table2\_1 and table2\_2.


**dispatchType** \(optional\) is a string. It can be:

-   "hash" \(default\) - Apply a hash algorithm on *keyColumn* and distribute records based on the hash result. Hash distribution can be uneven across tables.

-   "uniform" - Evenly distribute records across output tables based on *keyColumn* values.

-   "saltedHash" - Apply a salted hash algorithm on *keyColumn* and distribute records based on the hash result. Salting ensures unique hashes even with the same input. This option is more suitable for scenarios that involve multi-level hash distribution \(e.g., a dispatch engine with all nested engines using hash for data distribution\).


The default "hash" is recommended unless data distribution is highly uneven and impacts performance. In that case, try "uniform".

**hashByBatch** \(optional\) is a Boolean value. The default is false, indicating that for each batch of data ingested into the engine, group records by *keyColumn* values first, then distribute groups across tables based on *dispatchType*.

To set *hashByBatch* to true, *dispatchType* must be 'hash'. In this case, for each ingested batch of data, the engine randomly selects a key, computes its hash value, and distributes the entire batch based on the hash result.

Note: Setting *hashbyBatch* = false ensures that records with identical keys are output to the same table. However, grouping records by key adds processing cost.

**outputLock** \(optional\) is a Boolean value indicating whether to apply a lock output table\(s\) to prevent concurrent write conflicts. The default is true \(recommended\). False means not to apply lock to the output table\(s\).

An output table, essentially an in-memory table, does not allow concurrent writes. As threads working for other streaming engines or subscriptions may also write to the output tables, the lock ensures thread safety. However, locking comes at a performance cost. If it can be guaranteed no other threads will write to the output tables concurrently, *outputLock* can be set to false to optimize performance.

**queueDepth** \(optional\) is a positive integer controlling the queue or buffer size for each output thread. The default is 4096 \(records\).

-   When *mode* = "buffer", *queueDepth* sets the size of the cache table for each thread of an output table;

-   When *mode* = "queue", *queueDepth* sets the maximum depth of each output queue.


Set *queueDepth* based on the expected data volume: if the ingested amount is small, a large *queueDepth* wastes memory; if the ingested amount is large, a small *queueDepth* may cause output blocking.

**outputElapsedTime** \(optional\) is a Boolean value indicating whether to print the elapsed time to process each ingested batch, from ingestion to output. The default is false. If *outputElapsedTime* = true, two extra columns are added to each output table: a LONG column for the time elapsed in microseconds to process each data batch internally, and an INT column for nanosecond timestamps of when each batch was output.

**mode** \(optional\) is a string. It can be:

-   "buffer" \(default\) - For each thread working for an output table, the engine creates an in-memory cache table to buffer pending writes. It copies data into the cache before writing to output table. Use this mode when \(1\) the input source\(s\) may have concurrent reads/writes while ingesting data into the engine; or \(2\) the input source\(s\) frequently appends small batches of data to the engine.

-   "queue" - For each thread working for an output table, the engine maintains a queue per with references to input data. Input data is not copied, only referenced. This requires no concurrent reads/writes to the input source\(s\) during ingestion. This mode is best when the input source\(s\) infrequently append large batches of data.


## Examples {#examples}

Distribute data of a stream table to 3 reactive state streaming engines for metric computation using the stream dispatch engine. The final results are output into one single table.

```
// define the input stream table for the stream dispatch engine
share streamTable(1:0, `sym`price, [STRING,DOUBLE]) as tickStream
share streamTable(1000:0, `sym`factor1, [STRING,DOUBLE]) as resultStream

// define the output table for the reactive state engines
for(i in 0..2){
rse = createReactiveStateEngine(name="reactiveDemo"+string(i), metrics =<cumavg(price)>, dummyTable=tickStream, outputTable=resultStream, keyColumn="sym")
}
// create the stream dispatch engine
dispatchEngine=createStreamDispatchEngine(name="dispatchDemo", dummyTable=tickStream, keyColumn=`sym, outputTable=[getStreamEngine("reactiveDemo0"),getStreamEngine("reactiveDemo1"),getStreamEngine("reactiveDemo2")])

// the stream dispatch engine subscribes to the stream table
subscribeTable(tableName=`tickStream, actionName="sub", handler=tableInsert{dispatchEngine}, msgAsTable = true)
    
// ingest data to the stream dispatch engine
n=100000
symbols=take(("A" + string(1..10)),n)
prices=100+rand(1.0,n)
t=table(symbols as sym, prices as price)
tickStream.append!(t)

select count(*) from resultStream
// output: 100,000

// check the status of the reactive state engines
getStreamEngineStat().ReactiveStreamEngine
```

|name|user|status|lastErrMsg|numGroups|numRows|numMetrics|memoryInUsed|snapshotDir|snapshotInterval|snapshotMsgId|snapshotTimestamp|
|----|----|------|----------|---------|-------|----------|------------|-----------|----------------|-------------|-----------------|
|reactiveDemo2|admin|OK|1|10,000|1|921|-1|||||
|reactiveDemo1|admin|OK|5|50,000|1|1,437|-1|||||
|reactiveDemo0|admin|OK|4|40,000|1|1,308|-1|||||

