StreamGraph::sourceByName

StreamGraph::sourceByName(name)

Returns a submitted shared stream table object in the stream graph by name.

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.

A DStream object.

First, create and submit the stream graph "aggregation".

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

aggGraph = createStreamGraph("aggregation")
aggGraph.source("trade", `time`sym`price, [TIMESTAMP, SYMBOL, FLOAT])
  .timeSeriesEngine(windowSize=60, step=60, metrics=[<sum(price) as price>], timeColumn="time", keyColumn="sym")
  .sink("aggregated")
aggGraph.submit()

In another stream graph, use sourceByName to get the output stream table "aggregated" from the submitted "aggregation" stream graph.

def EMA(S, N) {
	return ::ewmMean(S, span = N, adjust = false)
}
indicatorGraph = createStreamGraph("indicators")
indicatorGraph.sourceByName("aggregated")
  .reactiveStateEngine(metrics=[<EMA(price, 20)>], keyColumn=`sym)
  .sink("indicators")
indicatorGraph.submit()