AWS IoT のペイロード値で Kinesis レコードをパーティショニングする

IoT トピックのペイロードに含まれる顧客 ID を Kinesis のパーティションキーとして使い、顧客ごとのレコード順序を維持する方法を解説します。

Takahiro Iwasa

2020-04-16

3 min read

AWS IoT のトピックルールでは、メッセージのペイロードに含まれる値を Kinesis レコードのパーティションキーとして使用できます。これは、後続処理で顧客ごとのレコード順序を維持したい場合に役立ちます。以下の例では、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 シャードへ送信できます。Kinesis はパーティションキーを基にシャードを選択するため、後続のコンシューマーは同じキーを持つレコードを順番に処理できます。

負荷が均等に分散されるのは、customer_id ごとのトラフィック量がおおむね均等な場合です。特定の顧客のトラフィック量が多いと、パーティションキーを正しく設定していても、1 つのシャードに負荷が集中する可能性があります。

記事一覧に戻る