Flink Firehose Sink 把实时流数据稳定写进 Amazon Kinesis Data Firehose

1、先看版本坑：Flink 2.2 目前没有可用的 Firehose Connector

如果你正在用Flink 2.2，官方文档明确写了：Flink 2.2 暂无可用的 Firehose connector；PyFlink 侧也标注暂无 SQL jar。 (nightlies.apache.org)

如果你用的是已发布的稳定版本（例如 1.20），对应的 Maven 依赖在中央仓库是有的；比如org.apache.flink:flink-connector-aws-kinesis-firehose的最新版本之一是5.1.0-1.20（2026-01-12 发布）。 (mvnrepository.com)

实战建议：
你如果必须在 2.2 上接 Firehose，短期通常只有两条路：等待 2.2 适配版本发布，或评估切换到已有连接器的稳定版本线（按你公司的版本策略来）。

2、工作原理一句话：AWS SDK v2 + 批量写入

KinesisFirehoseSink使用AWS v2 SDK for Java将 Flink 流中的数据写入 Firehose delivery stream。 (nightlies.apache.org)

你需要做的核心只有两件事：

• 配好 AWS 区域/凭证等 client 参数
• 提供序列化方式（把你的对象变成要发给 Firehose 的字节/字符串）

3、Java 快速开始：最小可用 Demo

下面就是官方示例的核心写法（你可以直接放进博客当“跑通第一条”）：

PropertiessinkProperties=newProperties();// RequiredsinkProperties.put(AWSConfigConstants.AWS_REGION,"eu-west-1");// Optional, provide via alternative routes e.g. environment variablessinkProperties.put(AWSConfigConstants.AWS_ACCESS_KEY_ID,"aws_access_key_id");sinkProperties.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY,"aws_secret_access_key");KinesisFirehoseSink<String>kdfSink=KinesisFirehoseSink.<String>builder().setFirehoseClientProperties(sinkProperties)// Required.setSerializationSchema(newSimpleStringSchema())// Required.setDeliveryStreamName("your-stream-name")// Required.setFailOnError(false)// Optional.setMaxBatchSize(500)// Optional.setMaxInFlightRequests(50)// Optional.setMaxBufferedRequests(10_000)// Optional.setMaxBatchSizeInBytes(4*1024*1024)// Optional.setMaxTimeInBufferMS(5000)// Optional.setMaxRecordSizeInBytes(1000*1024)// Optional.build();flinkStream.sinkTo(kdfSink);

这段代码里真正“必须有”的只有 3 个：region、serializationSchema、deliveryStreamName。 (nightlies.apache.org)

4、配置项逐个讲透：吞吐、延迟、背压都靠它们

Flink Firehose Sink 的 builder 提供了一组很典型的“批量写入 + 流控”参数（括号里是默认值）： (nightlies.apache.org)

4.1 必选项

• setFirehoseClientProperties(...)：AWS 区域、凭证等 client 参数（必选） (nightlies.apache.org)
• setSerializationSchema(...)：序列化方式（必选） (nightlies.apache.org)
• setDeliveryStreamName(...)：目标 delivery stream 名（必选） (nightlies.apache.org)

4.2 错误处理

• setFailOnError(false)：写入失败是否当作致命异常（默认 false）。

  • 你要“宁可停任务也不能丢数据/漏写”：考虑设为true
  • 你要“允许部分失败，先跑起来再补偿”：用false，但一定要配合监控与告警 (nightlies.apache.org)

4.3 批量写入三件套（决定吞吐与延迟）

• setMaxBatchSize(500)：单批最大 record 数（默认 500） (nightlies.apache.org)
• setMaxBatchSizeInBytes(4 * 1024 * 1024)：单批最大字节数（默认 4MiB） (nightlies.apache.org)
• setMaxTimeInBufferMS(5000)：单条记录在缓冲区最多停留多久就必须 flush（默认 5000ms） (nightlies.apache.org)

为什么默认值看起来“刚好卡在某些数字上”？因为 Firehose 的PutRecordBatch本身就有限制：

