DStream::sharedKeyedTable

DStream::sharedKeyedTable(name, keyColumns, X, [X1], [X2], .....)

or

DStream::sharedKeyedTable(name, keyColumns, capacity:size, colNames, colTypes)

or

DStream::sharedKeyedTable(name, keyColumns, table)

Creates a shared keyed table in Orca, which is only usable within DStream::udfEngine. For details about keyed tables, refer to keyedTable.

name is a STRING scalar indicating the shared keyed table name.

keyColumn is a string scalar or vector indicating the name(s) of the primary key column(s). The column type must be INTEGRAL, TEMPORAL or LITERAL.

For the third scenario, table is a table. Please note that keyColumns in table cannot have duplicate values.

A keyed table.

In this example, we use DStream::sharedKeyedTable and DStream::udfEngine to implement a historical delta computation.

The DStream::sharedKeyedTable is used to maintain the most recent record for each ID. When a new record arrives, if an entry with the same id already exists in the table, the UDF outputs the difference between the new value and the previously stored (historical) value. If the ID does not exist, the new record is inserted into the table without producing any output.

if(existsCatalog("orcaCatalog")) dropCatalog("orcaCatalog")
createCatalog("orcaCatalog")
go
use catalog orcaCatalog
// Create stream graph
g = createStreamGraph("compare")
g.sharedKeyedTable("history", "id", 1:0, `id`value, [INT, DOUBLE])
g.source("data", `id`value`time, [INT, DOUBLE, TIMESTAMP])
  .udfEngine(def(msg) {
    history = orcaObj("history")
    diffTable = table(100:0, `id`diff, [INT, DOUBLE])
    for(i in 0:msg.size()) {
        idVal = msg.id[i]
        valueVal = msg.value[i]
        // Read historical value
        old = select value from history where id = idVal
        // Write new value
        newRow = table(idVal as id, valueVal as value)
        history.append!(newRow)
        // Compute delta
        if(old.size() > 0) {
            diffTable.append!(table(idVal as id, (valueVal - old.value[0]) as diff))
        }
    }
    return diffTable
  })
  .sink("comparison")
g.submit()
// Generate mock data
mockData = table(1..5 as id, rand(100.0, 5) as value, now() + 1..5 as time)
// Insert data
appendOrcaStreamTable("orcaCatalog.orca_table.data", mockData)
//  Generate data with duplicate IDs
mockData = table(1..5 as id, rand(100.0, 5) as value, now() + 1..5 as time)
// Insert data
appendOrcaStreamTable("orcaCatalog.orca_table.data", mockData)
// Wait for processing and inspect results
sleep(1000)
select * from orcaCatalog.orca_table.comparison