Data enrichment

Data enrichment means augmenting stream events with extra fields from a lookup dataset. For example, an event may contain only an identifier, and the lookup adds a human-readable name or other attributes. The lookup can be a local table or object storage (S3).

In streaming queries, the lookup is attached with a JOIN. The stream must be on the left side of the join, the lookup on the right.

Warning

The lookup is fully loaded into memory when the query starts. If the lookup data changes, restart the query to pick up the new version: drop it with DROP STREAMING QUERY and create it again with CREATE STREAMING QUERY.

You can enrich data from local and external topics.

In the examples below:

  • ext_source — a pre-created external data source for topics in another YDB database;
  • input_topic and output_topic — topics in the current or external YDB database

Streaming queries for data enrichment

The examples below read events from an input topic, join each event with a service name from the lookup on ServiceId, and write the result to an output topic.

Functions used in the queries:

Enrichment from a local table

In this example the lookup is stored in a table services_dict in the current database.

Create a streaming query that performs enrichment:

CREATE STREAMING QUERY query_with_table_join AS
DO BEGIN

-- Reading events from the input topic
$topic_data = SELECT
    *
FROM
    ext_source.input_topic -- or local topic input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Time String NOT NULL,
        ServiceId Uint32 NOT NULL,
        Message String NOT NULL
    )
);

-- Joining reference data to the stream by ServiceId
$joined_data = SELECT
    s.Name AS Name,
    t.*
FROM
    $topic_data AS t
LEFT JOIN
    services_dict AS s
ON
    t.ServiceId = s.ServiceId;

-- Writing to the output topic (JSON)
INSERT INTO
    output_topic -- or external topic ext_source.output_topic
SELECT
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(TableRow()))))
FROM
    $joined_data;

END DO

Enrichment from S3

The lookup is stored in S3 and connected via an external data source.

Create an additional external data source to read the lookup from S3:

-- S3 data source for reading reference data
CREATE EXTERNAL DATA SOURCE s3_source WITH (
    SOURCE_TYPE = "ObjectStorage",
    LOCATION = "<s3_endpoint>",
    AUTH_METHOD = "NONE"
);

Where:

  • <s3_endpoint> is the S3 endpoint URL, for example https://storage.yandexcloud.net/<bucket>/ in Yandex Cloud.

Create a streaming query that performs enrichment:

CREATE STREAMING QUERY query_with_join AS
DO BEGIN

-- Reading events from the input topic
$topic_data = SELECT
    *
FROM
    input_topic -- or external topic ext_source.input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Time String NOT NULL,
        ServiceId Uint32 NOT NULL,
        Message String NOT NULL
    )
);

-- Reading service reference data from S3
$s3_data = SELECT
    *
FROM
    s3_source.`file.csv`
WITH (
    FORMAT = csv_with_names,
    SCHEMA = (
        ServiceId Uint32,
        Name Utf8
    )
);

-- Joining reference data to the stream by ServiceId
$joined_data = SELECT
    s.Name AS Name,
    t.*
FROM
    $topic_data AS t
LEFT JOIN
    $s3_data AS s
ON
    t.ServiceId = s.ServiceId;

-- Writing the result to the output topic in JSON format
INSERT INTO
    ext_source.output_topic -- or local topic output_topic
SELECT
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(TableRow()))))
FROM
    $joined_data;

END DO

For supported data formats (json_each_row, csv_with_names, and others), see Data formats for reading/writing from topics.

Previous
Writing to Tables
Next
Data formats for reading/writing topics