Lambda无法向Kinesis Firehose返回记录的问题排查求助

核心错误点分析 #

• Lambda返回格式不符合要求：Kinesis Firehose要求转换函数必须返回包含records字段的对象，你的代码仅生成了处理后的数组但未按规范返回，导致Firehose无法识别结果。
• 时间处理逻辑存在冗余风险：通过new Date(time).toLocaleString()再转Date的方式会引入字符串解析环节，可能引发时区解析偏差。
• 数据双重序列化错误：finalRecord已是JSON.stringify的结果，后续再次执行JSON.stringify会导致数据被嵌套序列化，格式完全错误。
• 分区键类型不匹配：Firehose要求partitionKeys的所有值必须为字符串类型，你的代码中year/month/day是数字类型，会触发序列化异常。

修复后的Lambda代码 #

exports.handler = async (event, context) => {
    const output = event.records.map((record) => {
        // 解码并解析原始数据
        const firehoseData = Buffer.from(record.data, 'base64').toString();
        const payload = JSON.parse(firehoseData);
        
        // 直接用毫秒时间戳初始化Date，避免二次解析
        const timestamp = new Date(payload.time);
        // 用Intl.DateTimeFormat直接格式化IST时间
        const formatter = new Intl.DateTimeFormat('en-US', {
            timeZone: 'Asia/Kolkata',
            day: '2-digit',
            month: '2-digit',
            year: 'numeric',
            hour: '2-digit',
            minute: '2-digit',
            second: '2-digit',
            hour12: false
        });
        const [datePart, timePart] = formatter.format(timestamp).split(', ');
        const finalTimestamp = `${datePart.replace(/\//g, '-')} ${timePart}`;
        
        // 构造最终记录（无需提前序列化）
        const finalRecord = {
            ...payload,
            time: finalTimestamp
        };
        
        // 确保分区键值为字符串类型
        const partitionKeys = {
            site_name: payload.device_name,
            year: timestamp.getFullYear().toString(),
            month: (timestamp.getMonth() + 1).toString().padStart(2, '0'),
            day: timestamp.getDate().toString().padStart(2, '0')
        };
        
        return {
            recordId: record.recordId,
            result: 'Ok',
            // 仅序列化一次最终记录
            data: Buffer.from(JSON.stringify(finalRecord)).toString('base64'),
            metadata: { partitionKeys: partitionKeys }
        };
    });
    
    // 按Firehose要求返回包含records字段的结果
    return { records: output };
};

额外验证步骤 #

• 权限确认：你的Lambda已拥有kinesis:PutRecords和基础执行角色，满足Firehose调用的权限要求，无需调整。
• 单条记录测试：在Lambda控制台使用模拟Firehose格式的测试事件（包含records数组，每个元素含recordId和base64编码的data），验证返回格式是否合规。
• 错误日志复盘：若修复后仍有问题，查看S3 error/前缀下的日志，重点关注errorMessage中的具体栈信息（Lambda抛出未捕获异常时，Firehose会返回更详细的错误内容）。

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