如何基于数据内容将Kinesis Firehose数据写入S3指定子文件夹？

当然可以实现！完全不用为每个ID单独创建Kinesis流，结合Kinesis Firehose的Lambda数据转换功能就能轻松搞定这个按id字段分流到S3指定子文件夹的需求，完美适配你ID数量众多的场景。

核心思路 #

利用Firehose的Lambda数据转换能力，在数据进入Firehose后触发Lambda函数：

1. 解析每条JSON记录中的id字段
2. 生成包含id和日期的动态S3前缀（比如你的例子345_2018_03_05/）
3. 告诉Firehose将这条数据写入对应前缀的子文件夹中

具体实现步骤 #

1. 创建并配置Lambda函数 #

这个函数负责解析数据、提取关键字段，并传递分区信息给Firehose。以下是Python示例代码：

import json
import base64
from datetime import datetime

def lambda_handler(event, context):
    output_records = []
    
    for record in event["records"]:
        try:
            # 解码Firehose传递的base64编码数据
            payload = base64.b64decode(record["data"]).decode("utf-8")
            data = json.loads(payload)
            
            # 提取id字段，转换为字符串避免格式问题
            user_id = str(data["id"])
            # 生成日期格式（这里用当前时间，也可以用数据自带的时间字段）
            target_date = datetime.now().strftime("%Y_%m_%d")
            
            # 构造返回给Firehose的记录，传递分区键信息
            output_record = {
                "recordId": record["recordId"],
                "result": "Ok",
                "data": record["data"],  # 保留原始数据，如需修改可在这里处理
                "metadata": {
                    "partitionKeys": {
                        "user_id": user_id,
                        "date": target_date
                    }
                }
            }
            output_records.append(output_record)
        
        except Exception as e:
            # 处理解析失败的记录，标记为处理失败，Firehose会将其写入错误路径
            error_record = {
                "recordId": record["recordId"],
                "result": "ProcessingFailed",
                "data": record["data"]
            }
            output_records.append(error_record)
    
    return {"records": output_records}

2. 配置Kinesis Firehose流 #

1. 创建（或修改）Firehose流，将S3设置为目标存储桶
2. 在数据转换部分，启用“启用数据转换”，关联你刚才创建的Lambda函数
3. 在S3目标配置中，设置前缀为：

   !{partitionKeys.user_id}_!{partitionKeys.date}/
   

   Firehose会自动替换Lambda返回的partitionKeys值，生成如345_2018_03_05/的子文件夹路径
4. 确保Firehose拥有调用该Lambda函数的权限（可以通过IAM角色自动配置）

关键注意事项 #

• 批量处理：Firehose会批量发送记录给Lambda（默认最多500条/批），所以函数要支持批量处理，避免单条处理的性能瓶颈
• 错误处理：示例中加入了异常捕获，失败的记录会被Firehose写入S3的错误前缀下（默认是error/），方便后续排查
• 日期准确性：如果你的数据本身包含时间字段，建议用数据里的时间生成日期前缀，避免因网络延迟导致的日期偏差
• 性能优化：根据你的数据量调整Lambda的内存配置（内存越高，CPU和网络性能越好），避免处理超时

这种方案只需要一个Firehose流和一个Lambda函数，就能支持无限数量的ID动态分流，完全符合你的需求。

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