[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for PostgreSQL](../index.md) > [Tutorials](index.md) > Delivering data to Yandex Managed Service for Apache Kafka® using Yandex Data Transfer

# Delivering data to Yandex Managed Service for Apache Kafka® using Yandex Data Transfer

# Delivering data from Yandex Managed Service for PostgreSQL to Yandex Managed Service for Apache Kafka® using Yandex Data Transfer

You can track data changes in a Managed Service for PostgreSQL _source cluster_ and send them to a Managed Service for Apache Kafka® _target cluster_ using change data capture (CDC).

To set up CDC using Data Transfer:

1. [Prepare the source cluster](#prepare-source).
1. [Set up the target cluster](#prepare-target).
1. [Prepare and activate your transfer](#prepare-transfer).
1. [Test the transfer](#verify-transfer).

If you no longer need the resources you created, [delete them](#clear-out).


## Required paid resources {#paid-resources}

* Managed Service for PostgreSQL cluster: computing resources allocated to hosts, storage and backup size (see [Managed Service for PostgreSQL pricing](../pricing.md)).
* Managed Service for Apache Kafka® cluster: computing resources allocated to hosts, storage and backup size (see [Managed Service for Apache Kafka® pricing](../../managed-kafka/pricing.md)).
* Public IP addresses if public access is enabled for cluster hosts (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).
* Each transfer: use of computing resources and the number of transferred data rows (see [Data Transfer pricing](../../data-transfer/pricing.md)).


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

{% note info %}

Public access to cluster hosts is required if you plan to connect to the cluster via the internet. This connection option is simpler and is recommended for the purposes of this guide. You can connect to non-public hosts as well but only from Yandex Cloud virtual machines located in the same cloud network as the cluster.

{% endnote %}

1. [Create a Managed Service for PostgreSQL source cluster](../operations/cluster-create.md) with any suitable configuration, using the following settings:

    * Database: `db1`
    * User: `pg-user`
    * Hosts: Publicly available

1. [Create a Managed Service for Apache Kafka® target cluster](../../managed-kafka/operations/cluster-create.md) using any suitable configuration with publicly accessible hosts.


1. If using security groups, configure them to allow internet access to your clusters:

    * [Guide for Managed Service for PostgreSQL](../operations/connect/index.md#configuring-security-groups)
    * [Guide for Managed Service for Apache Kafka®](../../managed-kafka/operations/connect/index.md#configuring-security-groups)


1. Install the `kcat` (`kafkacat`) [utility](https://github.com/edenhill/kcat) and the [PostgreSQL command-line client](https://www.postgresql.org/download/) on your local machine. For example, on Ubuntu 20.04, use this command:

    ```bash
    sudo apt update && sudo apt install kafkacat postgresql-client --yes
    ```

    Check that you can use it to [connect to the Managed Service for Apache Kafka® source cluster over SSL](../../managed-kafka/operations/connect/clients.md#bash-zsh).

## Prepare the source cluster {#prepare-source}

1. For Data Transfer to receive data change notifications from a Managed Service for PostgreSQL cluster, you must create a publication on the source cluster. [Assign](../operations/grant.md) `pg-user` the `mdb_replication` role to allow publication creation.

1. [Connect](../operations/connect/index.md) to the `db1` database as `pg-user`.

1. Populate the database with test data. In this example, we will use a simple table with car sensor information.

    Create a table:

    ```sql
    CREATE TABLE public.measurements (
        "device_id" text PRIMARY KEY NOT NULL,
        "datetime" timestamp NOT NULL,
        "latitude" real NOT NULL,
        "longitude" real NOT NULL,
        "altitude" real NOT NULL,
        "speed" real NOT NULL,
        "battery_voltage" real,
        "cabin_temperature" real NOT NULL,
        "fuel_level" real
    );
    ```

    Populate the table with data:

    ```sql
    INSERT INTO public.measurements VALUES
        ('iv9a94th6rzt********', '2020-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
        ('rhibbh3y08qm********', '2020-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32),
        ('iv9a94th678t********', '2020-06-07 15:00:10', 55.70985913, 37.62141918,  417.0, 15.7, 10.3, 17, NULL);
    ```

## Set up the target cluster {#prepare-target}

The settings vary depending on the [topic management method](../../managed-kafka/concepts/topics.md#management) in use. Topic names follow the same convention as [Debezium](https://debezium.io/documentation/reference/connectors/postgresql.html#postgresql-topic-names): `<topic_prefix>.<schema_name>.<table_name>`. In this tutorial, we will use the `cdc` example prefix.

{% list tabs group=topic_management %}

- Yandex Cloud interfaces {#yc}

    When managing topics using native Yandex Cloud interfaces (management console, CLI, Terraform, or API):

    1. [Create a topic](../../managed-kafka/operations/cluster-topics.md#create-topic) named `cdc.public.measurements`.

        If you need to track data changes in multiple tables, create a separate topic for each one.

    1. [Create a user](../../managed-kafka/operations/cluster-accounts.md#create-account) named `kafka-user` with `ACCESS_ROLE_CONSUMER` and `ACCESS_ROLE_PRODUCER` roles applying to the created topics.

- Admin API {#api}

    When managing topics via the Kafka Admin API:

    1. Create an [admin user](../../managed-kafka/operations/cluster-accounts.md) named `kafka-user`.

    1. In addition to the `ACCESS_ROLE_ADMIN` role, assign the admin user the `ACCESS_ROLE_CONSUMER` and `ACCESS_ROLE_PRODUCER` roles for topics whose names begin with the `cdc` prefix.

        The system will automatically create the required topics when the first change occurs in the monitored tables of the source cluster. While this approach can be convenient for tracking changes across multiple tables, it requires reserving free storage space in your cluster. For more information, see [Storage in Managed Service for Apache Kafka®](../../managed-kafka/concepts/storage.md).

{% endlist %}


## Prepare and activate a transfer {#prepare-transfer}

1. [Create endpoints](../../data-transfer/operations/endpoint/index.md#create).

    * Source endpoint:

        * **Database type**: `PostgreSQL`.
        * **Endpoint parameters**:
            * **Connection settings**: `Managed Service for PostgreSQL cluster`.
            * **Managed Service for PostgreSQL cluster**: Your previously [created](#before-you-begin) Managed Service for PostgreSQL cluster.
            * **Database**: `db1`.
            * **User**: `pg-user`.
            * **Password**: `pg-user` password.
            * **Included tables**: `public.measurements`.

    * Target endpoint:

        * **Database type**: `Kafka`.
        * **Endpoint parameters**:
            * **Connection type**: `Managed Service for Apache Kafka cluster`.
                * **Managed Service for Apache Kafka cluster**: Select the target cluster.
                * **Authentication**: Specify the `kafka-user` credentials.

            * **Topic**: `Topic full name`.
            * **Topic full name**: `cdc.public.measurements`.

            If you need to track changes in multiple tables, fill in the fields as follows:

            * **Topic**: `Topic prefix`.
            * **Topic prefix**: Enter the `cdc` prefix you used to generate topic names.

1. [Create a transfer](../../data-transfer/operations/transfer.md#create) with the following settings:

    * **Endpoints**:
        * **Source**: Source endpoint you created earlier.
        * **Target**: Target endpoint you created earlier.
    * **Transfer type**: **Replication**.

1. [Activate the transfer](../../data-transfer/operations/transfer.md#activate) and wait for its status to change to **Replicating**.

## Test the transfer {#verify-transfer}

1. In a separate terminal, run `kafkacat` in consumer mode:

    ```bash
    kafkacat \
        -C \
        -b <broker_host_1_FQDN>:9091,...,<broker_host_N_FQDN>:9091 \
        -t cdc.public.measurements \
        -X security.protocol=SASL_SSL \
        -X sasl.mechanisms=SCRAM-SHA-512 \
        -X sasl.username=kafka-user \
        -X sasl.password=<password> \
        -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
        -Z \
        -K:
    ```

    You can get the FQDNs of broker hosts with the [list of hosts in the Managed Service for Apache Kafka® cluster](../../managed-kafka/operations/cluster-hosts.md).

    The utility will print the `public.measurements` table schema along with the data from previously inserted rows.

    {% cut "Message snippet example" %}

    ```json
    {
      "payload": {
        "consumer":"dttuhfpp97l3********"
      },
      "schema": {
        "fields": [
          {
            "field": "consumer",
            "optional":false,
            "type":"string"
          }
        ],
        "name": "__data_transfer_stub.public.__consumer_keeper.Key",
        "optional":false,
        "type":"struct"
      }
    }:{
      "payload": {
        "after": {
          "consumer":"dttuhfpp97l3********l",
          "locked_by":"dttuhfpp97l3********-1",
          "locked_till":"2022-05-15T09:55:18Z"
        },
      "before": null,
      "op":"u",
      "source": {
        "connector":"postgresql",
        "db":"db1",
        "lsn":85865797008,
        "name":"__data_transfer_stub",
        "schema":"public",
        "snapshot":"false",
        "table":"__consumer_keeper",
        "ts_ms":1652608518883,
        "txId":245165,
        "version":"1.1.2.Final",
        "xmin":null
      },
    ...
    ```

    {% endcut %}

1. Connect to the source cluster and populate the `measurements` table with data:

    ```sql
    INSERT INTO public.measurements VALUES ('iv7b74th678t********', '2020-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, 20.5, 5.3, 20, NULL);
    ```

1. Check that the terminal running `kafkacat` shows the information about the new row.

## Delete the resources you created {#clear-out}

To minimize resource consumption, delete the resources you no longer need:

1. [Deactivate](../../data-transfer/operations/transfer.md#deactivate) and [delete](../../data-transfer/operations/transfer.md#delete) the transfer.

1. [Delete the endpoints](../../data-transfer/operations/endpoint/index.md#delete).

1. Delete the clusters:

    * [Managed Service for Apache Kafka®](../../managed-kafka/operations/cluster-delete.md).
    * [Managed Service for PostgreSQL](../operations/cluster-delete.md).

1. If you used static public IP addresses to access the cluster hosts, release and [delete](../../vpc/operations/address-delete.md) them.