---
metadata:
  - name: generator
    content: Diplodoc Platform v5.56.0
alternate:
  - https://ydb.tech/docs/en/dev/streaming-query/enrichment.md?version=v26.2
  - https://ydb.tech/docs/ru/dev/streaming-query/enrichment.md?version=v26.2
  - href: en/dev/streaming-query/enrichment.md
    type: text/markdown
    title: Markdown version
  - href: ../../llms.txt
    type: text/markdown
    title: llms.txt
sourcePath: en/core/dev/streaming-query/enrichment.md
---
> **Documentation Index:** Fetch the complete configuration index at https://ydb.tech/docs/en/llms.txt

# Data enrichment

Data enrichment is adding additional information from a reference to events in a stream. For example, an event contains only an identifier, and the reference allows adding a name or other attributes to it. You can use data from a [local table](#enrichment-local-table) or from an [S3 object storage](#enrichment-s3) as a reference.

In streaming queries, the reference is connected using the `JOIN` construct. The stream must be on the left, the reference on the right.

{% note warning %}

The reference is fully loaded into memory when the query starts. If the data in the reference has changed, to get the current version of the reference, you need to restart the query — delete it using [DROP STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/drop-streaming-query.md?version=v26.2) and recreate it using [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=v26.2).

{% endnote %}

Data enrichment from [local and external topics](https://ydb.tech/docs/en/concepts/query_execution/topics.md?version=v26.2#local-external-topics) is possible.

In the examples below:

- `ext_source` — a pre-created [external data source](https://ydb.tech/docs/en/concepts/datamodel/external_data_source.md?version=v26.2) 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 queries in the examples below read events from the input topic, attach the service name from the reference by `ServiceId` to each event, and write the result to the output topic.

For more details on the functions used in the queries:

- [TableRow](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#tablerow)
- [Yson::From](https://ydb.tech/docs/en/yql/reference/udf/list/yson.md?version=v26.2#ysonfrom)
- [Yson::SerializeJson](https://ydb.tech/docs/en/yql/reference/udf/list/yson.md?version=v26.2#ysonserializejson)
- [Unwrap](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#unwrap)
- [ToBytes](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#to-from-bytes).

### Enrichment from a local table {#enrichment-local-table}

In this example, the reference is stored in the [table](https://ydb.tech/docs/en/concepts/datamodel/table.md?version=v26.2) `services_dict` in the current database.

Create a streaming query that performs enrichment:


```yql
CREATE STREAMING QUERY query_with_table_join AS
DO BEGIN

-- Reading events from 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 to 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 {#enrichment-s3}

The reference is stored in S3 and connected via an [external data source](https://ydb.tech/docs/en/concepts/query_execution/federated_query/s3/external_data_source.md?version=v26.2).

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


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


Where:

- `<s3_endpoint>` — the S3 storage URL, for example `https://storage.yandexcloud.net/<bucket>/` for Yandex Cloud.

Create a streaming query that performs enrichment:


```yql
CREATE STREAMING QUERY query_with_join AS
DO BEGIN

-- Reading events from 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 from S3
$s3_data = SELECT
    *
FROM
    s3_source.`file.csv`
WITH (
    FORMAT = csv_with_names,
    SCHEMA = (
        ServiceId Uint32,
        Name Utf8
    )
);

-- Joining reference to 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;

-- Write result to 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 more details on data formats (`json_each_row`, `csv_with_names`, etc.): [Data formats for reading/writing from topics](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=v26.2).