環境モニタを作ろう with ESP32 DEV Module 〜 No6. AWS-IoTからDynamoDBへ #aws-iot

ここまででできたこと

さて、今回は、AWS-IoTにPublishされたデータをDynamoDBに保存できるようにしましょう！

AWS IoTのルール設定
DynamoDBへ保存
ルール作成のSQLの演算子の話

AWS IoTのルール設定

IoTルールは、Publishされたトピックやメッセージの内容を基に、他のAWSサービスに連携できる機能です。メッセージを保存したり、Lambdaを起動したりするといったことができるようになります。

Amazon DynamoDB
Amazon S3
Amazon SNS
Amazon SQS
AWS Lambda
Amazon Kinesis Streams
Amazon Kinesis Firehose
AWS IoTの別Topicへのメッセージ転送

今回は、PublishされたデータをDynamoDBに保存するルールを作成しましょう

DynamoDBへ保存

DynamoDBのテーブルを作成する

センサーのデータを格納するテーブルを作成しましょう。

テーブル名：ueda_sensor_table

プライマリパーティションキー	id (文字列)
プライマリソートキー	timestamp (文字列)

「TableArn」が出力されるので、控えておきましょう。

aws dynamodb create-table  --table-name ueda_sensor_table \
--attribute-definitions AttributeName=id,AttributeType=S AttributeName=timestamp,AttributeType=S \
--key-schema AttributeName=id,KeyType=HASH AttributeName=timestamp,KeyType=RANGE \
--provisioned-throughput ReadCapacityUnits=1,WriteCapacityUnits=1

{
 "TableDescription": {
 "AttributeDefinitions": [
 {
 "AttributeName": "id",
 "AttributeType": "S"
 },
 {
 "AttributeName": "timestamp",
 "AttributeType": "S"
 }
 ],
 "TableName": "ueda_sensor_table",
 "KeySchema": [
 {
 "AttributeName": "id",
 "KeyType": "HASH"
 },
 {
 "AttributeName": "timestamp",
 "KeyType": "RANGE"
 }
 ],
 "TableStatus": "CREATING",
 "CreationDateTime": 1511494873.387,
 "ProvisionedThroughput": {
 "NumberOfDecreasesToday": 0,
 "ReadCapacityUnits": 1,
 "WriteCapacityUnits": 1
 },
 "TableSizeBytes": 0,
 "ItemCount": 0,
 "TableArn": "arn:aws:dynamodb:ap-northeast-1:xxxxxxx:table/ueda_sensor_table"
 }
}

AWS-IOTからDynamoDBにPutできるIAMロールを作成する

DynamoDBにデータをPutできるポリシーを作成します

「aws_iot_dynamoDB」という名前で、ロールを作成します。

vi assume.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal":{
         "Service":"iot.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

 aws  iam create-role --role-name aws_iot_dynamoDB --assume-role-policy-document file://assume.json

作成したロール「aws_iot_dynamoDB」に対して、ポリシーを適用します。

vi dynamo-policy.json

{
  "Version": "2012-10-17",
  "Statement": {
    "Effect": "Allow",
    "Action": "dynamodb:PutItem",
    "Resource":  "arn:aws:dynamodb:ap-northeast-1:xxxxxxxxxxx:table/ueda_sensor_table"
  }
}

aws  iam put-role-policy --role-name aws_iot_dynamoDB --policy-name dynamodb_put_policy --policy-document file://dynamo-policy.json

AWS-IOTからDynamoDBにPutするルールを作成

ルールの作成サンプルは、以下の公式サイトにあリマス。それを自分用にアレンジしてください。

AWS IoT ルールの作成 – AWS IoT

ルールを記述したjsonファイルを作成します

“sql”はどのトピックから取得するか
“dynamoDB”の部分は、対象となるテーブル名、AWS-IoTからDynamoにPUTするためのroleArnを記述し、それぞれのキーに何の値を格納するかを書きます

vi dynamo-rule.json

{
    "ruleName": "SensorRuleDynamo",
    "topicRulePayload": {
        "sql": "SELECT * FROM 'topic/sensor'",
        "description": "",
        "actions": [
            {
                "dynamoDB": {
                    "tableName": "ueda_sensor_table",
                    "roleArn" : "arn:aws:iam::749789197689:role/aws_iot_dynamoDB",
                    "hashKeyField": "id",
                    "hashKeyValue": "${topic(2)}",
                    "rangeKeyField": "timestamp",
                    "rangeKeyValue": "${timestamp()}"
                }
            }
        ],
        "ruleDisabled": false
    }
}

ルールを作成します

aws iot create-topic-rule --cli-input-json file://dynamo-rule.json

作成したルールを確認します

$ aws  iot list-topic-rules  --region ap-northeast-1
{
    "rules": [
        {
            "ruleArn": "arn:aws:iot:ap-northeast-1:749789197689:rule/SensorRuleDynamo",
            "ruleName": "SensorRuleDynamo",
            "topicPattern": "topic/sensor",
            "createdAt": 1511495449.0,
            "ruleDisabled": false
        }
    ]
}

ESP32からPublish

コード抜粋

元となるコードは、こちらのサイトにあります！

ESP32でAWS IoTに繋いでThing Shadowを弄る – コーヒーサーバは香炉である

トピックの名前だけ変更して実行。

char *pubTopic = "topic/sensor";

void mqttLoop() {
  if (!mqttClient.connected()) {
    connectAWSIoT();
  }
  mqttClient.loop();

  long now = millis();
  if (now - messageSentAt > 5000) {
    messageSentAt = now;
    sprintf(pubMessage, "{\"state\": {\"desired\":{\"foo\":\"%d\"}}}", dummyValue++);
    mqttClient.publish(pubTopic, pubMessage);
  }
}

void loop() {
  mqttLoop();
}

AWS-IoTからDynamoDBへ

「topic/sensor」でサブスクライブして、データが届いていることを確認します。

DynamoDBにデータが入っています！

ルール作成のSQLの演算子の話

ルールを作成する際のSQL 式の SELECT 句または WHERE 句では、組み込み関数を使うことができます。

関数 – AWS IoT

また、先ほどのRule作成の時に、「timestamp()」を使いましたが、「clientid()」で、クライアントIDを取得することもできます

clientid
- メッセージを送信している MQTT クライアントの ID を返すか、または、メッセージが MQTT クライアント経由で送られていない場合は、n/a を返します。SQL バージョン 2015-10-8 以降でサポートされています。

以下のように、ハッシュキーにclientidを指定すると、DynamoDBにはクライアントIDがキーで入るようになります

  "dynamoDB": {
 　　"tableName": "ueda_sensor_table",
 　　"roleArn" : "arn:aws:iam::749789197689:role/aws_iot_dynamoDB",
 　　"hashKeyField": "id",
 　　"hashKeyValue": "${clientid()}",
 　　"rangeKeyField": "timestamp",
 　　"rangeKeyValue": "${timestamp()}"
}

DynamoDBの状態