DStream::map

DStream::map(func)

Applies the specified function to the input streams. It is suitable for stateless data processing scenarios (e.g., data filtering).

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.

Note: This function must be a pure function, meaning it must satisfy the following properties:

• Deterministic: Given the same input, it always produces the same output.
• No side effects: The function must not cause any observable interaction with the outside world. Specifically, it must not:
  • Modify global or external variables.
  • Write to databases or files.
  • Call external APIs or services.
  • Mutate the input table (e.g., by modifying its values in-place).

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")

Related function: DStream::udfEngine