Ingest MQTT Data into Azure Blob Storage

Azure Blob Storage is Microsoft's cloud-based object storage solution, designed specifically for handling large volumes of unstructured data. Unstructured data refers to data types that do not follow a specific data model or format, such as text files or binary data. EMQX Cloud can efficiently store MQTT messages in Blob Storage containers, providing a versatile solution for storing Internet of Things (IoT) data.

This page provides a detailed introduction to the data integration between EMQX Cloud and Azure Blob Storage and offers practical guidance on the rule and Sink creation.

How It Works

Azure Blob Storage data integration in the EMQX Cloud is a ready-to-use feature that can be easily configured for complex business development. In a typical IoT application, the EMQX Cloud acts as the IoT platform responsible for device connectivity and message transmission, while Azure Blob Storage serves as the data storage platform, handling message data storage.

EMQX Cloud utilizes rules engines and Sinks to forward device events and data to Azure Blob Storage. Applications can read data from Azure Blob Storage for further data applications. The specific workflow is as follows:

1. Device Connection to EMQX Cloud: IoT devices trigger an online event upon successfully connecting via the MQTT protocol. The event includes device ID, source IP address, and other property information.
2. Device Message Publishing and Receiving: Devices publish telemetry and status data through specific topics. EMQX Cloud receives the messages and compares them within the rules engine.
3. Rules Engine Processing Messages: The built-in rules engine processes messages and events from specific sources based on topic matching. It matches corresponding rules and processes messages and events, such as data format transformation, filtering specific information, or enriching messages with context information.
4. Writing to Azure Blob Storage: The rule triggers an action to write the message to the Storage Container. Using the Azure Blob Storage Sink, users can extract data from processing results and send it to Blob Storage. Messages can be stored in text or binary format, or multiple lines of structured data can be aggregated into a single CSV file, depending on the message content and the Sink configuration.

After events and message data are written to the Storage Container, you can connect to Azure Blob Storage to read the data for flexible application development, such as:

• Data archiving: Store device messages as objects in Azure Blob Storage for long-term preservation to meet compliance requirements or business needs.
• Data analysis: Import data from Storage Container into analytics services like Snowflake for predictive maintenance, device efficiency evaluation, and other data analysis services.

Features and Advantages

Using Azure Blob Storage data integration in the EMQX Cloud can bring the following features and advantages to your business:

• Message Transformation: Messages can undergo extensive processing and transformation in EMQX Cloud rules before being written to Azure Blob Storage, facilitating subsequent storage and use.
• Flexible Data Operations: With the Azure Blob Storage Sink, specific fields of data can be conveniently written into Azure Blob Storage containers, supporting the dynamic setting of containers and object keys for flexible data storage.
• Integrated Business Processes: The Azure Blob Storage Sink allows device data to be combined with the rich ecosystem applications of Azure Blob Storage, enabling more business scenarios like data analysis and archiving.
• Low-Cost Long-Term Storage: Compared to databases, Azure Blob Storage offers a highly available, reliable, and cost-effective object storage service, suitable for long-term storage needs.

These features enable you to build efficient, reliable, and scalable IoT applications and benefit from business decisions and optimizations.

Before You Start

This section introduces the preparations required before creating an Azure Blob Storage Sink in EMQX.

Prerequisites

• Understand Data Integration.
• Familiarize yourself with Rules.

Prepare Azure Storage

1. To access Azure Storage, you'll need an Azure subscription. If you don't already have a subscription, create a free account before you begin.

2. All access to Azure Storage takes place through a storage account. For this quickstart, create a storage account using the Azure portal, Azure PowerShell, or Azure CLI. For help creating a storage account, see Create a storage account.

3. To create a container in the Azure portal, navigate to your new storage account in the Azure portal. In the left menu for the storage account, scroll to the Data storage section, then select Containers. Select the + Container button, use iot-data as a name for your new container, and click Create to create the container.

   

4. Navigate to Security+Networking -> Access keys in the storage account, and copy the Key. You will need this key to configure the Sink in EMQX.

   

Configure Network Access

Choose the network access method that matches your deployment environment.

Public Access

• Same cloud platform and region: Public access is available without enabling a NAT Gateway.
• Different regions or cloud platforms: Enable the NAT Gateway for public access.

Private Access via VNet Peering

Before creating the connector, complete the following network configurations:

1. Ensure that your EMQX Cloud deployment uses the Dedicated or Dedicated Flex edition, runs on Azure, and is in the same region as the VNet where you will create the Azure Blob Storage private endpoint.

2. Create a VPC Peering connection between the EMQX Cloud deployment and the Azure VNet.

3. In the Azure portal, open the Storage Account and go to Networking -> Private endpoint connections. Click + Private endpoint.

4. In the Create a private endpoint wizard, configure the Basics page. Select the Subscription and Resource group, complete the other required endpoint details, and click Next: Resource.

5. On the Resource page, configure the following settings:

   • Keep Connection method set to Connect to an Azure resource in my directory.
   • Select the Subscription that contains the Storage Account.
   • Set Resource type to Microsoft.Storage/storageAccounts.
   • For Resource, select the Storage Account.
   • Set Target sub-resource to blob.

   

   Click Next: Virtual Network.

6. On the Virtual Network page, select the Virtual network and Subnet connected to the EMQX Cloud deployment through VPC Peering.

   

   Click Next: DNS.

7. On the DNS page, under Private DNS integration, keep Integrate with private DNS zone set to Yes and use the default privatelink.blob.core.windows.net private DNS zone.

   

8. Click Next: Tags. Add tags if needed, go to Review + create, and then click Create after the configuration passes validation.

9. After the private endpoint deployment completes, open the private endpoint resource. Under Settings, select DNS configuration.

