Accelerate Your IoT Projects and Analytics: Deploying OpenSearch with Coreflux MQTT Broker

With the exponential growth of connected devices in today’s data-driven world, the ability to process billions of pieces of information at scale has become essential. MQTT (Message Queuing Telemetry Transport) is a lightweight messaging protocol designed for resource-constrained devices and low-bandwidth networks, making it ideal for IoT applications. An MQTT broker like Coreflux and a scalable cloud platform like DigitalOcean can solve the challenges of processing and analyzing IoT data.

This tutorial will teach you how to connect an MQTT broker with a managed OpenSearch service on DigitalOcean. This seamless setup enables real-time data collection and storage, making monitoring, analyzing, and visualizing your IoT data more accessible.

Coreflux provides a Lightweight MQTT Broker for Efficient IoT Communication on DigitalOcean.

Before diving into the integration, make sure you have the following:

1. A DigitalOcean account. If you don’t have one, sign up for an account at DigitalOcean.

2. Coreflux Broker Setup: Coreflux Broker should be running and accessible. If it’s not set up yet, refer to the Coreflux documentation or check the initial steps in this guide.

3. MQTT Explorer: This tool interacts with the MQTT broker. You can download it from MQTT Explorer.

4. Python Environment: Ensure you have Python installed and the necessary libraries like paho-mqtt and OpenSearch-py.

5. Python Script: You’ll need the Python script that bridges the Coreflux MQTT broker with your DigitalOcean OpenSearch instance. This script checks published MQTT messages, processes them, and stores them in OpenSearch.

If you are new to Python? Here’s a basic outline of what the Python script above does:

Connects to Coreflux: The script uses paho-mqtt to connect to your Coreflux MQTT broker.

• Connects to Coreflux: The script uses paho-mqtt to connect to your Coreflux MQTT broker.
• Subscribes to Topics: It listens for messages on specific topics you define in the Python script.
• Processes and Indexes Data: The script parses published JSON messages and attempts to index them into OpenSearch using OpenSearch-py. • Publishes Feedback: After processing, the script can publish feedback messages back to the MQTT broker alerting of errors or task completion.

You can watch the Coreflux Tutorial on how to start a Free Trial of the Online MQTT Broker quickly:

Or you can follow this step-by-step guide:

1. Create a Coreflux Account

   • Go to the Coreflux website and sign up for a free account if you don’t already have one
   • After signing up, verify your email address to activate your account

2. Start a Free Trial Broker

   • Once logged in, navigate to MQTT Broker
   • Click on Start Free Trial to create a new MQTT broker
   • Choose a Data Center Region: Select a region geographically close to your IoT devices or the DigitalOcean data center where you plan to deploy OpenSearch for lower latency
   • Confirm your choices to start the trial

3. Receive Broker Credentials

   • Coreflux will send the credentials to access your broker (such as the broker URL, port, username, and password) to your registered email after creating your broker
   • Keep these credentials safe, as you’ll need them to configure your IoT devices and the Python script later

4. Set Up MQTT Explorer

   • Download and install MQTT Explorer if you haven’t already
   • Open MQTT Explorer and configure it to connect to your Coreflux broker using the credentials you received:
     • Broker Address: Enter the broker URL
     • Port: Use the port provided (typically 8883 for SSL connections)
     • Username/Password: Enter the credentials provided in your email
   • Connect to the broker and try subscribing to a topic to ensure everything works

5. Test the Broker Connection

   • Publish a test message using MQTT Explorer to one of the topics on your Coreflux broker
   • Verify that the message is received and displayed correctly in the Explorer. This confirms that your broker is up and running

Now that your Coreflux MQTT broker is set up and tested, it’s time to connect it to a managed OpenSearch instance on DigitalOcean. Here’s how:

1. Log in to DigitalOcean:

   • Head over to the DigitalOcean dashboard and login with your credentials.

2. Create a New Database:

   • On the dashboard, click on Databases in the left-hand menu.
   • Select Create Database Cluster.
   • Choose OpenSearch from the list of available database types.

3. Configure Your OpenSearch Instance:

   • Select a Data Center Region:
     • Choose a region that’s geographically close to your IoT devices or Coreflux broker for lower latency.
   • Choose Your Plan:
     • You can start with a basic plan that is suitable for your current needs.
     • You can always scale up later as your data grows.

4. Create the Cluster:

   • Once configured, click Create Cluster
   • Wait for DigitalOcean to provision your OpenSearch instance. This may take a few minutes.

5. Get Your Connection Details:

   • Go to the Connection Details tab in your database cluster after creating the cluster
   • Note down the following details:
     • Host
     • Port
     • Username
     • Password
   • You’ll need these details to connect from your Python script.

Before you start indexing data from your Coreflux MQTT broker into OpenSearch, you need to define the mapping for your index. Mapping is the schema for your index, specifying the data types for each field in your documents. This step is crucial for ensuring the data is stored correctly and can be searched effectively.

Here’s how to create and map an index in your OpenSearch instance:

{
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date"
      },
      "machine_id": {
        "type": "keyword"
      },
      "temperature": {
        "type": "float"
      },
      "status": {
        "type": "keyword"
      },
      "error_code": {
        "type": "integer"
      }
    }
  }
}

After defining your mappings, review the settings and click on Create Index.

OpenSearch will now create the index with the mappings you specified. This index is now ready to store and organize the data that will be published from your Coreflux MQTT broker.

{
  "timestamp": "2024-08-23T10:30:00Z",
  "machine_id": "MACHINE123",
  "temperature": 75.5,
  "status": "operational",
  "error_code": 0
}

With Coreflux and OpenSearch set up, it’s time to link them together using a Python script. This script will connect to the Coreflux broker, process published messages, and store them in OpenSearch.

MQTT_BROKER=<your-coreflux-broker-url>
MQTT_PORT=1883
MQTT_USERNAME=<your-coreflux-username>
MQTT_PASSWORD=<your-coreflux-password>

OPENSEARCH_HOST=<your-opensearch-host>
OPENSEARCH_USERNAME=<your-opensearch-username>
OPENSEARCH_PASSWORD=<your-opensearch-password>

pip install paho-mqtt Opensearch-py python-dotenv

python mqttToOS.py

Integrating Coreflux MQTT Broker with DigitalOcean’s Managed OpenSearch service provides a powerful solution for real-time IoT data processing and analytics. Following this tutorial, you have set up a seamless data pipeline that allows you to collect, process, and visualize IoT data efficiently.

With Coreflux’s scalability and reliability and OpenSearch’s robust search and analytics capabilities, you can handle large volumes of data and gain valuable insights in real-time. Whether you are monitoring industrial systems, tracking environmental data, or managing smart home devices, this integration empowers you to make data-driven decisions quickly and effectively.

You can check here to learn how to start with OpenSearch on DigitalOcean.

Get a free Coreflux Online MQTT Broker trial or learn more with the Coreflux Docs and Tutorials.