---
metadata:
  - name: generator
    content: Diplodoc Platform v5.50.6
alternate:
  - https://ydb.tech/docs/en/dev/streaming-query/enrichment.md?version=main
  - https://ydb.tech/docs/ru/dev/streaming-query/enrichment.md?version=main
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 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](#enrichment-local-table) or [object storage (S3)](#enrichment-s3).

In [streaming queries](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main), the lookup is attached with a `JOIN`. The stream must be on the left side of the join, the lookup on the right.

{% note 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](https://ydb.tech/docs/en/yql/reference/syntax/drop-streaming-query.md?version=main) and create it again with [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=main).

{% endnote %}

You can enrich data from [local and external topics](https://ydb.tech/docs/en/dev/streaming-query/local-and-external-topics.md?version=main).

In the examples below:

- `ext_source` — a pre-created [external data source](https://ydb.tech/docs/en/concepts/datamodel/external_data_source.md?version=main) 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:

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

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

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

Create a [streaming query](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main) that performs enrichment:


```yql
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 {#enrichment-s3}

The lookup 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=main).

Create an additional [external data source](https://ydb.tech/docs/en/yql/reference/syntax/create-external-data-source.md?version=main) to read the lookup from S3:


```yql
-- 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](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main) that performs enrichment:


```yql
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](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=main).