Fetching data from RabbitMQ to Yandex Managed Service for ClickHouse®

You can supply data from RabbitMQ to a Managed Service for ClickHouse® cluster in real time. Managed Service for ClickHouse® will automatically insert the data routed into particular exchange points of the specified RabbitMQ queues into a RabbitMQ table.

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

1. Set up integration with RabbitMQ for the Managed Service for ClickHouse® cluster.
2. In the Managed Service for ClickHouse® cluster, create a RabbitMQ table.
3. Send the test data to the RabbitMQ queue.
4. Check that the test data is present in the Managed Service for ClickHouse® cluster table.

If you no longer need the resources you created, delete them.

Required paid resourcesRequired paid resources

The support cost includes:

• Managed Service for ClickHouse® cluster fee: Using computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see {{ mch-name } pricing}).
• Fee for using public IP addresses if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).
• VM fee: using computing resources, storage, and, optionally, public IP address (see Compute Cloud pricing).

Getting startedGetting started

Prepare the infrastructurePrepare the infrastructure

Configure additional settingsConfigure additional settings

1. Connect to a virtual machine over SSH.

   1. Install RabbitMQ:

      sudo apt update && sudo apt install rabbitmq-server --yes
      

   2. Create a user for RabbitMQ:

      sudo rabbitmqctl add_user <username> <password>
      

   3. Grant this user permissions to connect to the server:

      sudo rabbitmqctl set_permissions -p / <username> ".*" ".*" ".*" && \
      sudo rabbitmqctl set_topic_permissions -p / <username> amq.topic "cars" "cars"
      

2. Install the amqp-publish and amqp-declare-queue utilities to work with RabbitMQ and jq for stream processing JSON files:

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

3. Check if you can create a queue named cars in RabbitMQ using amqp-declare-queue:

   amqp-declare-queue \
       --url=amqp://<username>:<password>@<IP_address_or_FQDN_of_the_RabbitMQ_server>:5672 \
       --queue=cars
   

4. Install the clickhouse-client utility 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://packages.clickhouse.com/deb stable main" | sudo tee \
      /etc/apt/sources.list.d/clickhouse.list
      

   2. Install the dependencies:

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

   3. Download the configuration file for clickhouse-client:

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

   Check that you can use clickhouse-client to connect to the Managed Service for ClickHouse® cluster over SSL.

Set up integration with RabbitMQ for the Managed Service for ClickHouse® clusterSet up integration with RabbitMQ for the Managed Service for ClickHouse® cluster

resource "yandex_mdb_clickhouse_cluster" "clickhouse-cluster" {
  ...
  clickhouse {
    ...
    config {
      rabbitmq {
        username = "<username>"
        password = "<password>"
      }
    }
    ...
  }
}

In the Managed Service for ClickHouse® cluster, create a RabbitMQ tableIn the Managed Service for ClickHouse® cluster, create a RabbitMQ table

Let's assume you put the following JSON car sensor data into the RabbitMQ queue named cars at the exchange point named exchange:

• device_id: String ID of the device.

• datetime: Date and time when the data was generated, in YYYY-MM-DD HH:MM:SS format.

• Car coordinates:

  • latitude: Latitude.
  • longitude: Longitude.
  • altitude: Altitude above sea level.

• speed: Current speed.

• battery_voltage: Battery voltage (for electric cars; for cars with internal combustion engine, this parameter is null).

• cabin_temperature: Temperature inside the car.

• fuel_level: Fuel level (for cars with internal combustion engine; for electric cars, this parameter is null).

This data will be transmitted as RabbitMQ messages. Each message will contain a JSON object as a string in the following format:

{"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}

The Managed Service for ClickHouse® cluster will use the JSONEachRow format for data inserted into the table, which converts strings from RabbitMQ messages to the appropriate column values.

Create a table in the Managed Service for ClickHouse® cluster to accept data from RabbitMQ:

1. Connect to the db1 database in the Managed Service for ClickHouse® cluster using clickhouse-client.

2. Run this request:

   CREATE TABLE IF NOT EXISTS db1.cars (
       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 = RabbitMQ
   SETTINGS
       rabbitmq_host_port = '<internal_IP_address_of_VM_with_RabbitMQ>:5672',
       rabbitmq_routing_key_list = 'cars',
       rabbitmq_exchange_name = 'exchange',
       rabbitmq_format = 'JSONEachRow';
   

This table will be automatically filled with messages read from the cars queue at RabbitMQ's exchange point named exchange. When reading the data, Managed Service for ClickHouse® uses the authentication credentials provided earlier.

Send the test data to the RabbitMQ queueSend the test data to the RabbitMQ queue

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

   {
       "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
   }
   

2. Send the data from the sample.json file to the previously created cars queue of the exchange point named exchange using jq and amqp-publish.

   jq \
   --raw-output \
   --compact-output . ./sample.json |\
   amqp-publish \
   --url=amqp://<RabbitMQ_username>:<password>@<IP_address_or_FQDN_of_the_RabbitMQ_server>:5672 \
   --routing-key=cars \
   --exchange=exchange
   

Check that the test data is present in the Managed Service for ClickHouse® cluster tableCheck that the test data is present in the Managed Service for ClickHouse® cluster table

To access the data, use a materialized view (MATERIALIZED VIEW). When a materialized view is added to a RabbitMQ table, it starts collecting data in the background. This allows you to continuously receive messages from RabbitMQ and convert them to the required format using SELECT.

To create a materialized view for the db1.cars table:

1. Connect to the db1 database in the Managed Service for ClickHouse® cluster using clickhouse-client.

2. Run the following requests:

   CREATE TABLE IF NOT EXISTS db1.cars_data_source (
       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;
   
   CREATE MATERIALIZED VIEW db1.cars_view TO db1.cars_data_source
     AS SELECT * FROM db1.cars;
   

To get all the data from the db1.cars_view materialized view:

1. Connect to the db1 database in the Managed Service for ClickHouse® cluster using clickhouse-client.

2. Run this request:

   SELECT * FROM db1.cars_view;
   

The query will return a table with data sent to RabbitMQ.

Delete the resources you createdDelete the resources you created

Delete the resources you no longer need to avoid paying for them:

ClickHouse® is a registered trademark of ClickHouse, Inc.