Kinesis Firehose の動的パーティショニングで NDJSON 処理を簡素化する

Kinesis Data Firehose の動的パーティショニングを使い、変換用の Lambda 関数を用意せずに NDJSON（改行区切り JSON）を S3 へ出力します。

Takahiro Iwasa

2022-07-26

4 min read

Kinesis Data Firehose は 2021 年に動的パーティショニングをサポートしました。パーティションキーの抽出とレコード区切り文字の追加ができるため、NDJSON（改行区切り JSON）への変換だけを担う Lambda 関数が不要になります。

構築

動的パーティショニングの設定は、以下の CloudFormation テンプレートに記述します。

1
AWSTemplateFormatVersion: 2010-09-09
2
Description: Kinesis Data Firehose streaming NDJSON sample with dynamic partitioning
3
Resources:
4
  KinesisFirehoseDeliveryStream:
5
    Type: AWS::KinesisFirehose::DeliveryStream
6
    Properties:
7
      DeliveryStreamName: ndjson-firehose
8
      DeliveryStreamType: DirectPut
9
      ExtendedS3DestinationConfiguration:
10
        BucketARN: !GetAtt S3Bucket.Arn
11
        BufferingHints:
12
          IntervalInSeconds: 60
13
        DynamicPartitioningConfiguration:
14
          Enabled: true
15
        Prefix: "success/user_id=!{partitionKeyFromQuery:user_id}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
16
        ErrorOutputPrefix: "error/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/type=!{firehose:error-output-type}/"
17
        ProcessingConfiguration:
18
          Enabled: true
19
          Processors:
20
            - Type: AppendDelimiterToRecord
21
              Parameters:
22
                - ParameterName: Delimiter
23
                  ParameterValue: '\\n'
24
            - Type: MetadataExtraction
25
              Parameters:
26
                - ParameterName: MetadataExtractionQuery
27
                  ParameterValue: '{user_id: .user_id}'
28
                - ParameterName: JsonParsingEngine
29
                  ParameterValue: JQ-1.6
30
        RoleARN: !GetAtt IAMRoleKinesisFirehose.Arn
31

32
  S3Bucket:
33
    Type: AWS::S3::Bucket
34
    Properties:
35
      BucketName: !Sub ${AWS::Region}-${AWS::AccountId}-ndjson-s3
36
      BucketEncryption:
37
        ServerSideEncryptionConfiguration:
38
          - ServerSideEncryptionByDefault:
39
              SSEAlgorithm: AES256
40
      PublicAccessBlockConfiguration:
41
        BlockPublicAcls: true
42
        BlockPublicPolicy: true
43
        IgnorePublicAcls: true
44
        RestrictPublicBuckets: true
45

46
  IAMRoleKinesisFirehose:
47
    Type: AWS::IAM::Role
48
    Properties:
49
      RoleName: ndjson-firehose-role
50
      AssumeRolePolicyDocument:
51
        Version: 2012-10-17
52
        Statement:
53
          - Effect: Allow
54
            Principal:
55
              Service: firehose.amazonaws.com
56
            Action: sts:AssumeRole
57
      MaxSessionDuration: 3600
58
      Policies:
59
        - PolicyName: policy1
60
          PolicyDocument:
61
            Version: 2012-10-17
62
            Statement:
63
              - Effect: Allow
64
                Action:
65
                  - glue:GetTable
66
                  - glue:GetTableVersion
67
                  - glue:GetTableVersions
68
                Resource:
69
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog
70
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%
71
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%
72
              - Effect: Allow
73
                Action:
74
                  - s3:AbortMultipartUpload
75
                  - s3:GetBucketLocation
76
                  - s3:GetObject
77
                  - s3:ListBucket
78
                  - s3:ListBucketMultipartUploads
79
                  - s3:PutObject
80
                Resource:
81
                  - !GetAtt S3Bucket.Arn
