Streamlining NDJSON Processing with Kinesis Firehose Dynamic Partitioning

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).

Takahiro Iwasa

2022-07-26

3 min read

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

The dynamic partitioning configuration lives in the CloudFormation template below.

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

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==

The encoded records can then be sent to the Firehose stream via the AWS CLI:

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

Remove the resources provisioned by this example with:

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

Conclusion

Enabling dynamic partitioning on a Firehose delivery stream produced partitioned, newline-delimited JSON in S3 without a transformation Lambda in the pipeline. DynamicPartitioningConfiguration: Enabled: true combined with the MetadataExtraction processor is what lets the Prefix template reference !{partitionKeyFromQuery:user_id} directly, pulling the partition key out of each record’s JSON body via the JQ query {user_id: .user_id} rather than requiring a Lambda function to compute it beforehand. Pairing that with the AppendDelimiterToRecord processor’s \\n delimiter covers the other half of what a transformation Lambda used to be needed for — producing clean NDJSON output. Since a MetadataExtractionQuery that doesn’t match the actual record shape routes silently to the error prefix instead of raising an error, it’s worth checking the error/ path in S3 alongside success/ the first time real traffic flows through.