Streamlining NDJSON Processing with Kinesis Firehose Dynamic Partitioning

Use Kinesis Data Firehose dynamic partitioning to write NDJSON (Newline Delimited JSON) to S3 without a transformation Lambda function.

Takahiro Iwasa

2022-07-26

3 min read

Kinesis Data Firehose introduced support for dynamic partitioning in 2021. It can extract partition keys and append record delimiters, eliminating the need for a Lambda function that only converts records to NDJSON (Newline Delimited JSON).

Configuration

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

As described in the official documentation, data blobs must be Base64-encoded before they are sent.

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.

If MetadataExtractionQuery does not match the record structure, Firehose writes the record under the error prefix. Check the error/ path in S3 alongside success/ when sending real traffic for the first time.

Back to posts