• 每次PutRecordBatch最多 500 条
• 每次请求总大小最多 4 MiB
• 单条 record最大 1000 KB（base64 前）
  而且这些限制“不能调整”。 (AWS 文檔)

所以你调参时的底线是：

• maxBatchSize不要超过 500（超过也没意义，最终还是要拆）
• maxBatchSizeInBytes不要超过 4MiB（否则必然被服务端拒绝）
• maxRecordSizeInBytes建议与 1000KB 保持一致或更保守（避免大字段把你批次顶爆） (AWS 文檔)

4.4 背压与缓冲：防止“把下游写爆”

• setMaxInFlightRequests(50)：在途请求上限，超过就开始背压（默认 50） (nightlies.apache.org)
• setMaxBufferedRequests(10_000)：最多可缓存多少条记录，超过就背压（默认 10000） (nightlies.apache.org)

简单理解：

• in-flight控制“同时飞出去多少个批次”
• buffered控制“本地最多囤多少条还没发出去的数据”
  两者都是保护下游（Firehose）与保护 Flink 自己内存的关键阀门。

5、两套常用调参模板（直接抄到生产排障手册里）

5.1 低延迟优先（更快落地）

• maxTimeInBufferMS调小（例如 200~1000ms）
• maxBatchSize不用太大（例如 50~200）
• maxInFlightRequests视下游能力适当提高，但别把 Firehose 写到限流

适用：告警链路、在线检索/近实时看板。

5.2 高吞吐优先（更省请求、更高写入效率）

• maxBatchSize靠近 500
• maxBatchSizeInBytes靠近 4MiB，但要考虑单条 record 体积波动
• maxTimeInBufferMS可略大（例如 2s~5s）

适用：日志、埋点、离线落盘到 S3 的“吞吐型”链路。

这些参数的“合法上限”本质受 Firehose API quota 约束（500 records / 4MiB / 1000KB）。 (AWS 文檔)

6、PyFlink 怎么用：记得带依赖 JAR

Flink 2.2 文档同样提示：PyFlink 使用需要对应的依赖 JAR，但 2.2 暂无 SQL jar。 (nightlies.apache.org)

示例（官方风格）：

sink_properties={'aws.region':'eu-west-1','aws.credentials.provider.basic.accesskeyid':'aws_access_key_id','aws.credentials.provider.basic.secretkey':'aws_secret_access_key'}kdf_sink=KinesisFirehoseSink.builder()\.set_firehose_client_properties(sink_properties)\.set_serialization_schema(SimpleStringSchema())\.set_delivery_stream_name('your-stream-name')\.set_fail_on_error(False)\.set_max_batch_size(500)\.set_max_in_flight_requests(50)\.set_max_buffered_requests(10000)\.set_max_batch_size_in_bytes(4*1024*1024)\.set_max_time_in_buffer_ms(5000)\.set_max_record_size_in_bytes(1*1024*1024)\.build()

7、本地联调神器：自定义 Endpoint（VPC Endpoint / Localstack）

很多团队会用 Localstack 做集成测试，或者在 VPC Endpoint 场景下需要指定 endpoint。Flink 文档给了明确的配置方式：同时设置AWS_ENDPOINT与AWS_REGION（region 用于签名）。 (nightlies.apache.org)

PropertiesproducerConfig=newProperties();producerConfig.put(AWSConfigConstants.AWS_REGION,"us-east-1");producerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID,"aws_access_key_id");producerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY,"aws_secret_access_key");producerConfig.put(AWSConfigConstants.AWS_ENDPOINT,"http://localhost:4566");

8、写在最后：上线检查清单

• delivery stream 已创建且权限正确（否则全是失败重试/报错） (nightlies.apache.org)
• maxBatchSize / maxBatchSizeInBytes / maxRecordSizeInBytes不要突破 Firehose 的硬限制 (AWS 文檔)
• failOnError=false时要补齐：失败监控、告警、补偿/重放策略（否则“悄悄丢”最致命） (nightlies.apache.org)
• Flink 2.2 用户先确认：当前版本确实没有官方 connector（避免在依赖上耗一整天） (nightlies.apache.org)