AWS CDK Python：Kinesis Firehose S3备份记录如何添加换行符？

为什么你的处理配置未生效？ #

Kinesis Firehose 的 processing_configuration 仅作用于主S3目标（即转换为Parquet的存储桶）的处理流程，而S3备份桶的记录是Firehose在应用任何处理或格式转换之前就完成备份的原始数据。数据流的执行顺序为：

• 事件进入Firehose → 备份到S3备份桶（当s3_backup_mode="Enabled"时）→ 执行主目标的处理器 → 格式转换为Parquet → 写入主S3桶

因此你配置的AppendDelimiterToRecord处理器是在备份完成后才执行的，备份桶自然无法获取带换行符的记录。

能否给备份模式下的记录添加换行符？ #

可以，但需要调整处理逻辑的执行时机，让换行符在备份前就被添加到记录中，推荐两种可行方案：

方案1：使用Lambda处理器前置处理 #

在Firehose的输入阶段配置Lambda处理器，在记录备份前为每条JSON添加换行符。这样备份桶和主目标都会拿到带换行的原始记录，主目标后续仍可正常转换为Parquet格式。

示例CDK核心配置代码：

# 创建用于添加换行符的Lambda函数
add_newline_lambda = aws_lambda.Function(
    self, "AddNewlineLambda",
    runtime=aws_lambda.Runtime.PYTHON_3_11,
    handler="index.lambda_handler",
    code=aws_lambda.Code.from_inline("""
import base64
import json

def lambda_handler(event, context):
    output = []
    for record in event['records']:
        # 解码原始记录并添加换行符
        payload = base64.b64decode(record['data']).decode('utf-8') + '\\n'
        # 重新编码为base64格式返回
        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(payload.encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)
    return {'records': output}
""")
)

# 配置Firehose的顶层处理器（会在备份前执行）
processing_configuration = aws_kinesisfirehose.CfnDeliveryStream.ProcessingConfigurationProperty(
    enabled=True,
    processors=[
        aws_kinesisfirehose.CfnDeliveryStream.ProcessorProperty(
            type="Lambda",
            parameters=[
                aws_kinesisfirehose.CfnDeliveryStream.ProcessorParameterProperty(
                    parameter_name="LambdaArn",
                    parameter_value=add_newline_lambda.function_arn
                )
            ]
        )
    ]
)

# 创建Firehose流，将处理器配置在顶层
firehose_stream = aws_kinesisfirehose.CfnDeliveryStream(
    self, "EventFirehose",
    delivery_stream_type="DirectPut",
    extended_s3_destination_configuration=extended_s3_destination_config,
    processing_configuration=processing_configuration
)

# 授予Firehose调用Lambda的权限
add_newline_lambda.add_permission(
    "FirehoseInvokePermission",
    principal=aws_iam.ServicePrincipal("firehose.amazonaws.com"),
    action="lambda:InvokeFunction",
    source_arn=firehose_stream.attr_arn
)

方案2：调整主目标输出逻辑，同步备份JSON文件 #

如果你不需要保留完全原始的输入记录，可以关闭Firehose内置的S3备份功能，改为在主目标的格式转换配置中同时输出带换行的JSON文件，再通过S3跨区域复制或Lambda触发将JSON文件同步到备份桶。这种方式需要额外配置数据格式转换规则，相对复杂，优先推荐方案1。

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