Get started
Yandex Data Transfer

Delivering data from Yandex Managed Service for Apache Kafka® to Yandex Managed Service for ClickHouse

Written by

A Managed Service for ClickHouse cluster can get data from Apache Kafka® topics in real time. Managed Service for ClickHouse automatically inserts data sent to ClickHouse tables on the Kafka engine.

To set up data delivery from Managed Service for Apache Kafka® to Managed Service for ClickHouse:

  1. Send test data to Managed Service for Apache Kafka® topic.
  2. Prepare and activate the transfer.
  3. Test the transfer.

If you no longer need these resources, delete them.

Before you begin

Prepare the infrastructure

  1. Create a source Managed Service for Apache Kafka® cluster of any suitable configuration. To connect to the cluster from the user's local machine instead of the Yandex Cloud cloud network, enable public access to the cluster when creating it.

  2. Create a topic in the Managed Service for Apache Kafka® cluster.

  3. Create users so that the producer and consumer can access the topic in the Managed Service for Apache Kafka® cluster:

    • With the ACCESS_ROLE_PRODUCER role for the producer.
    • With the ACCESS_ROLE_CONSUMER role for the consumer.

  4. Create a Managed Service for ClickHouse target cluster with any suitable configuration. To connect to the cluster from the user's local machine instead of the Yandex Cloud cloud network, enable public access to the cluster when creating it.

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

  1. If you don't have Terraform, install and configure it.

  2. Download the file with provider settings. Place it in a separate working directory and specify the parameter values.

  3. Download the configuration file data-transfer-mkf-mch.tf to the same working directory.

    This file describes:

    • Network.
    • Subnet.
    • security group and rules required to connect to the clusters from the internet.
    • A source Managed Service for Apache Kafka® cluster with a description of a topic and two users on whose behalf the producer and consumer will connect to the topic.
    • Managed Service for ClickHouse target cluster.
    • Target endpoint.
    • Transfer.

  4. In the data-transfer-mkf-mch.tf file, specify:

    • Source Managed Service for Apache Kafka® cluster parameters:

      • source_user_producer and source_password_producer: Username and password for the producer user.
      • source_user_consumer and source_password_consumer: Username and password of the consumer user.
      • source_topic_name: Name of the topic.

    • Parameters of the target Managed Service for ClickHouse cluster that are also used as the target endpoint parameters:

      • target_db_name: Database nameManaged Service for ClickHouse.
      • target_user and target_password: Username and password of the database owner.

  5. Run the command terraform init in the directory with the configuration file. This command initializes the provider specified in the configuration files and enables you to use the provider resources and data sources.

  6. Make sure the Terraform configuration files are correct using the command:

    terraform validate

    If there are errors in the configuration files, Terraform will point to them.

  7. Create the required infrastructure:

    1. Run the command to view planned changes:

      terraform plan

      If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.

    2. If you are happy with the planned changes, apply them:

      1. Run the command:

        terraform apply

      2. Confirm the update of resources.

      3. Wait for the operation to complete.

    All the required resources will be created in the specified folder. You can check resource availability and their settings in the management console.

Configure additional settings

  1. Install the utilities:

    • kafkacat: To read and write data to the Apache Kafka® topic.

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

      Check that you can use it to connect to Managed Service for Apache Kafka® clusters over SSL.

    • clickhouse-client: to connect to the database in the Managed Service for ClickHouse cluster.

      1. Connect the DEB repository ClickHouse:

        sudo apt update && sudo apt install --yes apt-transport-https ca-certificates dirmngr && \
sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv E0C56BD4 && \
echo "deb https://repo.clickhouse.com/deb/stable/ main/" | sudo tee \
/etc/apt/sources.list.d/clickhouse.list

      2. Install the dependencies:

        sudo apt update && sudo apt install --yes clickhouse-client

      3. Download the configuration file for clickhouse-client:

        mkdir --parents ~/.clickhouse-client && \
wget "https://storage.yandexcloud.net/doc-files/clickhouse-client.conf.example" \
  --output-document ~/.clickhouse-client/config.xml

      Check that you can use it to connect to the Managed Service for ClickHouse cluster over SSL.

    • jq for JSON file stream processing.

      sudo apt update && sudo apt-get install --yes jq

Send test data to the Managed Service for Apache Kafka® topic

Suppose that your Apache Kafka® topic receives data from car sensors. This data will be transmitted as Apache Kafka® messages in JSON format:

{
    "device_id":"iv9a94th6rztooxh5ur2",
    "datetime":"2020-06-05 17:27:00",
    "latitude":"55.70329032",
    "longitude":"37.65472196",
    "altitude":"427.5",
    "speed":"0",
    "battery_voltage":"23.5",
    "cabin_temperature":"17",
    "fuel_level":null
}

The Managed Service for ClickHouse cluster will use JSONEachRow format to insert data to Kafka tables. This format converts strings from Apache Kafka® messages to relevant column values.

  1. Create a sample.json file with the following test data:

    sample.json
    {
    "device_id": "iv9a94th6rztooxh5ur2",
    "datetime": "2020-06-05 17:27:00",
    "latitude": 55.70329032,
    "longitude": 37.65472196,
    "altitude": 427.5,
    "speed": 0,
    "battery_voltage": 23.5,
    "cabin_temperature": 17,
    "fuel_level": null
}

{
    "device_id": "rhibbh3y08qmz3sdbrbu",
    "datetime": "2020-06-06 09:49:54",
    "latitude": 55.71294467,
    "longitude": 37.66542005,
    "altitude": 429.13,
    "speed": 55.5,
    "battery_voltage": null,
    "cabin_temperature": 18,
    "fuel_level": 32
}

