Q: AWS Kinesis Firehose + Lambda: How to handle multiple record creations/updates that occur at the same time I have a Lambda function linked to my Kinesis data stream as part of a Kinesis Firehose workflow. All the lambda function really does is take serialized records that are coming from DynamoDB, and deserialize them (as well as the usual S3 encoding/decoding stuff):
def lambda_handler(event, context):
    deserializer = TypeDeserializer()
    output = []
    try:
        for record in event["records"]:
            # Everything stored in S3 is base64 encoded, so we must first decode, deserialize, and
            # then encode the records again before we send them to S3
            decoded_payload = json.loads(base64.b64decode(record["data"]).decode())

            if decoded_payload["dynamodb"] and decoded_payload["dynamodb"]["NewImage"]:
                updated_record = decoded_payload["dynamodb"]["NewImage"]
                deserialized_record = {
                    k: deserializer.deserialize(v) for k, v in updated_record.items()
                }

                # Add newline after each record??? (otherwise Athena will only "see" the first?)
                encoded_record_with_line_break = (json.dumps(deserialized_record, cls=DecimalEncoder) + "\n").encode()
                output_record = {
                    "recordId": record["recordId"],
                    "result": "Ok",
                    "data": base64.b64encode(encoded_record_with_line_break).decode(),
                }
            else:
                print(
                    f"Dropping payload containing no updated DynamoDB image: {decoded_payload}"
                )
                output_record = {
                    "recordId": record["recordId"],
                    "result": "Dropped",
                    "data": record["data"],
                }

            output.append(output_record)

            print(f"Successfully processed {len(event['records'])} records.")
        except Exception as exc:
            print(f"Unhandled exception raised during data transformation: {exc}")
        finally:
            return {"records": output}

As you may have noticed, I have added '\n' to the record before I encode it and append it to the output array.
This is because I was running into an issue where my Glue Crawlers were not registering all the record updates that my lambda was outputting to S3, SPECIFICALLY WHEN MULTIPLE RECORDS WERE BEING CREATED/UPDATED AT THE SAME TIME.
Basically, the Kinesis Data Firehose sends JSON records inline; they are not delimited by a comma or a newline. After getting in touch with AWS support, they found an AWS blog post that says this:

By default, Kinesis Data Firehose sends JSON records inline, which causes Athena to query only the first record in each S3 object. To overcome this, we use a Lambda function to transform records before sending them to Amazon S3 by adding an end of line (EOL) character.

Link: https://aws.amazon.com/blogs/big-data/build-seamless-data-streaming-pipelines-with-amazon-kinesis-data-streams-and-amazon-kinesis-data-firehose-for-amazon-dynamodb-tables/
So we added the newline, as you see in my code above. Unfortunately, this did not seem to fix my issue- with the added newline, I noticed my lambda Cloudwatch logs seemed to be recording duplicates. Each update to a single record resulted in 2 files being created instead of just 1, and querying these records in Athena would fail due to invalid data format (HIVE_BAD_DATA).
What is the correct way to handle this situation?

A: I think my issue was actually stemming from some incorrect encode/decode logic in my original function.
You ultimately need the '\n' between each processed record, as detailed in the doc I linked.
I also found one more article that backed up this theory, so here are the two links:

*

*https://medium.com/analytics-vidhya/append-newline-to-amazon-kinesis-firehose-json-formatted-records-with-python-f58498d0177a

*https://aws.amazon.com/blogs/big-data/build-seamless-data-streaming-pipelines-with-amazon-kinesis-data-streams-and-amazon-kinesis-data-firehose-for-amazon-dynamodb-tables/
And here is the working Python code:
decoded_payload = json.loads(base64.b64decode(record["data"]).decode())
if decoded_payload["dynamodb"] and decoded_payload["dynamodb"]["NewImage"]: 
    updated_record = decoded_payload["dynamodb"]["NewImage"]
    deserialized_record = {
        k: deserializer.deserialize(v) for k, v in updated_record.items()
    }
    # Add newline after each record, otherwise Athena will only "see" the first
    record_with_line_break = json.dumps(deserialized_record, cls=DecimalEncoder) + "\n"
    output_record = {
        "recordId": record["recordId"],
        "result": "Ok",
        "data": base64.b64encode(record_with_line_break.encode()),
    }
    output.append(output_record)

