Ingest MQTT Data into CockroachDB Cloud

CockroachDB is a distributed SQL database designed for high availability, scalability, and strong consistency across multiple nodes and regions. It simplifies the complexity of deploying and maintaining a database infrastructure while providing the benefits of a traditional SQL database. CockroachDB is particularly well-suited for applications that require strong consistency and ACID transactions across distributed environments. By integrating CockroachDB with the EMQX Platform, you can efficiently ingest MQTT messages and event data into CockroachDB for reliable storage and advanced analytics.

How It Works

The integration of CockroachDB into the EMQX Platform combines the real-time data collection capabilities of MQTT with the robustness of CockroachDB's distributed SQL architecture. The built-in rule engine on the EMQX Platform helps simplify the data ingestion process, ensuring that MQTT messages are routed, processed, and stored efficiently in CockroachDB without the need for complex coding.

The diagram below illustrates a typical architecture for data integration between the EMQX Platform and CockroachDB.

Ingesting MQTT data into CockroachDB works as follows:

1. Message Publishing and Reception: Industrial IoT devices connect to the EMQX Platform via the MQTT protocol, transmitting real-time data based on their operational states, sensor readings, or triggered events. The EMQX Platform receives these MQTT messages, which include valuable data for further processing and storage.

2. Message Data Processing: Upon receiving the messages, the EMQX Platform routes them through a SQL-based rule engine. The rule engine is responsible for matching incoming messages against predefined rules to determine which messages should be sent to CockroachDB. If necessary, the rule engine can also perform data transformations such as filtering, aggregation, or enrichment, transforming the data into a format suitable for storage in CockroachDB.

3. Data Ingestion into CockroachDB: Once the rule engine identifies that certain messages need to be stored in CockroachDB, it triggers the action to insert the processed data into the database. The strong consistency and ACID properties of CockroachDB ensure that all transactions are reliable and the data remains consistent, even in a distributed environment.

4. Data Storage and Utilization: With data securely stored in CockroachDB, businesses can leverage its powerful SQL querying capabilities for various applications. For instance, IoT data collected from temperature sensors, GPS trackers, or industrial equipment can be stored in CockroachDB, enabling real-time analysis for predictive maintenance, operational monitoring, and decision-making.

Features and Benefits

The integration of CockroachDB with the EMQX Platform offers several features and benefits to optimize data transmission, storage, and utilization:

• Real-time Data Streaming: The EMQX Platform is optimized to handle real-time data streams, ensuring fast and reliable transmission of data from IoT devices to CockroachDB. This capability is ideal for applications that require real-time insights, such as monitoring and controlling IoT devices in manufacturing or energy sectors.
• High Availability and Scalability: CockroachDB's distributed architecture ensures that data is replicated across multiple nodes, providing high availability and fault tolerance. As data volumes increase, CockroachDB scales horizontally, maintaining consistent performance even with large datasets.
• **Flexibility in Data Transformation:**The SQL-based rule engine on the EMQX Platform provides a wide range of data transformation features. Organizations can preprocess data before it is ingested into CockroachDB, including filtering out unnecessary data, aggregating sensor readings, or enriching the payload with additional context.
• Strong Consistency and ACID Transactions: CockroachDB provides strong consistency guarantees, ensuring that all transactions are ACID-compliant, even in distributed environments. This makes it suitable for applications that require transactional integrity, such as financial systems or systems that manage critical infrastructure.
• Easy Deployment and Management: The EMQX Platform offers an intuitive interface to configure data sources, set up preprocessing rules, and manage CockroachDB storage settings. This simplifies the process of setting up and maintaining the data integration pipeline, reducing the complexity of system management.
• Advanced Analytics: With CockroachDB’s powerful SQL query capabilities, users can perform advanced data analytics on IoT data. The ability to run complex queries on large datasets enables businesses to derive valuable insights, such as anomaly detection, trend analysis, and predictive modeling.
• Integration with Existing Systems: CockroachDB supports a rich set of SQL features, making it compatible with existing data tools and systems. Organizations can easily integrate their IoT data with other enterprise applications, including business intelligence tools, dashboards, and machine learning platforms, to derive deeper insights.

Before You Start

This section describes the preparations you need to complete before you start to create the CockroachDB Cloud data integration in the EMQX Platform Console.

Prerequisites

• Knowledge about data integration
• Knowledge about data integration rules

Start a Cockroach Cloud Service

1. Sign up for a Cockroach Labs account at Cockroach Labs and create a new project and cluster.

2. Follow the Cockroach Labs Quick Start Guide to create your cluster.

   Important Notice

   VPC Peering and AWS PrivateLink are only supported on Standard Deployments and higher.

3. After you have created your Cockroach Cloud cluster, click SQL Shell from the left navigation menu of the Cockroach Cloud Console. This will open an interface where you can execute SQL queries.

