Streamlining NDJSON Processing with Kinesis Firehose Dynamic Partitioning

Takahiro Iwasa

7/26/2022

3 min read

Firehose Kinesis

Kinesis Data Firehose recently introduced support for dynamic partitioning. With this, developers no longer require Lambda functions just to convert to NDJSON (Newline Delimited JSON).

Building

Create a CloudFormation template with the following content.

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:*

Deploy the CloudFormation stack with the following command:

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

Testing

Data blobs must be Base64 encoded before being sent described in the official documentation.

The data blob, which is base64-encoded when the blob is serialized. The maximum size of the data blob, before base64-encoding, is 1,000 KiB.

Encode your sample JSON as follows:

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

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

Use the AWS CLI to send data to the Firehose stream:

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

Check the objects in your S3 bucket:

$ 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"}

Cleaning Up

Clean up all the AWS resources provisioned during this example with the following command:

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