AWS IoT Core における Kinesis Firehose のレコードセパレーター設定

IoT Core のトピックルールにある Firehose アクションでレコードセパレーターを設定し、S3 に改行区切りでレコードを格納する方法。

Takahiro Iwasa

2020-05-13

5 min read

IoT Core のトピックルールでは、個々の Firehose レコードの間にレコードセパレーターを挿入できます。これは、後続のコンシューマーが一続きのブロブではなく改行区切りの出力を期待している場合に重要になります。

構築

ポイントとなるのは、13〜14 行目にある Separator: |+ <NEW_LINE> の設定です。

1
AWSTemplateFormatVersion: "2010-09-09"
2

3
Resources:
4
  TopicRule:
5
    Type: AWS::IoT::TopicRule
6
    Properties:
7
      RuleName: topic_rule_firehose_separator_test
8
      TopicRulePayload:
9
        Actions:
10
          - Firehose:
11
              DeliveryStreamName: !Ref Firehose
12
              RoleArn: !GetAtt IamTopicRule.Arn
13
              Separator: |+
14

15
        AwsIotSqlVersion: 2016-03-23
16
        RuleDisabled: false
17
        Sql: !Sub
18
          SELECT * FROM 'topic_rule_firehose_separator_test'
19

20
  Firehose:
21
    Type: AWS::KinesisFirehose::DeliveryStream
22
    Properties:
23
      DeliveryStreamName: topic-rule-firehose-separator-test
24
      DeliveryStreamType: DirectPut
25
      S3DestinationConfiguration:
26
        BucketARN: !GetAtt S3.Arn
27
        BufferingHints:
28
          IntervalInSeconds: 60
29
          SizeInMBs: 5
30
        CompressionFormat: GZIP
31
        ErrorOutputPrefix: "error/!{firehose:error-output-type}/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/"
32
        Prefix: "success/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/"
33
        RoleARN: !GetAtt IamFirehose.Arn
34

35
  S3:
36
    Type: AWS::S3::Bucket
37
    Properties:
38
      BucketEncryption:
39
        ServerSideEncryptionConfiguration:
40
          - ServerSideEncryptionByDefault:
41
              SSEAlgorithm: AES256
42
      BucketName: topic-rule-firehose-separator-test
43
      PublicAccessBlockConfiguration:
44
        BlockPublicAcls: TRUE
45
        BlockPublicPolicy: TRUE
46
        IgnorePublicAcls: TRUE
47
        RestrictPublicBuckets: TRUE
48

49
  IamTopicRule:
50
    Type: AWS::IAM::Role
51
    Properties:
52
      AssumeRolePolicyDocument:
53
        Version: '2012-10-17'
54
        Statement:
55
          - Effect: Allow
56
            Principal:
57
              Service: iot.amazonaws.com
58
            Action: sts:AssumeRole
59
      Policies:
60
        - PolicyDocument:
61
            Version: '2012-10-17'
62
            Statement:
63
              - Effect: Allow
64
                Action: firehose:PutRecord
65
                Resource:
66
                  - !GetAtt Firehose.Arn
67
          PolicyName: policy
68
      RoleName: iam-topic-rule
69

70
  IamFirehose:
71
    Type: AWS::IAM::Role
72
    Properties:
73
      AssumeRolePolicyDocument:
74
        Version: '2012-10-17'
75
        Statement:
76
          - Effect: Allow
77
            Principal:
78
              Service: firehose.amazonaws.com
79
            Action: sts:AssumeRole
80
            Condition:
81
              StringEquals:
82
                sts:ExternalId: !Ref AWS::AccountId
83
      Policies:
84
        - PolicyDocument:
85
            Version: '2012-10-17'
86
            Statement:
87
              - Effect: Allow
88
                Action:
89
                  - glue:GetTable
90
                  - glue:GetTableVersion
91
                  - glue:GetTableVersions
92
                Resource: "*"
93
              - Effect: Allow
94
                Action:
95
                  - s3:AbortMultipartUpload
96
                  - s3:GetBucketLocation
97
                  - s3:GetObject
98
                  - s3:ListBucket
99
                  - s3:ListBucketMultipartUploads
100
                  - s3:PutObject
101
                Resource:
102
                  - !GetAtt S3.Arn
103
                  - Fn::Sub:
104
                      - ${arn}/*
105
                      - {arn: !GetAtt S3.Arn}
106
              - Effect: Allow
107
                Action:
108
                  - lambda:InvokeFunction
109
                  - lambda:GetFunctionConfiguration
110
                Resource: !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:%FIREHOSE_DEFAULT_FUNCTION%:%FIREHOSE_DEFAULT_VERSION%
111
              - Effect: Allow
112
                Action:
113
                  - logs:PutLogEvents
114
                Resource:
115
                  - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/topic-rule-firehose-separator-test
116
              - Effect: Allow
117
                Action:
118
                  - kinesis:DescribeStream
119
                  - kinesis:GetShardIterator
120
                  - kinesis:GetRecords
121
                Resource: !Sub arn:aws:kinesis:${AWS::Region}:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
122
              - Effect: Allow
123
                Action:
124
                  - kms:Decrypt
125
                Resource:
126
                  - !Sub arn:aws:kms:${AWS::Region}:${AWS::AccountId}:key/%SSE_KEY_ID%
127
                Condition:
128
                  StringEquals:
129
                    kms:ViaService: kinesis.%REGION_NAME%.amazonaws.com
130
                  StringLike:
131
                    kms:EncryptionContext:aws:kinesis:arn: !Sub arn:aws:kinesis:%REGION_NAME%:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
132
          PolicyName: policy
133
      RoleName: iam-firehose

CloudFormation スタックをデプロイします。

aws cloudformation deploy \
  --template template.yaml \
  --stack-name topic-rule-firehose-separator-test \
  --capabilities CAPABILITY_NAMED_IAM

トピックルールの設定を確認します。

aws iot get-topic-rule \
  --rule-name topic_rule_firehose_separator_test

出力の 12 行目に separator の設定が表示されるはずです。

1
{
2
  "ruleArn": "arn:aws:iot:<YOUR_REGION>:<YOUR_ACCOUNT_ID>:rule/topic_rule_firehose_separator_test",
3
  "rule": {
4
    "ruleName": "topic_rule_firehose_separator_test",
5
    "sql": "SELECT * FROM 'topic_rule_firehose_separator_test'",
6
    "createdAt": "2020-05-13T10:29:18+09:00",
7
    "actions": [
8
      {
9
        "firehose": {
10
          "roleArn": "arn:aws:iam::<YOUR_ACCOUNT_ID>:role/iam-topic-rule",
11
          "deliveryStreamName": "topic-rule-firehose-separator-test",
12
          "separator": "\n"
13
        }
14
      }
15
    ],
16
    "ruleDisabled": false,
17
    "awsIotSqlVersion": "2016-03-23"
18
  }
19
}

テスト

トピック topic_rule_firehose_separator_test にテストメッセージを送信します。

aws iot-data publish \
  --topic topic_rule_firehose_separator_test \
  --payload '{"id": 1, "message": "Hello from AWS IoT"}' \
  --cli-binary-format raw-in-base64-out

aws iot-data publish \
  --topic topic_rule_firehose_separator_test \
  --payload '{"id": 2, "message": "Hello from AWS IoT"}' \
  --cli-binary-format raw-in-base64-out

生成されたオブジェクトを S3 から取得して確認できます。

# Check an object.
$ aws s3 ls topic-rule-firehose-separator-test --recursive
2020-05-14 11:30:59         68 success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz

# Download the object.
$ aws s3 cp s3://topic-rule-firehose-separator-test/success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz ./result.gz

出力には、設定したセパレーターで区切られた 2 件のレコードが含まれているはずです。

# Check the JSON.
$ gunzip -c result.gz > result.json
$ cat result.json
{"id": 1, "message": "Hello from AWS IoT"}
{"id": 2, "message": "Hello from AWS IoT"}

# Delete the results.
$ rm result.gz result.json
$ aws s3 rm s3://topic-rule-firehose-separator-test/success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz

後片付け

作業が終わったらスタックを削除します。

aws cloudformation delete-stack --stack-name topic-rule-firehose-separator-test

まとめ

IoT トピックルールの Firehose アクションに Separator を追加してデプロイし、ダウンロードした S3 オブジェクトを確認したところ、2 件のテストレコードが一続きのブロブではなく改行区切りの JSON として格納されていました。Separator: |+ は YAML のブロックスカラーで、リテラルな改行に解決されます。このたった一つのプロパティが、きれいに区切られたレコードになるか、後段で独自の分割ロジックが必要な出力になるかの分かれ目です。実際のトラフィックがパイプラインを流れる前に、aws iot get-topic-rule で設定を確認し、さらにダウンロードして gunzip したオブジェクトで実際のセパレーターを確認しておく価値があります。セパレーターが設定されていなくてもエラーにはならず、単に全レコードが黙って連結されてしまうためです。

記事一覧に戻る