82
                  - !Sub ${S3Bucket.Arn}/*
83
              - Effect: Allow
84
                Action:
85
                  - lambda:InvokeFunction
86
                  - lambda:GetFunctionConfiguration
87
                Resource: !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%
88
              - Effect: Allow
89
                Action:
90
                  - kms:GenerateDataKey
91
                  - kms:Decrypt
92
                Resource:
93
                  - !Sub arn:aws:kms:${AWS::Region}:${AWS::AccountId}:key/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%
94
                Condition:
95
                  StringEquals:
96
                    kms:ViaService: !Sub s3.${AWS::Region}.amazonaws.com
97
                  StringLike:
98
                    kms:EncryptionContext:aws:s3:arn:
99
                      - arn:aws:s3:::%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/*
100
                      - arn:aws:s3:::%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%
101
              - Effect: Allow
102
                Action:
103
                  - logs:PutLogEvents
104
                Resource:
105
                  - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%:log-stream:*

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

aws cloudformation deploy \
  --template-file stack.yaml \
  --stack-name firehose-ndjson-sample \
  --s3-bucket <YOUR_S3_BUCKET> \
  --s3-prefix firehose-ndjson-sample/$(date +%Y/%m/%d/%H) \
  --capabilities CAPABILITY_NAMED_IAM

テスト

公式ドキュメントに記載されているとおり、データブロブは送信前に Base64 エンコードする必要があります。

データブロブは、シリアライズ時にBase64エンコードされます。Base64エンコード前のデータブロブの最大サイズは1,000 KiBです。

サンプルの JSON を以下のようにエンコードします。

$ echo -n '{"user_id": 1, "message": "Hello"}' | base64
eyJ1c2VyX2lkIjogMSwgIm1lc3NhZ2UiOiAiSGVsbG8ifQ==

$ echo -n '{"user_id": 1, "message": "World"}' | base64
eyJ1c2VyX2lkIjogMSwgIm1lc3NhZ2UiOiAiV29ybGQifQ==

エンコードしたレコードを AWS CLI 経由で Firehose ストリームに送信します。

echo '{
    "DeliveryStreamName": "ndjson-firehose",
    "Records": [
        {"Data": "eyJ1c2VyX2lkIjogMSwgIm1lc3NhZ2UiOiAiSGVsbG8ifQ=="},
        {"Data": "eyJ1c2VyX2lkIjogMSwgIm1lc3NhZ2UiOiAiV29ybGQifQ=="}
    ]
}' > input.json
aws firehose put-record-batch --cli-input-json file://~/input.json

S3 バケット内のオブジェクトを確認します。

$ aws s3 ls --recursive s3://<AWS_REGION>-<AWS_ACCOUNT>-ndjson-s3/success/
2022-07-26 23:52:47         69 success/user_id=1/year=2022/month=07/day=26/hour=14/ndjson-firehose-4-2022-07-26-14-50-22-b95618c0-e518-3b66-b06c-693b059cc751

$ aws s3 cp s3://<AWS_REGION>-<AWS_ACCOUNT>-ndjson-s3/success/user_id=1/year=2022/month=07/day=26/hour=14/ndjson-firehose-4-2022-07-26-14-50-22-b95618c0-e518-3b66-b06c-693b059cc751 ./
download: s3://<AWS_REGION>-<AWS_ACCOUNT>-ndjson-s3/success/user_id=1/year=2022/month=07/day=26/hour=14/ndjson-firehose-4-2022-07-26-14-50-22-b95618c0-e518-3b66-b06c-693b059cc751 to ./ndjson-firehose-4-2022-07-26-14-50-22-b95618c0-e518-3b66-b06c-693b059cc751

$ cat -n ndjson-firehose-4-2022-07-26-14-50-22-b95618c0-e518-3b66-b06c-693b059cc751
     1  {"user_id": 1, "message": "Hello"}
     2  {"user_id": 1, "message": "World"}

後片付け

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

aws s3 rm --recursive s3://<AWS_REGION>-<AWS_ACCOUNT>-ndjson-s3/
aws cloudformation delete-stack --stack-name firehose-ndjson-sample

まとめ

Firehose 配信ストリームで動的パーティショニングを有効にすると、変換用の Lambda 関数を使わず、パーティション分けされた改行区切りの JSON を S3 に出力できます。

DynamicPartitioningConfiguration: Enabled: true と MetadataExtraction プロセッサを組み合わせると、Prefix テンプレートから !{partitionKeyFromQuery:user_id} を直接参照できます。JQ クエリ {user_id: .user_id} が各レコードの JSON 本文からパーティションキーを抽出するため、Lambda 関数で事前に計算する必要はありません。さらに、AppendDelimiterToRecord プロセッサで \\n を追加すると、整形された NDJSON を出力できます。

MetadataExtractionQuery がレコード構造と一致しない場合、Firehose はレコードをエラー用プレフィックスへ出力します。実際のトラフィックを初めて流すときは、S3 の success/ と併せて error/ パスも確認してください。

記事一覧に戻る