10. Record the following information:

    • FQDN (for example: <storage-account-name>.blob.core.windows.net)
    • Private IP address

    

11. Submit the FQDN and IP address to EMQX Cloud through a support ticket. The EMQX Cloud SRE team will add the corresponding private DNS record to the VNet where your EMQX Cloud deployment is located.

12. Ensure that the standard Storage Account domain name resolves to the Private Endpoint IP address.

13. Ensure that the routes and network security policies on both sides of the VNet Peering connection allow HTTPS traffic on port 443.

When configuring the connector, enter the Storage Account Name and Account Key as usual. You do not need to enter the Private Endpoint IP address. Use the standard Storage Account domain name and rely on private DNS to resolve it to the Private Endpoint IP address.

Troubleshoot Private Access

If the connector cannot access Azure Blob Storage through VNet Peering, check for the following common issues:

• The Storage Account domain name cannot be resolved.
• HTTPS traffic on port 443 cannot reach the Private Endpoint.
• The Storage Account network policy rejects the connection.

Create a Connector

Before adding the Azure Blob Storage Sink, you need to create the corresponding connector.

1. In the deployment menu, select Data Integration, then choose Azure Blob Storage service under the data persistence services category. If you have already created other connectors, click New Connector, then select Azure Blob Storage service under the data persistence services category.

2. Connector Name: The system will automatically generate a name for the connector.

3. Enter the connection information:

   • Account Name: Your Storage Account name
   • Account Key: Your Storage Account key from the previous step
   • Advanced Settings (Optional): Refer to Advanced Configuration。

4. Click the Test Connection button; if the Azure Blob Storage can be accessed normally, a success message will be returned.

   If the connection test fails, check for the following common issues:

   • Azure Blob Storage returns an AuthorizationFailure error.
   • The Storage Account Name or Account Key is incorrect.

5. Click the Create button to complete the creation of the connector.

Create a Rule

Next, you need to create a rule that specifies the data to be written and add response actions to forward the processed data to Azure Blob Storage.

1. Click the new rule icon under the Actions column in the connector list or click New Rule in the Rules List to enter the Create New Rule step page.

2. Input the following rule SQL in the SQL editor:

   SELECT
     *
   FROM
       "t/#"

   TIP

   If you're new to SQL, you can click SQL Examples and Try It Out to learn and test the results of the rule SQL.

3. Click Next to start creating actions.

4. From the Use Connector dropdown, select the connector you previously created.

5. Set the Container by entering iot-data.

6. Select the Upload Method. The differences between the two methods are as follows:

   • Direct Upload: Each time the rule is triggered, data is uploaded directly to Azure Storage according to the preset object key and content. This method is suitable for storing binary or large text data. However, it may generate a large number of files.
   • Aggregated Upload: This method packages the results of multiple rule triggers into a single file (such as a CSV file) and uploads it to Azure Storage, making it suitable for storing structured data. It can reduce the number of files and improve write efficiency.

   The configuration parameters differ for each method. Please configure according to the selected method:

   Direct Upload requires configuring the following fields:

   • Blob Name: Defines the object's location to be uploaded to the container. It supports placeholders in the format of ${var} and can use / to specify storage directories. It's also necessary to set the object's suffix for management and differentiation. Here, enter msgs/${clientid}_${timestamp}.json, where ${clientid} is the client ID and ${timestamp} is the timestamp of the message. This ensures that each device's messages are written to different objects.
   • Object Content: By default, this is in JSON text format containing all fields. It supports placeholders in the format of ${var}. Here, enter ${payload} to use the message body as the object content. The storage format of the object depends on the format of the message body, supporting compressed files, images, or other binary formats.

   Aggregate Upload requires configuring the following parameters:

   • Blob Name: Used to specify the storage path of the object. The following variables can be used:

     • ${action}: Action name (required).
     • ${node}: Name of the EMQX node performing the upload (required).
     • ${datetime.{format}}: Start date and time of the aggregation, with the format specified by the {format} string (required):
       • ${datetime.rfc3339utc}: RFC3339 date and time in UTC format.
       • ${datetime.rfc3339}: RFC3339 date and time in local time zone format.
       • ${datetime.unix}: Unix timestamp.
     • ${datetime_until.{format}}: End date and time of the aggregation, with format options as above.
     • ${sequence}: Sequence number for aggregated uploads within the same time interval (required).

     Note that if all placeholders marked as required are not used in the template, these placeholders will be automatically added to the Blob Name as path suffixes to avoid duplication. All other placeholders are considered invalid.

   • Aggregation Type: Currently, only CSV is supported. Data will be written to Azure Storage in comma-separated CSV format.

   • Column Order: Adjust the order of rule result columns through a dropdown selection. The generated CSV file will first be sorted by the selected columns, with unselected columns sorted alphabetically following the selected columns.

   • Max Records: When the maximum number of records is reached, the aggregation of a single file will be completed and uploaded, resetting the time interval.

   • Time Interval: When the time interval is reached, even if the maximum number of records has not been reached, the aggregation of a single file will be completed and uploaded, resetting the maximum number of records.

7. Configure advanced settings options as needed (optional), details can be found in Advanced Settings。

8. Click the Confirm button to complete the action configuration.

9. In the popup success message box, click Return to Rule List to complete the data integration configuration.

Test the Rule

This section shows how to test the rule configured with the direct upload method.

1. Use MQTTX to publish a message to the topic t/1:

mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello Azure" }'

2. After sending a few messages, log in to the Azure portal, navigate to the storage account, and open the iot-data container. You should see the uploaded objects in the container.

3. Check the runtime data in the EMQX Cloud Console. Click on the rule ID in the rules list, and you can view the statistics of the rule and all actions under this rule on the runtime statistics page.