DStream::map

DStream::map(func)

func A unary function that takes a table composed of streaming data messages as input and returns a table. Downstream engines infer processing logic based on the schema of the returned table. This function must be a pure function and cannot have any side effects such as writing to external tables.

Maps the input streams, i.e., applies the specified function to the input streams.

Return value: A DStream object.

Define map to generate OHLC data for AAPL:

use catalog test

g = createStreamGraph("graph")
g.source("trade", 1024:0, `symbol`datetime`price`volume, [SYMBOL,TIMESTAMP,DOUBLE,INT])
    .map(msg -> select * from msg where symbol == "AAPL")
    .timeSeriesEngine(60, 60, <[first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume]>, "datetime", false, "symbol")
    .sink("output")