AWS Lambda转换事件后如何将其重新发送到Kinesis Firehose

核心实现逻辑修正 #

作为Kinesis Firehose绑定的数据转换Lambda，你不需要手动调用Firehose SDK的putRecordBatch方法回传数据，直接在handleRequest方法中返回符合规范的结果对象即可，Firehose会自动完成后续的投递、失败记录分拣逻辑，手动调用SDK反而会导致数据重复、额外计费。

规范实现步骤 #

1. 修正Handler返回类型 #

你原来的Message类已经符合要求，只需要补充外层返回对象、修改Handler泛型定义即可：

// 外层响应对象，匹配Firehose要求的返回结构
public class FirehoseTransformationResponse {
    private List<Message> records;

    public List<Message> getRecords() {
        return records;
    }

    public void setRecords(List<Message> records) {
        this.records = records;
    }
}

// 调整Handler的泛型，返回自定义的响应对象
public class Application implements RequestHandler<KinesisFirehoseEvent, FirehoseTransformationResponse> {

    @Override
    public FirehoseTransformationResponse handleRequest(KinesisFirehoseEvent event, Context context) {
        EventService eventService = new EventService();
        // process方法改造为返回处理完成的Message集合
        List<Message> processedRecords = eventService.process(event);
        FirehoseTransformationResponse response = new FirehoseTransformationResponse();
        response.setRecords(processedRecords);
        return response;
    }
}

2. Message字段赋值要求 #

三个字段都要严格按照规则赋值：

• recordId：必须和你从输入的KinesisFirehoseEvent中拿到的对应记录的recordId完全一致，Firehose通过这个字段匹配原始记录
• data：必须是转换后的事件内容先转UTF-8字节数组、再Base64编码的字符串；如果最终投递到Splunk，建议每条内容末尾加\n再编码，避免Splunk把多条记录合并
• result：只能填三个枚举值：
  • Ok：处理成功，正常投递
  • ProcessingFailed：处理失败，Firehose会将这条记录归类到错误存储路径
  • Dropped：主动丢弃该记录，不计入错误也不投递

常见问题解答 #

能不能用HashMap构造状态模型？ #

完全可以。无论是自定义POJO还是HashMap，只要最终序列化后的JSON结构符合Firehose的要求，就能正常运行，两种方式没有本质区别。

为什么不同示例的Record结构不一样？ #

两个是完全不同场景的结构：

• 转换Lambda的返回结构，需要带recordId、result、data三个字段
• 直接向Firehose写入原始数据的PutRecordBatchRequest中的Record，仅需要data字段，自然存在差异

实践经验分享 #

• 单条记录处理异常时，将对应Message的result设为ProcessingFailed即可，不要抛出异常导致整批重试，避免数据重复
• Base64编码前一定要先把字符串转成UTF-8字节数组，不要直接编码字符串，避免乱码
• 投递到Splunk的场景，可以直接在data中拼接source、sourcetype等元数据，比在Firehose全局配置更灵活
• 建议开启Firehose的转换失败日志，方便排查单条记录的处理问题

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