4. Run the SQL Query to create a table:

   • Use the following SQL query to create a new table. For example:

     
     CREATE TABLE sensor_data (
         id SERIAL PRIMARY KEY,
         sensor_id STRING NOT NULL,
         temperature FLOAT,
         humidity FLOAT,
         timestamp TIMESTAMPTZ DEFAULT now(),
         created_at TIMESTAMPTZ DEFAULT now()
     );

5. Once the query is executed, you can navigate back to the Tables tab under the Databases section and refresh the page to see your new table listed.

Configure PrivateLink

Before connecting to CockroachDB Cloud, you need to create a PrivateLink in the EMQX Platform and configure a PrivateLink endpoint service in CockroachDB Cloud to establish a private network connection between your EMQX deployment and the CockroachDB Cloud server.

1. Log in to the EMQX Platform console and go to the overview page of your desired deployment.

2. Navigate to Network Management, and under the PrivateLink section, click the "+ Private Connection" button.

3. Confirm the prompt and click the Next Step.

4. Log in to the CockroachDB Cloud platform and open your cluster. Click Networking from the left navigation menu.

5. Find the Private endpoint, and click Add a private endpoint. Then, find the Service name and save it for later use.

   

6. Enter the Service name into the EMQX Platform as the endpoint service name. Click Create Private Connection, and wait for the Endpoint ID to show.

   

7. Go back to the CockroachDB Cloud platform, add the Endpoint ID from the EMQX Platform to Private Endpoints, provide a description (optional), and click Create Endpoint.

8. Wait for the connection status to change to Running on the deployment overview page.

9. Wait for the "success" message to appear. Then, check the Private endpoints section in CockroachDB Cloud to confirm that the newly configured Endpoint ID is displayed. This indicates that the PrivateLink between the EMQX Platform and CockroachDB Cloud has been successfully established.

Create a Connector

Before creating data integration rules, you need to create a CockroachDB connector to access the CockroachDB Cloud server.

1. Go to your deployment. Click Data Integration from the left navigation menu. If it is your first time creating a connector, select PostgreSQL under the Data Persistence category. If you have already created connectors, select New Connector and then select PostgreSQL under the Data Persistence category.

2. On the New Connector page, configure the following connection information:

   • Connector Name: The system generates a connector name automatically.

   • Server Host: Enter CockroachDB Cluster server host.

     In the CockroachDB Cloud Console, click Connect in the top right corner. If you have established a private link, choose Private connection as the connection type, and copy and paste the highlighted part in the picture.

     

   • Database Name: Enter the database name, for example, defaultdb (the default database when you start the Cockroach Cloud cluster).

   • Username and Password: Enter the name and password you created when you created the CockroachDB cloud deployment.

   • Enable TLS: Click the toggle switch to enable the TLS connection. In the SNI textbox, enter the server host.

3. Click the Test button. If the Cockroach Cloud service is accessible, a prompt indicating the connector is available will be returned.

4. Click the New button to complete the creation.

Create a Rule

This section demonstrates how to create a CockroachDB Rule and add action to the rule via the EMQX Platform Console.

1. Click New Rule in the Rules area or click the New Rule icon in the Actions column of the connector you just created.

2. Set the rules in the SQL Editor based on the feature to use. To trigger the engine when the client sends a temperature and humidity message to the sensor/# topic, you can use the following SQL:

   SELECT 
       payload.sensor_id AS sensor_id,
       payload.temperature AS temperature,
       payload.humidity AS humidity,
       payload.timestamp AS timestamp
   FROM "sensor/#"

3. Click Next to add an action.

4. Select the connector you just created from the Connector dropdown box.

5. Enter the following command in the SQL template (You can use the Rule Engine to ensure that strings in the input SQL statement are properly escaped to prevent SQL injection attacks):

   INSERT INTO public.sensor_data(
       id, 
       sensor_id, 
       temperature, 
       humidity, 
       "timestamp", 
       created_at
   )
   VALUES (
       DEFAULT,
       ${sensor_id},
       ${temperature},
       ${humidity},
       TO_TIMESTAMP(${timestamp} / 1000),
       DEFAULT
   );

6. Advanced settings (optional).

7. Click the Confirm button to complete the rule creation.

8. In the Successful new rule pop-up, click Back to Rules, thus completing the entire data integration configuration chain.

Test the Rule

You are recommended to use MQTTX to simulate temperature and humidity data reporting, but you can also use any other client.

1. Use MQTTX to connect to the deployment and send messages to the following topic.

   • Topic: sensor/#

   • Client ID: test_client

   • Payload:

     {
       "sensor_id": "sensor_001",
       "temperature": 25.5,
       "humidity": 60.0,
       "timestamp": 1674392048000
     }

2. Click the send button to send the message. An entry should have been inserted in the sensor_data table in the defaultdb database on the Cockroach Cloud server.