Common streaming query patterns

This section collects minimal examples of streaming queries for typical scenarios. It starts with a basic topic read, then shows end-to-end processing: handling data and writing results to a topic as JSON, to a topic as a plain string, and to a table. Each example can be used as a starting point for your own workloads.

⟦C1⟧ — topic to read from;

Data is read from a topic using SELECT ... FROM ... WITH (FORMAT, SCHEMA). The WITH block specifies the input data format and schema — which fields are expected in each message and their types. This pattern is used in all subsequent examples.

Note

Working with local and external topics is shown.

In the examples:

  • ext_source — a pre-created external data source.
  • input_topic — the topic from which data is read.
  • output_topic — the topic where results are written.
  • output_table — the YDB table where results are written.

The following fragment shows reading events from a topic in JSON format. It is used inside CREATE STREAMING QUERY in the DO BEGIN ... END DO block:

SELECT
    *
FROM
    ext_source.input_topic -- or local topic input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);

The following snippet reads JSON events from a topic. Use it inside CREATE STREAMING QUERY in a ⟦C1⟧ block:

Writing to a topic (JSON)

The query reads events from the input topic, builds a JSON object from fields, and writes to the output topic. AsStruct builds a structure from the fields, Yson::From converts it to Yson, Yson::SerializeJson serializes to a JSON string, and ToBytes converts to String, which is required for topic writes.

CREATE STREAMING QUERY write_json_example AS
DO BEGIN

INSERT INTO ext_source.output_topic -- or local topic output_topic
SELECT
    -- Forming JSON from individual fields
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(
        AsStruct(Id AS id, Name AS name)
    ))))
FROM
    input_topic -- or external topic ext_source.input_topic
WITH (
    FORMAT = json_each_row,  -- Input data format
    SCHEMA = (               -- Input data schema
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);

END DO

More on the functions:

Writing to a topic (string)

The query reads events from the input topic and writes a single field as a string to the output topic. Topic writes require SELECT to return a single column of type String or Utf8.

CREATE STREAMING QUERY write_utf8_example AS
DO BEGIN

INSERT INTO output_topic -- or external topic ext_source.output_topic
SELECT
    Name
FROM
    ext_source.input_topic -- or local topic input_topic
WITH (
    FORMAT = json_each_row,  -- Input data format
    SCHEMA = (               -- Input data schema
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);

END DO

More on write formats: Formats for writing data.

Writing to a table

The query reads events from a topic and writes them to output_table. Create the table beforehand with a schema that matches the selected columns.

Warning

Table writes in streaming queries support UPSERT only. INSERT INTO is not supported: with at-least-once delivery, retries would duplicate rows. With UPSERT, an existing row with the same primary key is updated; otherwise a new row is inserted, while INSERT INTO fails.

CREATE STREAMING QUERY write_table_example AS
DO BEGIN

-- Writing to table (only UPSERT, INSERT not supported)
UPSERT INTO output_table
SELECT
    Id,
    Name
FROM
    ext_source.input_topic -- or local topic input_topic
WITH (
    FORMAT = json_each_row,  -- Input data format
    SCHEMA = (               -- Input data schema
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);

END DO

More details: Writing to tables.

See also

Previous
Local and External Topics
Next
Writing to Tables