{
    "device_id": "iv9a94th6rztooxh5ur2",
    "datetime": "2020-06-07 15:00:10",
    "latitude": 55.70985913,
    "longitude": 37.62141918,
    "altitude": 417.0,
    "speed": 15.7,
    "battery_voltage": 10.3,
    "cabin_temperature": 17,
    "fuel_level": null
}

  2. Send data from the sample.json file to the Managed Service for Apache Kafka® topic using jq and kafkacat:

    jq -rc . sample.json | kafkacat -P \
   -b <FQDN broker host>:9091 \
   -t <topic name> \
   -k key \
   -X security.protocol=SASL_SSL \
   -X sasl.mechanisms=SCRAM-SHA-512 \
   -X sasl.username="<producer username>" \
   -X sasl.password="<user password for the producer>" \
   -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexCA.crt -Z

Prepare and activate the transfer

  1. Create a source endpoint:

    • DB type: Apache Kafka.

    • Endpoint parameters → Connection settings:

      • Connection settings: Managed Service for Apache Kafka® cluster.

        • Managed Service for Apache Kafka® cluster: Select the source cluster from the list.

        • Authentication:

          • Username: Enter the consumer username.
          • Password: Enter the consumer user password.

      • Full topic name: Enter the name of the topic in the Managed Service for Apache Kafka® cluster.

      • (optional) Advanced settings → Conversions rules:

        • Data format: JSON.

        • Data schema: JSON specification:

          Create and upload the json_schema.json file in JSON format:

          json_schema.json
          [
    {
        "name": "device_id",
        "type": "string"
    },
    {
        "name": "datetime",
        "type": "datetime"
    },
    {
        "name": "latitude",
        "type": "double"
    },
    {
        "name": "longitude",
        "type": "double"
    },
    {
        "name": "altitude",
        "type": "double"
    },
    {
        "name": "speed",
        "type": "double"
    },
    {
        "name": "battery_voltage",
        "type": "any"
    },
    {
        "name": "cabin_temperature",
        "type": "double"
    },
    {
        "name": "fuel_level",
        "type": "any"
    }
]

  2. Create an endpoint for the target and the transfer:

    1. Create a target endpoint:

      • Database type: ClickHouse.

      • Endpoint parameters:

        • Connection settings:

          • Connection settings: Managed Service for ClickHouse cluster.

            • Managed Service for ClickHouse cluster: Select the source cluster from the list.

          • Database: Enter the name of the database.

          • User and Password: Enter the name and password of the user who has access to the database, for example, the database owner.

        • Advanced settings → Upload data in JSON format: Enable this option if you enabled Conversion rules in the advanced settings of the source endpoint.

    2. Create a transfer of the Increment type that will use the created endpoints.

    3. Activate it.

    1. In the data-transfer-mkf-mch.tf file, uncomment:

      • The source_endpoint_id parameter and assign to it the value of the endpoint ID for the source created in the previous step.
      • The yandex_datatransfer_endpoint and yandex_datatransfer_transfer resources.

    2. Make sure the Terraform configuration files are correct using the command:

      terraform validate

      If there are errors in the configuration files, Terraform will point to them.

    3. Create the required infrastructure:

      1. Run the command to view planned changes:

        terraform plan

        If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.

      2. If you are happy with the planned changes, apply them:

        1. Run the command:

          terraform apply

        2. Confirm the update of resources.

        3. Wait for the operation to complete.

      Once created, a transfer is activated automatically.

Test the transfer

  1. Wait for the transfer status to change to Replicating.

  2. Make sure that the data from the source Managed Service for Apache Kafka® cluster has been moved to the Managed Service for ClickHouse database:

    1. Connect to the cluster using clickhouse-client:

    2. Run the query:

      SELECT * FROM <ClickHouse database name>.<Apache Kafka® topic name>

  3. Change values in the sample.json file and send data from it to the Managed Service for Apache Kafka® topic:

    jq -rc . sample.json | kafkacat -P \
   -b <FQDN broker host>:9091 \
   -t <topic name> \
   -k key \
   -X security.protocol=SASL_SSL \
   -X sasl.mechanisms=SCRAM-SHA-512 \
   -X sasl.username="<producer username>" \
   -X sasl.password="<user password for the producer>" \
   -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexCA.crt -Z

  4. Make sure that the new values are now in the Managed Service for ClickHouse database:

    1. Connect to the cluster using clickhouse-client:

    2. Run the query:

      SELECT * FROM <ClickHouse database name>.<Apache Kafka® topic name>

Delete the resources you created

Note

Before deleting the created resources, disable the transfer.

If you no longer need these resources, delete them:

  1. Delete the transfer.
  2. Delete the source endpoint.

Delete the other resources, depending on the method used to create them:

  1. In the terminal window, change to the directory containing the infrastructure plan.

  2. Delete the configuration file data-transfer-mkf-mch.tf.

  3. Make sure the Terraform configuration files are correct using the command:

    terraform validate

    If there are errors in the configuration files, Terraform will point to them.

  4. Confirm the update of resources.

    1. Run the command to view planned changes:

      terraform plan

      If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.

    2. If you are happy with the planned changes, apply them:

      1. Run the command:

        terraform apply

      2. Confirm the update of resources.

      3. Wait for the operation to complete.

    All the resources described in the configuration file data-transfer-mkf-mch.tf will be deleted.

In this article: