Delivering data from Yandex Managed Service for PostgreSQL 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.
2. Prepare the target cluster.
3. Prepare and activate the transfer.
4. Test the transfer.

If you no longer need these resources, delete them.

Before you begin

1. Create a source Managed Service for PostgreSQL cluster in any suitable configuration with the following settings:

   • With the db1 database.
   • With the pg-user user.
   • With publicly available hosts.

2. Create a Managed Service for Apache Kafka® target cluster in any applicable configuration with publicly available hosts.

3. Configure security groups for your clusters so you can connect to them from the internet:

   • Instructions for Managed Service for PostgreSQL
   • Instructions for Managed Service for Apache Kafka®

4. Install the kcat (kafkacat) utility and PostgreSQL command-line client on the local machine. For example, in Ubuntu 20.04, run:

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

Prepare the source cluster

1. For Data Transfer to get notifications of data changes from a Managed Service for PostgreSQL cluster, create a publication in a source cluster. The pg-user user can create a publication after you assign them the mdb_replication role.

2. Connect to the db1 database on behalf of pg-user.

3. Add test data to the database. In this example, a simple table with information from car sensors is used.

   Create a table:

   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:

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

Prepare the target cluster

The settings vary depending on the topic management method used. The same conventions are used for topic names as in Debezium: <topic prefix>.<schema name>.<table name>. In this tutorial, the cdc prefix is used as an example.

Prepare and activate the transfer

1. Create an endpoint.

   • Source endpoint:

     • DB type: PostgreSQL.
     • Endpoint parameters:
       • Connection settings: MDB cluster.
       • MDB cluster: Select the Managed Service for PostgreSQL cluster you created earlier.
       • Database name: db1.
       • Username: pg-user.
       • Password: Enter the password for pg-user.
       • List of included tables: public.measurements.

   • Target endpoint:

     • DB type: Kafka.

     • Endpoint parameters:

       • Connection: Managed Kafka.

       • Managed Kafka:

         • Managed Kafka cluster ID: Select a target cluster.
         • Authentication: Enter the details of the created kafka-user user.

       • Kafka topic settings: Full topic name.

       • Full topic name: cdc.public.measurements.

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

       • Kafka topic settings: Topic prefix.
       • Topic prefix: Enter the cdc prefix you used to generate topic names.

2. Create a transfer with the following settings:

   • Endpoints:
     • Source: The previously created endpoint for the source.
     • Target: The previously created endpoint for the target.
   • Transfer type: Increment.

3. Activate the transfer and wait for its status to change to Replicating.

Test the transfer

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

   kafkacat \
       -C \
       -b <FQDN broker-host-1>:9091,...,<FQDN broker-host N>: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/YandexCA.crt \
       -Z \
       -K:
   

   You can obtain the FQDNs of broker hosts with a list of hosts in the Managed Service for Apache Kafka® cluster.

   The data format schema of the public.measurements table and the information about the previously added rows will be printed.

   Example of the message fragment

   {
     "payload": {
       "consumer":"dttuhfpp97l30jaka3ql"
     },
     "schema": {
       "fields": [
         {
           "field": "consumer",
           "optional":false,
           "type":"string"
         }
       ],
       "name": "__data_transfer_stub.public.__consumer_keeper.Key",
       "optional":false,
       "type":"struct"
     }
   }:{
     "payload": {
       "after": {
         "consumer":"dttuhfpp97l30jaka3ql",
         "locked_by":"dttuhfpp97l30jaka3ql-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
     },
   ...
   

2. Connect to the source cluster and add data to the measurements table:

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

3. Make sure the terminal running kafkacat displays details about the added row.

Delete the resources you created

If you no longer need these resources, delete them:

1. Deactivate and delete the transfer.

2. Delete the endpoints.

3. Delete the clusters:

   • Managed Service for Apache Kafka®.
   • Managed Service for PostgreSQL.

4. If static public IP addresses were used for accessing the cluster hosts, release and delete them.