AWS IoT トピックペイロードによる Kinesis シャードの分離

IoT トピックのペイロードを顧客 ID ごとに異なる Kinesis シャードへルーティングし、後続処理でも顧客ごとの順序を保つ方法。

Takahiro Iwasa

2020-04-16

4 min read

Kinesis のシャードは IoT トピックのペイロードによって分離できます。これは、異なる顧客のレコードを後続処理のために別々のシャードへ振り分けたい場合に役立ちます。以下の例では、customer_id によってレコードをルーティングしています。

前提条件

以下があらかじめマシンにインストールされている必要があります。

AWS SAM CLI
Python 3.x

構築

AWS IoT のトピックルールでは PartitionKey: ${customer_id}（37 行目）を指定しており、これによってレコードがどのようにシャードへ振り分けられるかが決まります。

1
AWSTemplateFormatVersion: '2010-09-09'
2
Transform: AWS::Serverless-2016-10-31
3

4
Resources:
5
  Lambda:
6
    Type: AWS::Serverless::Function
7
    Properties:
8
      CodeUri: ./src/
9
      Events:
10
        Kinesis:
11
          Type: Kinesis
12
          Properties:
13
            BatchSize: 100
14
            BisectBatchOnFunctionError: true
15
            Enabled: true
16
            StartingPosition: LATEST
17
            Stream: !GetAtt Kinesis.Arn
18
      FunctionName: aws_iot_kinesis_partition_lambda
19
      Handler: lambda_function.lambda_handler
20
      Role: !GetAtt IamRole.Arn
21
      Runtime: python3.8
22

23
  Kinesis:
24
    Type: AWS::Kinesis::Stream
25
    Properties:
26
      Name: aws_iot_kinesis_partition_stream
27
      RetentionPeriodHours: 24
28
      ShardCount: 1
29

30
  TopicRule:
31
    Type: AWS::IoT::TopicRule
32
    Properties:
33
      RuleName: aws_iot_kinesis_partition_topic_rule
34
      TopicRulePayload:
35
        Actions:
36
          - Kinesis:
37
              PartitionKey: ${customer_id}
38
              RoleArn: !GetAtt IamRole.Arn
39
              StreamName: aws_iot_kinesis_partition_stream
40
        AwsIotSqlVersion: 2016-03-23
41
        RuleDisabled: false
42
        Sql: SELECT * FROM 'aws-iot-kinesis-partition-topic'
43

44
  IamRole:
45
    Type: AWS::IAM::Role
46
    Properties:
47
      AssumeRolePolicyDocument:
48
        Version: '2012-10-17'
49
        Statement:
50
          - Effect: Allow
51
            Principal:
52
              Service: lambda.amazonaws.com
53
            Action: sts:AssumeRole
54
          - Effect: Allow
55
            Principal:
56
              Service: iot.amazonaws.com
57
            Action: sts:AssumeRole
58
      ManagedPolicyArns:
59
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
60
      Policies:
61
        - PolicyDocument:
62
            Version: '2012-10-17'
63
            Statement:
64
              - Effect: Allow
65
                Action:
66
                  - kinesis:GetShardIterator
67
                  - kinesis:GetRecords
68
                  - kinesis:DescribeStream
69
                  - kinesis:PutRecord
70
                Resource:
71
                  - !GetAtt Kinesis.Arn
72
          PolicyName: policy
73

74
  LogGroup:
75
    Type: AWS::Logs::LogGroup
76
    Properties:
77
      LogGroupName: /aws/lambda/aws_iot_kinesis_partition_lambda
78
      RetentionInDays: 1

ストリームされたレコードを処理する Lambda 関数は、短い Python スクリプトです。

1
import json
2

3

4
def lambda_handler(event, context):
5
    for record in event['Records']:
6
        print(json.dumps(record['kinesis']))

SAM を使ってスタックをビルド・デプロイします。

sam build
sam deploy \
  --stack-name aws-iot-kinesis-partition-lambda \
  --capabilities CAPABILITY_NAMED_IAM

テスト

テスト用のペイロードは、AWS IoT のトピックに直接送信できます。

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

Lambda のログを確認し、partitionKey がペイロード内の customer_id と一致していることを確認します。

後片付け

この例で作成したリソースは、以下のコマンドで削除します。

sam delete --stack-name aws-iot-kinesis-partition-lambda

まとめ

IoT トピックルールで PartitionKey: ${customer_id} を設定したことで、Lambda コンシューマー側に何のパーティショニングロジックを持たせることなく、同一顧客のレコードを同じ Kinesis シャードへルーティングできました。ルーティングはこの一つの式だけで完結しており、レコードが event['Records'] に到達する時点ですでに同じシャードへ届いています。だからこそ後続処理で顧客ごとの順序を維持でき、顧客のイベントを順番に処理する必要がある処理にとっては重要なポイントになります。ただし注意すべき点として、これは customer_id の値がある程度均等に分布している場合にのみ負荷を均等に分散できるという前提があります。パーティションキーを正しく設定していても、特定の顧客の量が突出して多ければ、そのトラフィックは一つのシャードに集中してしまいます。

記事一覧に戻る