[Yandex Cloud documentation](../../index.md) > [Tutorials](../index.md) > [Building a data platform](index.md) > Fetching data from Managed Service for Apache Kafka® to Managed Service for ClickHouse®

# Fetching data from Yandex Managed Service for Apache Kafka® to Yandex Managed Service for ClickHouse®

A Managed Service for ClickHouse® cluster can ingest data from Apache Kafka® topics in real time. This data will be automatically inserted into ClickHouse® [`Kafka`](https://clickhouse.com/docs/enen/engines/table-engines/integrations/kafka)-engine tables.

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

1. [Set up Apache Kafka® integration for your Managed Service for ClickHouse® cluster](#configure-mch-for-kf).
1. [Create Kafka-engine tables in your Managed Service for ClickHouse® cluster](#create-kf-table).
1. [Send test data to your Managed Service for Apache Kafka® topics](#send-sample-data-to-kf).
1. [Check Managed Service for ClickHouse® cluster tables for test data](#fetch-sample-data).

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

{% note warning %}

This tutorial uses a single-host ClickHouse® cluster. If your cluster has more than one ClickHouse® host, use a [distributed query](https://clickhouse.com/docs/enen/sql-reference/statements/create/table) by specifying your cluster name in the SQL statements below: `CREATE ... ON CLUSTER '{cluster}'`. Additionally, use the `ReplicatedMergeTree` table engine to replace `MergeTree` wherever it appears in the statements.

{% endnote %}


## Required paid resources {#paid-resources}

The support cost for this solution includes:

* Managed Service for Apache Kafka® cluster fee: use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see [Apache Kafka® pricing](../../managed-kafka/pricing.md)).
* Managed Service for ClickHouse® cluster fee: use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see [Managed Service for ClickHouse® pricing](../../managed-clickhouse/pricing.md)).
* Fee for public IP addresses if public access is enabled for cluster hosts (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).


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

### Set up your infrastructure {#deploy-infrastructure}

{% list tabs group=instructions %}

- Manually {#manual}

    {% 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 the required number of Managed Service for Apache Kafka® clusters](../../managed-kafka/operations/cluster-create.md) of any suitable [configuration](../../managed-kafka/concepts/instance-types.md). To be able to connect to the clusters not only from within the Yandex Cloud network but also from a local machine, enable public access when creating them.

    1. [Create a Managed Service for ClickHouse® cluster](../../managed-clickhouse/operations/cluster-create.md) with a single shard and a database named `db1`. For connections to the cluster from the user's local machine, rather than the Yandex Cloud network, enable public access to the cluster when creating it.

        {% note info %}

        Integration with Apache Kafka® is available during [cluster creation](../../managed-clickhouse/operations/cluster-create.md). In this tutorial, however, we will configure the integration [at a later step](#configure-mch-for-kf).

        {% endnote %}

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

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


    1. [Create the required number of topics](../../managed-kafka/operations/cluster-topics.md#create-topic) in Managed Service for Apache Kafka® clusters. Make sure the topic names are unique.

    1. To enable the [producer and consumer](../../managed-kafka/concepts/producers-consumers.md) to access topics, [create](../../managed-kafka/operations/cluster-accounts.md#create-account) two users in each Managed Service for Apache Kafka® cluster:

        - With the `ACCESS_ROLE_PRODUCER` role for the producer.
        - With the `ACCESS_ROLE_CONSUMER` role for the consumer.

        Usernames may be the same across clusters.

- Terraform {#tf}

    1. If you do not have Terraform yet, [install it](../infrastructure-management/terraform-quickstart.md#install-terraform).
    1. [Get the authentication credentials](../infrastructure-management/terraform-quickstart.md#get-credentials). You can add them to environment variables or specify them later in the provider configuration file.
    1. [Configure and initialize a provider](../infrastructure-management/terraform-quickstart.md#configure-provider). There is no need to create a provider configuration file manually, you can [download it](https://github.com/yandex-cloud-examples/yc-terraform-provider-settings/blob/main/provider.tf).
    1. Place the configuration file in a separate working directory and [specify the parameter values](../infrastructure-management/terraform-quickstart.md#configure-provider). If you did not add the authentication credentials to environment variables, specify them in the configuration file.

    1. Download the [data-from-kafka-to-clickhouse.tf](https://github.com/yandex-cloud-examples/yc-clickhouse-fetch-data-from-kafka/blob/main/data-from-kafka-to-clickhouse.tf) configuration file to the same working directory.

        This file describes:

        * Network.
        * Subnet.
        * Default security group and inbound internet rules for the cluster.
        * Managed Service for Apache Kafka® cluster.
        * Topic and two Managed Service for Apache Kafka® users for producer and consumer access.

            To create multiple topics or clusters, duplicate the sections with their descriptions and provide a unique name for each. Usernames may be the same across clusters.

        * Managed Service for ClickHouse® cluster with a single shard and a database named `db1`.

    1. In the `data-from-kafka-to-clickhouse.tf` file, specify the following:

        * Managed Service for Apache Kafka® version.
        * Usernames and passwords of the accounts with the `ACCESS_ROLE_PRODUCER` and `ACCESS_ROLE_CONSUMER` roles in your Managed Service for Apache Kafka® clusters.
        * Names of topics in the Managed Service for Apache Kafka® clusters.
        * Username and password that will be used to access your Managed Service for ClickHouse® cluster.

    1. Validate your Terraform configuration files using this command:

       ```bash
       terraform validate
       ```

       Terraform will display any configuration errors detected in your files.

    1. Create the required infrastructure:

       1. Run this command to view the planned changes:
       
          ```bash
          terraform plan
          ```
       
          If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.
       
       1. If everything looks correct, apply the changes:
          1. Run this command:
       
             ```bash
             terraform apply
             ```
       
          1. Confirm updating the resources.
          1. 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](https://console.yandex.cloud).

{% endlist %}

### Configure additional settings {#additional-settings}

1. Install the following tools:

    - [kafkacat](https://github.com/edenhill/kcat): For reading from and writing to Apache Kafka® topics.

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

        Make sure you can use it to [establish SSL connections to your Managed Service for Apache Kafka® clusters](../../managed-kafka/operations/connect/clients.md#bash-zsh).

    - [clickhouse-client](https://clickhouse.com/docs/enen/interfaces/cli): For connecting to a database within the Managed Service for ClickHouse® cluster.

        1. Add the ClickHouse® [DEB repository](https://clickhouse.com/docs/enen/install#install-from-deb-packages):

            ```bash
            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://packages.clickhouse.com/deb stable main" | sudo tee \
            /etc/apt/sources.list.d/clickhouse.list
            ```

        1. Install the dependencies:

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

        1. Download the `clickhouse-client` configuration file:

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

        Verify that you can [establish an SSL connection to the Managed Service for ClickHouse® cluster](../../managed-clickhouse/operations/connect/clients.md) via `clickhouse-client`.

    - [jq](https://stedolan.github.io/jq/): For stream processing of JSON files.

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

## Set up Apache Kafka® integration for your Managed Service for ClickHouse® cluster {#configure-mch-for-kf}

Configuration depends on how many Managed Service for Apache Kafka® clusters you have:

* If you have a single Apache Kafka® cluster:

   {% list tabs group=instructions %}

   - Manually {#manual}

       Provide authentication data under **DBMS settings** → **Kafka** in [ClickHouse® settings](../../managed-clickhouse/operations/change-server-level-settings.md#yandex-cloud-interfaces). The Managed Service for ClickHouse® cluster will use these credentials when accessing any topic.

       Authentication data:

       * **Sasl mechanism**: `SCRAM-SHA-512`.
       * **Sasl password**: [Consumer user password](#before-you-begin).
       * **Sasl username**: [Consumer username](#before-you-begin).
       * **Security protocol**: `SASL_SSL`.

   - Terraform {#tf}

       1. In the `data-from-kafka-to-clickhouse.tf` file, uncomment the `clickhouse.config.kafka` section:

            ```hcl
            config = {
                kafka = {
                    security_protocol = "SECURITY_PROTOCOL_SASL_SSL"
                    sasl_mechanism    = "SASL_MECHANISM_SCRAM_SHA_512"
                    sasl_username     = "<username_for_consumer>"
                    sasl_password     = "<user_password_for_consumer>"
                }
            }
            ```

       1. Make sure the settings are correct.

           1. In the command line, navigate to the directory that contains the current Terraform configuration files defining the infrastructure.
           1. Run this command:
           
              ```bash
              terraform validate
              ```
           
              Terraform will show any errors found in your configuration files.

       1. Confirm updating the resources.

           1. Run this command to view the planned changes:
           
              ```bash
              terraform plan
              ```
           
              If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.
           
           1. If everything looks correct, apply the changes:
              1. Run this command:
           
                 ```bash
                 terraform apply
                 ```
           
              1. Confirm updating the resources.
              1. Wait for the operation to complete.

   {% endlist %}

* Multiple Apache Kafka® clusters: Create the required number of [named collections](https://clickhouse.com/docs/enen/operations/named-collections) containing authentication data for each Managed Service for Apache Kafka® topic:

   1. [Connect](../../managed-clickhouse/operations/connect/clients.md#clickhouse-client) to the `db1` database on your Managed Service for ClickHouse® cluster via `clickhouse-client`.

   1. Run the following query as many times as needed, providing the authentication data for each topic:

       ```sql
       CREATE NAMED COLLECTION <collection_name> AS
           kafka_broker_list = '<broker_host_FQDN>:9091',
           kafka_topic_list = '<topic_name>',
           kafka_group_name = 'sample_group',
           kafka_format = 'JSONEachRow'
           kafka_security_protocol = 'SASL_SSL',
           kafka_sasl_mechanism = 'SCRAM-SHA-512',
           kafka_sasl_username = '<username_for_consumer>',
           kafka_sasl_password = '<user_password_for_consumer>';
       ```

## Create Kafka-engine tables in your Managed Service for ClickHouse® cluster {#create-kf-table}

Let's assume your Apache Kafka® topics receive some car sensor data in JSON format. This data will be transmitted as Apache Kafka® messages, each containing a string like this:

```json
{"device_id":"iv9a94th6rzt********","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}
```

For `Kafka` table inserts, the Managed Service for ClickHouse® cluster will use the [JSONEachRow format](https://clickhouse.com/docs/enen/interfaces/formats#jsoneachrow), which converts rows from Apache Kafka® messages into the required column values.

For each Apache Kafka® topic, create a separate table for incoming data in your Managed Service for ClickHouse® cluster:

1. [Connect](../../managed-clickhouse/operations/connect/clients.md#clickhouse-client) to the `db1` database on your Managed Service for ClickHouse® cluster via `clickhouse-client`.
1. Run this query:

    {% list tabs group=instructions %}

    - Single Apache Kafka® cluster

        ```sql
        CREATE TABLE IF NOT EXISTS db1.<table_name_for_topic>
        (
            device_id String,
            datetime DateTime,
            latitude Float32,
            longitude Float32,
            altitude Float32,
            speed Float32,
            battery_voltage Nullable(Float32),
            cabin_temperature Float32,
            fuel_level Nullable(Float32)
        ) ENGINE = Kafka()
        SETTINGS
            kafka_broker_list = '<broker_host_FQDN>:9091',
            kafka_topic_list = '<topic_name>',
            kafka_group_name = 'sample_group',
            kafka_format = 'JSONEachRow';
        ```

    - Multiple Apache Kafka® clusters

        ```sql
        CREATE TABLE IF NOT EXISTS db1.<table_name_for_topic>
        (
            device_id String,
            datetime DateTime,
            latitude Float32,
            longitude Float32,
            altitude Float32,
            speed Float32,
            battery_voltage Nullable(Float32),
            cabin_temperature Float32,
            fuel_level Nullable(Float32)
        ) ENGINE = Kafka(<name_of_collection_with_authentication_data>);
        ```

    {% endlist %}

These tables will be automatically populated with messages consumed from Managed Service for Apache Kafka® topics. When reading data, Managed Service for ClickHouse® uses the [previously configured settings](#configure-mch-for-kf) for [users with the `ACCESS_ROLE_CONSUMER` role](#before-you-begin).

For more information about creating `Kafka` tables, see [this ClickHouse® guide](https://clickhouse.com/docs/enen/engines/table-engines/integrations/kafka).

## Send test data to your Managed Service for Apache Kafka® topics {#send-sample-data-to-kf}

1. Create a file named `sample.json` with test data:

    ```json
    {
        "device_id": "iv9a94th6rzt********",
        "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": "rhibbh3y08qm********",
        "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": "iv9a94th6rzt********",
        "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
    }
    ```

1. Send data from `sample.json` to each Managed Service for Apache Kafka® topic using `jq` and `kafkacat`:

    ```bash
    jq -rc . sample.json | kafkacat -P \
       -b <broker_host_FQDN>:9091 \
       -t <topic_name> \
       -k key \
       -X security.protocol=SASL_SSL \
       -X sasl.mechanisms=SCRAM-SHA-512 \
       -X sasl.username="<username_for_producer>" \
       -X sasl.password="<user_password_for_producer>" \
       -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/RootCA.crt -Z
    ```

Data is sent using the credentials of users with the [`ACCESS_ROLE_PRODUCER` role](#before-you-begin). To learn more about setting up an SSL certificate and using `kafkacat`, see [Connecting to a Apache Kafka® cluster from applications](../../managed-kafka/operations/connect/clients.md).

## Check Managed Service for ClickHouse® cluster tables for test data {#fetch-sample-data}

To access the data, use a materialized view. Once a materialized view is attached to a `Kafka` table, it starts collecting data in the background automatically. This enables the system to continuously consume messages from Apache Kafka® and convert them to the required format using `SELECT`.

{% note info %}

We do not recommend reading data directly from the table because ClickHouse® can read a message from a topic only once.

{% endnote %}

To create a materialized view:

1. [Connect](../../managed-clickhouse/operations/connect/clients.md#clickhouse-client) to the `db1` database on your Managed Service for ClickHouse® cluster via `clickhouse-client`.
1. Run the following queries for each `Kafka` table:

    ```sql
    CREATE TABLE db1.temp_<table_name_for_topic>
    (
        device_id String,
        datetime DateTime,
        latitude Float32,
        longitude Float32,
        altitude Float32,
        speed Float32,
        battery_voltage Nullable(Float32),
        cabin_temperature Float32,
        fuel_level Nullable(Float32)
    ) ENGINE = MergeTree()
    ORDER BY device_id;
    ```

    ```sql
    CREATE MATERIALIZED VIEW db1.<view_name> TO db1.temp_<table_name_for_topic>
        AS SELECT * FROM db1.<table_name_for_topic>;
    ```

To get all data from the materialized view:

1. [Connect](../../managed-clickhouse/operations/connect/clients.md#clickhouse-client) to the `db1` database on your Managed Service for ClickHouse® cluster via `clickhouse-client`.
1. Run this query:

    ```sql
    SELECT * FROM db1.<view_name>;
    ```

This query will return a table with data sent to the respective Managed Service for Apache Kafka® topic.

To learn more about working with data supplied from Apache Kafka®, see [this ClickHouse® guide](https://clickhouse.com/docs/enen/engines/table-engines/integrations/kafka).

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

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

{% list tabs group=instructions %}

- Manually {#manual}

    - Delete the clusters:

        - [Yandex Managed Service for ClickHouse®](../../managed-clickhouse/operations/cluster-delete.md).
        - [Yandex Managed Service for Apache Kafka®](../../managed-kafka/operations/cluster-delete.md).

    
    - If you reserved public static IP addresses for your clusters, release and [delete them](../../vpc/operations/address-delete.md).


- Terraform {#tf}

    1. In the terminal window, go to the directory containing the infrastructure plan.
    
        {% note warning %}
    
        Make sure the directory has no Terraform manifests with the resources you want to keep. Terraform deletes all resources that were created using the manifests in the current directory.
    
        {% endnote %}
    
    1. Delete resources:
    
        1. Run this command:
    
            ```bash
            terraform destroy
            ```
    
        1. Confirm deleting the resources and wait for the operation to complete.
    
        All the resources described in the Terraform manifests will be deleted.

{% endlist %}

_ClickHouse® is a registered trademark of [ClickHouse, Inc](https://clickhouse.com)._