Format Conversion Failed when Pushing JSON File into Firehose

user_2300967 0
Viewed 78

I have a pandas dataframe stocks_df which is as follows:

| Date      | Price|Symbol|
|-----------|------|------|
| 2022-07-10|150.54| APPL |

I need to convert it to JSON and push it into firehose

{"Date":"2022-07-11","Price":65.09,"Symbol":"ACM"}

I am getting some error and a folder is created in my S3 bucket format-conversion-failed/ with the following in the file:

{"attemptsMade":1,"arrivalTimestamp":1657602058357,"lastErrorCode":"DataFormatConversion.MalformedData","lastErrorMessage":"The input JSON contained a primitive at the top level. The top level must be an object or array.","attemptEndingTimestamp":1657602335593,"rawData":"IntcIkRhdGVcIjpcIjIwMjItMDctMTFcIixcIlByaWNlXCI6OC4yNixcIlN5bWJvbFwiOlwiQUNSXCJ9Ig==","sequenceNumber":"49631306863317095792192735938176634752612190707410534402","subSequenceNumber":null,"dataCatalogTable":{"catalogId":null,"databaseName":"XXXXXX","tableName":"XXXXXXXXX","region":"us-east-1","versionId":"LATEST","roleArn":"arn:aws:iam::XXXXXXXXXXXXXXXX:role/service-role/XXXXXXXXXXXXXXXXXXXXXXX-PUT-S3-bndnet-us-east-1-XXXXXXXXXXXXXXX"}}

this is the decoded value from RawData: "{\"Date\":\"2022-07-11\",\"Price\":0.16,\"Symbol\":\"ACQRW\"}"

This is my function:

def push_file_to_firehose(stocks_df):
    fh = boto3.client('firehose')
    stocks_json = stocks_df.to_json(orient='records').strip('[]')
    logger.info(stocks_json)
    try: 
        fh.put_record(DeliveryStreamName='PUT-S3-bndnetworks', \
        Record={'Data': json.dumps(stocks_json)} )
        logger.info("Successfully pushed the file into Firehose")
    except Exception as e:
        logging.error("Error while pushing the file into Firehose")
        logging.error(e)
amazon-kinesis-firehose pandas python json
1 Answers

The value of Data must be base64 encoded (bytes), not a JSON string (str).

Example:

import base64
import json

fh.put_record(
    DeliveryStreamName='PUT-S3-bndnetworks',
    Record={
        'Data': base64.b64encode(
            json.dumps(stocks_json).encode('utf-8')
        ).decode('utf-8')
    }
)

Reference: put_record() (boto) (admittedly, the documentation is kind of confusing)

user_14360310
Related