[Yandex Cloud documentation](../../index.md) > [Yandex Query](../index.md) > [Tutorials](index.md) > Processing Debezium CDC streams

# Processing Debezium CDC streams

[Debezium](https://debezium.io) is a change data capture (CDC) tool that sends database changes to other systems for processing. You can use Yandex Data Streams to capture these changes and Yandex Query to process them. You can do the following with processed data:

* Send it to Yandex Monitoring to make charts and use it in alerting.
* Write it to a stream in Data Streams and then send to Yandex Cloud Functions for processing.
* Write it to a stream in Data Streams and then transfer to Yandex Data Transfer to then [distribute to various storage systems](../../data-streams/tutorials/data-ingestion.md).

![debezium-architecture](../../_assets/query/debezium-architecture.png)

In this use case, you will send [PostgreSQL](https://www.postgresql.org/) database changes to a stream in Data Streams using Debezium and then query them with Query. The query will return the number of changes in DB tables grouped by 10s interval. It is assumed that Debezium is installed on the server with PostgreSQL set up and running.

For this tutorial:

1. [Create a data stream in Data Streams](#create-yds-stream).
1. [Set the stream connection credentials](#credentials).
1. [Set up Debezium Server](#debezium-server).
1. [Connect Query to your data stream](#connect-query).
1. [Query the data](#query).

## Getting started {#before-you-begin}

Sign up for Yandex Cloud and create a [billing account](../../billing/concepts/billing-account.md):
1. Navigate to the [management console](https://console.yandex.cloud) and log in to Yandex Cloud or create a new account.
1. On the **[Yandex Cloud Billing](https://center.yandex.cloud/billing/accounts)** page, make sure you have a billing account linked and it has the `ACTIVE` or `TRIAL_ACTIVE` [status](../../billing/concepts/billing-account-statuses.md). If you do not have a billing account, [create one](../../billing/quickstart/index.md) and [link](../../billing/operations/pin-cloud.md) a cloud to it.

If you have an active billing account, you can create or select a [folder](../../resource-manager/concepts/resources-hierarchy.md#folder) for your infrastructure on the [cloud page](https://console.yandex.cloud/cloud).

[Learn more about clouds and folders here](../../resource-manager/concepts/resources-hierarchy.md).

## Create a data stream in Data Streams {#create-yds-stream}

[Create a data stream](../../data-streams/operations/manage-streams.md#create-data-stream) named `debezium`.

## Set the stream connection credentials {#credentials}

1. [Create](../../iam/operations/sa/create.md) a service account and [assign](../../iam/operations/sa/assign-role-for-sa.md) it the `editor` role for your folder.
1. [Create](../../iam/operations/authentication/manage-access-keys.md#create-access-key) a static access key.
1. On the server where PostgreSQL is set up and running, configure the [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-welcome.html):
    1. [Install the AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) and run this command:

        ```bash
        aws configure
        ```

    1. Enter the following, one by one:

        * `AWS Access Key ID [None]:`: Service account [key ID](../../iam/concepts/authorization/access-key.md).
        * `AWS Secret Access Key [None]:`: Service account [secret key](../../iam/concepts/authorization/access-key.md).
        * `Default region name [None]:`: `ru-central1` availability zone.

## Set up Debezium Server {#debezium-server}

On the server where PostgreSQL is set up and running:

1. Install the Debezium Server using [this guide](https://debezium.io/documentation/reference/stable/operations/debezium-server.html).
1. Go to the `conf` folder and create a file named `application.properties` with the following contents:

    ```text
    debezium.sink.type=kinesis
    debezium.sink.kinesis.region=ru-central1
    debezium.sink.kinesis.endpoint=<endpoint>
    debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
    debezium.source.offset.storage.file.filename=data/offsets.dat
    debezium.source.offset.flush.interval.ms=0
    debezium.source.database.hostname=localhost
    debezium.source.database.port=5432
    debezium.source.database.user=<user_name>
    debezium.source.database.password=<user_password>
    debezium.source.database.dbname=<DB_name>
    debezium.source.database.server.name=debezium
    debezium.source.plugin.name=pgoutput

    debezium.source.transforms=Reroute
    debezium.source.transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
    debezium.source.transforms.Reroute.topic.regex=(.*)
    debezium.source.transforms.Reroute.topic.replacement=<data_stream>
    ```

    Where:

    * `<endpoint>`: Endpoint for a Data Streams data stream, e.g., `https://yds.serverless.yandexcloud.net/ru-central1/b1g89ae43m6he********/etn01eg4rn1********`. You can see the endpoint on the stream page (see [Viewing a list of streams](../../data-streams/operations/manage-streams.md#list-data-streams)).
    * `<data_stream>`: Data Streams data stream name.
    * `<DB_name>`: PostgreSQL database name.
    * `<user_name>`: User name for connecting to the PostgreSQL database.
    * `<user_password>`: User password for connecting to the PostgreSQL database.
1. Run Debezium using this command:

    ```bash
    JAVA_OPTS=-Daws.cborEnabled=false ./run.sh
    ```

1. Make some changes to the PostgreSQL database, e.g., insert data into a table.
1. If the settings are correct, the Debezium console will show messages like:

    ```text
    2022-02-11 07:31:12,850 INFO  [io.deb.con.com.BaseSourceTask] (pool-7-thread-1) 1 records sent during previous 00:19:59.999, last recorded offset: {transaction_id=null, lsn_proc=23576408, lsn_commit=23576120, lsn=23576408, txId=580, ts_usec=1644564672582666}
    ```

## Connect Query to your data stream {#connect-query}

1. [Create a connection](../operations/connection.md#create) named `yds-connection` of the `Data Streams` type.
1. On the binding creation page:
    * Enter the binding name: `debezium`.
    * Specify the data stream: `cdebezium`.
    * Add a column titled `data` with `JSON` for type.
1. Click **Create**.

## Query the data {#query}

Open the query editor in the Query interface and run this query:

```sql
$debezium_data = 
SELECT 
    JSON_VALUE(data,"$.payload.source.table") AS table_name, 
    DateTime::FromMilliseconds(cast(JSON_VALUE(data,"$.payload.source.ts_ms") AS Uint64)) AS `timestamp`
FROM bindings.`debezium`;

SELECT 
    table_name, 
    HOP_END() 
FROM 
    $debezium_data 
GROUP BY 
    HOP(`timestamp`, "PT10S", "PT10S", "PT10S"),
    table_name
LIMIT 2;
```

{% note info %}

Data from a streaming source is delivered as an infinite stream. To prevent infinite streaming and get output in the console, the example uses the `LIMIT` clause that limits the number of result rows.

{% endnote %}

## Useful links {#see-also}

* [Reading data from Data Streams via Query connections](../sources-and-sinks/data-streams.md)