Ingest Data into DynamoDB
EMQX supports integration with DynamoDB, so you can save MQTT messages and client events to DynamoDB, or use events to trigger the update or removal of data to record the online status or online/offline of clients.
TIP
EMQX Enterprise Edition features. EMQX Enterprise Edition provides comprehensive coverage of key business scenarios, rich data integration, product-level reliability, and 24/7 global technical support. Experience the benefits of this enterprise-ready MQTT messaging platform today.
Prerequisites
- Knowledge about EMQX data integration rules
- Knowledge about data bridge
Features List
Quick Start Tutorial
This section introduces how to configure the DynamoDB data bridge, covering topics like how to set up the DynamoDB server, create data bridges and rules for forwarding data to DynamoDB and test the data bridges and rules.
This tutorial assumes that you run both EMQX and DynamoDB on the local machine. If you have Dynamo and EMQX running remotely, adjust the settings accordingly.
Install DynamoDB Local Server
- Prepare a docker-compose file,
dynamo.yaml, to set up the Dynamodb local server.
version: '3.8'
services:
dynamo:
command: "-jar DynamoDBLocal.jar -sharedDb"
image: "amazon/dynamodb-local:latest"
container_name: dynamo
ports:
- "8000:8000"
environment:
AWS_ACCESS_KEY_ID: root
AWS_SECRET_ACCESS_KEY: public
AWS_DEFAULT_REGION: us-west-2- Start the server.
docker-compose -f dynamo.yaml up- Prepare a table definition and save it to your home directory as
mqtt_msg.json.
{
"TableName": "mqtt_msg",
"KeySchema": [
{ "AttributeName": "id", "KeyType": "HASH" }
],
"AttributeDefinitions": [
{ "AttributeName": "id", "AttributeType": "S" }
],
"ProvisionedThroughput": {
"ReadCapacityUnits": 5,
"WriteCapacityUnits": 5
}
}- Create a new table via this file.
docker run --rm -v ${HOME}:/dynamo_data -e AWS_ACCESS_KEY_ID=root -e AWS_SECRET_ACCESS_KEY=public -e AWS_DEFAULT_REGION=us-west-2 amazon/aws-cli dynamodb create-table --cli-input-json file:///dynamo_data/mqtt_msg.json --endpoint-url http://host.docker.internal:8000- Check if the table was created successfully.
docker run --rm -e AWS_ACCESS_KEY_ID=root -e AWS_SECRET_ACCESS_KEY=public -e AWS_DEFAULT_REGION=us-west-2 amazon/aws-cli dynamodb list-tables --endpoint-url http://host.docker.internal:8000The following JSON will be printed if the table was created successfully.
{
"TableNames": [
"mqtt_msg"
]
}Create a DynamoDB Data Bridge
Go to EMQX Dashboard, and click Integration -> Data Bridge.
Click Create on the top right corner of the page.
In the Create Data Bridge page, click to select DynamoDB, and then click Next.
Input a name for the data bridge. The name should be a combination of upper/lower case letters and numbers.
Input the connection information:
- Database Url: Input
http://127.0.0.1:8000, or the actual URL if the DynamoDB server is running remotely. - Table Name: Input
mqtt_msg. - AWS Access Key ID: Input
root. - AWS Secret Access Key: Input
public.
- Database Url: Input
Leave the Template empty by default.
TIP
When this value is empty the whole message will be stored in the database. The actual value is JSON template data.
Advanced settings (optional): Choose whether to use sync or async query mode as needed. For details, see Data Integration.
Before clicking Create, you can click Test Connectivity to test that the bridge can connect to the server.
Then click Create to finish the creation of the data bridge.
A confirmation dialog will appear and ask if you like to create a rule using this data bridge, you can click Create Rule to continue creating rules to specify the data to be saved into DynamoDB. You can also create rules by following the steps in Create Rules for DynamoDB Data Bridge.
Now the data bridge should appear in the data bridge list (Integration -> Data Bridge) with Resource Status as Connected.
Create a Rule for DynamoDB Data Bridge
Now that you have successfully created the data bridge to DynamoDB, you can continue to create rules to specify the data to be saved into DynamoDB. You need to create two different rules for messages forward and event records.
Go to EMQX Dashboard, and click Integration -> Rules.
Click Create on the top right corner of the page.
Input
my_ruleas the rule ID, and set the rules in the SQL Editor based on the feature to use:To create a rule for message storage, input the following statement, which means the MQTT messages under topic
t/#will be saved to DynamoDB.Note: If you want to specify your own SQL syntax, make sure that you have included all fields required by the data bridge in the
SELECTpart.sqlSELECT * FROM "t/#"To create a rule for online/offline status recording, input the following statement:
sqlSELECT str(event) + timestamp as id, * FROM "$events/client_connected", "$events/client_disconnected"TIP
For convenience, the
mqtt_msgtopic will be reused to receive online/offline events.
Click the Add Action button, select Forwarding with Data Bridge from the dropdown list, and then select the data bridge you just created under Data Bridge. Click the Add button.
Click the Create button to finish the setup.
Now you have successfully created the data bridge to DynamoDB. You can click Integration -> Flows to view the topology. It can be seen that the messages under topic t/# are sent and saved to DynamoDB after parsing by rule my_rule.
Test Data Bridge and Rule
Use MQTT X to send a message to topic t/1 to trigger an online/offline event.
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello DynamoDB" }'Check the running status of the data bridge, there should be one new incoming and one new outgoing message.
Check whether the data is written into the mqtt_msg data table.
docker run --rm -e AWS_ACCESS_KEY_ID=root -e AWS_SECRET_ACCESS_KEY=public -e AWS_DEFAULT_REGION=us-west-2 amazon/aws-cli dynamodb scan --table-name=mqtt_msg --endpoint-url http://host.docker.internal:8000The output will be:
{
"Items": [
{
"metadata": {
"S": "{\"rule_id\":\"90d98f59\"}"
},
"peerhost": {
"S": "127.0.0.1"
},
"clientid": {
"S": "emqx_c"
},
"flags": {
"S": "{\"retain\":false,\"dup\":false}"
},
"node": {
"S": "emqx@127.0.0.1"
},
"qos": {
"N": "0"
},
"payload": {
"S": "{ \"msg\": \"hello DynamoDB\" }"
},
"pub_props": {
"S": "{\"User-Property\":{}}"
},
"publish_received_at": {
"N": "1678263363503"
},
"topic": {
"S": "t/1"
},
"id": {
"S": "0005F65F239F03FEF44300000BB40002"
},
"event": {
"S": "message.publish"
},
"username": {
"S": "undefined"
},
"timestamp": {
"N": "1678263363503"
}
},
{
"conn_props": {
"S": "{\"User-Property\":{},\"Request-Problem-Information\":1}"
},
"peername": {
"S": "127.0.0.1:59582"
},
"metadata": {
"S": "{\"rule_id\":\"703890a5\"}"
},
"clientid": {
"S": "emqx_c"
},
"is_bridge": {
"S": "false"
},
"keepalive": {
"N": "30"
},
"proto_ver": {
"N": "5"
},
"proto_name": {
"S": "MQTT"
},
"connected_at": {
"N": "1678263363499"
},
"receive_maximum": {
"N": "32"
},
"sockname": {
"S": "127.0.0.1:1883"
},
"mountpoint": {
"S": "undefined"
},
"node": {
"S": "emqx@127.0.0.1"
},
"id": {
"S": "client.connected1678263363499"
},
"expiry_interval": {
"N": "0"
},
"event": {
"S": "client.connected"
},
"username": {
"S": "undefined"
},
"timestamp": {
"N": "1678263363499"
},
"clean_start": {
"S": "true"
}
},
{
"reason": {
"S": "normal"
},
"peername": {
"S": "127.0.0.1:59582"
},
"metadata": {
"S": "{\"rule_id\":\"703890a5\"}"
},
"clientid": {
"S": "emqx_c"
},
"proto_ver": {
"N": "5"
},
"proto_name": {
"S": "MQTT"
},
"sockname": {
"S": "127.0.0.1:1883"
},
"disconn_props": {
"S": "{\"User-Property\":{}}"
},
"node": {
"S": "emqx@127.0.0.1"
},
"id": {
"S": "client.disconnected1678263363503"
},
"event": {
"S": "client.disconnected"
},
"disconnected_at": {
"N": "1678263363503"
},
"username": {
"S": "undefined"
},
"timestamp": {
"N": "1678263363503"
}
}
],
"Count": 3,
"ScannedCount": 3,
"ConsumedCapacity": null
}