将JSON推送到Firehose时格式转换失败，请求技术排查

问题根源 #

你当前的代码存在双重JSON序列化问题：

1. 先用stocks_df.to_json(orient='records')生成数组格式的JSON字符串，再通过strip('[]')得到单个对象的JSON字符串
2. 接着又用json.dumps()将这个JSON字符串再次序列化为JSON，导致最终发送给Firehose的数据是一个被引号包裹的字符串（比如"{\"Date\":\"2022-07-11\",\"Price\":0.16,\"Symbol\":\"ACQRW\"}"）
3. Firehose格式转换要求顶级数据必须是JSON对象或数组，而这种被包裹的字符串属于primitive类型，因此触发DataFormatConversion.MalformedData错误

修正方案 #

取消多余的json.dumps()操作，直接将DataFrame生成的JSON字符串编码为字节流（Firehose的Data字段要求字节类型），根据数据量选择以下两种方式：

方式1：推送整个JSON数组（适合小批量数据） #

def push_file_to_firehose(stocks_df):
    fh = boto3.client('firehose')
    # 生成数组格式的JSON字符串，保留顶级数组结构
    stocks_json = stocks_df.to_json(orient='records')
    logger.info(stocks_json)
    try: 
        fh.put_record(
            DeliveryStreamName='PUT-S3-bndnetworks', 
            # 直接将JSON字符串编码为字节，无需二次序列化
            Record={'Data': stocks_json.encode('utf-8')}
        )
        logger.info("Successfully pushed data into Firehose")
    except Exception as e:
        logging.error("Error while pushing data into Firehose")
        logging.error(e)

方式2：逐行推送单个JSON对象（适合大批量数据，避免单条Record过大） #

def push_file_to_firehose(stocks_df):
    fh = boto3.client('firehose')
    # 遍历DataFrame每一行，生成单个对象的JSON
    for _, row in stocks_df.iterrows():
        row_json = row.to_json()
        logger.info(row_json)
        try: 
            fh.put_record(
                DeliveryStreamName='PUT-S3-bndnetworks', 
                Record={'Data': row_json.encode('utf-8')}
            )
            logger.info("Successfully pushed a row into Firehose")
        except Exception as e:
            logging.error("Error while pushing row into Firehose")
            logging.error(e)

关键注意点 #

• Firehose的put_record接口要求Data字段必须是字节类型，因此必须用encode('utf-8')将JSON字符串转为字节流
• 如果只需要推送单行数据，也可以用row.to_json()直接生成单个对象的JSON，比strip('[]')更稳妥

内容的提问来源于stack exchange，提问作者Didier