Fetching data from RabbitMQ to 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 RabbitMQ integration for your Managed Service for ClickHouse® cluster.
2. Create a RabbitMQ table in the Managed Service for ClickHouse® cluster.
3. Send the test data to the RabbitMQ queue.
4. Make sure the test data appears 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 for this solution includes:

• Managed Service for ClickHouse® cluster fee, which covers the use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Managed Service for ClickHouse® pricing).
• Fee for public IP addresses if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).
• VM fee, which covers the use of computing resources, storage, and, optionally, public IP address (see Compute Cloud pricing).

Getting startedGetting started

Set up your infrastructureSet up your 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 tools for RabbitMQ and jq for processing JSON streams:

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

3. Use amqp-declare-queue to create a queue named cars in RabbitMQ:

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

4. Install clickhouse-client to connect to the database in your Managed Service for ClickHouse® cluster.

   1. Add the ClickHouse® DEB repository:

      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 clickhouse-client configuration file:

      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 via clickhouse-client.

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

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

Create a RabbitMQ engine table in your Managed Service for ClickHouse® clusterCreate a RabbitMQ engine table in your Managed Service for ClickHouse® cluster

Suppose, you publish the following JSON car sensor data to the RabbitMQ exchange named exchange, which routes it to the cars queue:

• device_id: Device string identifier.

• datetime: Date and time of data generation in YYYY-MM-DD HH:MM:SS format.

• Car coordinates:

  • latitude: Latitude.
  • longitude: Longitude.
  • altitude: Height above mean sea level.

• speed: Current speed.

• battery_voltage: Battery voltage for electric cars. null for ICE vehicles.

• cabin_temperature: Temperature inside the car.

• fuel_level: Fuel level for ICE cars. null for electric cars.

This data will be transmitted as RabbitMQ messages. Each message will contain a string containing a serialized JSON object with the following structure:

{"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 table inserts, the Managed Service for ClickHouse® cluster will use the JSONEachRow format that parses rows from RabbitMQ messages into the required column values.

In the Managed Service for ClickHouse® cluster, create a table to store data incoming from RabbitMQ:

1. Connect to the db1 database on your Managed Service for ClickHouse® cluster via clickhouse-client.

2. Run this query:

   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 populated with messages consumed from the cars queue, which is bound to RabbitMQ 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 file named sample.json 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. Use jq and amqp-publish to send data from sample.json to the previously created cars queue via exchange.

   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
   

Make sure the test data appears in the Managed Service for ClickHouse® cluster tableMake sure the test data appears in the Managed Service for ClickHouse® cluster table

To access the data, use a materialized view. Once a materialized view is attached to a RabbitMQ table, it starts gathering data in the background automatically. This enables the system to continuously consume 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 on your Managed Service for ClickHouse® cluster via clickhouse-client.

2. Run the following queries:

   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 retrieve all data from the db1.cars_view materialized view:

1. Connect to the db1 database on your Managed Service for ClickHouse® cluster via clickhouse-client.

2. Run this query:

   SELECT * FROM db1.cars_view;
   

The query results will show all 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.