Flink 2.2 连接器打通 AWS（DynamoDB/Kinesis/Firehose）与 Elasticsearch/MongoDB/JDBC 的数据链路

1. 先搭一张“全链路”脑图

一个典型的链路长这样（按你给的 connector 组合）：

• 上游数据产生变化

  • DynamoDB 表的 CDC：DynamoDB Streams → Flink（DataStream Source / SQL Source）(Apache Nightlies)
  • 事件流：Kinesis Data Streams → Flink（SQL Connector）(Apache Nightlies)

• Flink 做实时 ETL / 聚合 / 维表关联

  • 维表：MongoDB / JDBC（Temporal Join + Lookup Cache）(Apache Nightlies)

• 下游交付与检索

  • Firehose Sink（投递到 S3/Redshift/OpenSearch 等典型目的地）(Apache Nightlies)
  • Elasticsearch Sink（DataStream 或 SQL 写入索引）(Apache Nightlies)
  • MongoDB/JDBC Upsert 落库(Apache Nightlies)

2. DynamoDB：用 Streams 做 CDC（源）+ BatchWriteItem（汇）

2.1 Streams Source：从 DynamoDB Streams 读变更

Flink 的DynamoDbStreamsSource直接消费 DynamoDB Streams，起始位置可选LATEST或TRIM_HORIZON（24 小时保留窗口，越早越可能被裁剪）。(Apache Nightlies)

核心代码骨架（按你贴的写法整理）：

ConfigurationsourceConfig=newConfiguration();sourceConfig.set(DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION,DynamodbStreamsSourceConfigConstants.InitialPosition.TRIM_HORIZON);DynamoDbStreamsSource<String>source=DynamoDbStreamsSource.<String>builder().setStreamArn("arn:aws:dynamodb:...:table/test/stream/...").setSourceConfig(sourceConfig).setDeserializationSchema(dynamodbDeserializationSchema).build();DataStream<String>stream=env.fromSource(source,WatermarkStrategy.<String>forMonotonousTimestamps().withIdleness(Duration.ofSeconds(1)),"DynamoDB Streams source");

事件顺序性（非常实用）

DynamoDB Streams 的顺序语义是：同一主键的事件会被写入同一条 shard lineage，因此在同一主键范围内保持有序；发生 shard split 时，只要先读完 parent shard，再读 child shard，顺序仍能保持。Flink 的实现会尊重 parent-child shard 关系，避免“子 shard 抢跑”。(Apache Nightlies)

这对“订单状态流”“用户画像更新”等场景很关键：你通常可以把同一主键当成天然的顺序分区，减少自己在应用层做乱序合并的成本。

2.2 DynamoDB Sink：BatchWriteItem 批写

Sink 侧是直接走 DynamoDB 的BatchWriteItem批量写入能力（你贴的 DynamoDB Connector 文档就写了这一点）。(Apache Nightlies)
落地建议：优先设计幂等写（例如主键覆盖写、版本号条件写），否则在失败恢复重放时会出现重复副作用。

3. Kinesis Data Firehose Sink：适合“投递型”下游

Firehose Sink 用 AWS v2 SDK 写到 delivery stream，builder 里比较常用的调参点就是：

• setMaxBatchSize：批次数量上限（默认 500）
• setMaxBatchSizeInBytes：批次字节上限（默认 4MiB）
• setMaxInFlightRequests/setMaxBufferedRequests：吞吐与背压关键旋钮
• setMaxTimeInBufferMS：缓冲超时强制 flush
• setMaxRecordSizeInBytes：单条过大直接拒绝（默认 1000KiB）(Apache Nightlies)

本地/测试环境常见需求是对接 Localstack 或 VPC Endpoint：文档明确支持用aws.endpoint覆盖，并要求同时配置 region 参与签名。(Apache Nightlies)

4. Elasticsearch：两条路（DataStream Sink / SQL Sink），语义与调参不一样

4.1 DataStream Elasticsearch Sink：BulkProcessor + Checkpoint（At-least-once）

DataStream 版本的 ES Sink 每个并行实例内部用BulkProcessor缓冲并批量 flush，请求是串行 flush（不会并发两个 flush）。(Apache Nightlies)

容错语义方面：启用 checkpoint 后，sink 会在 checkpoint 时等待当时 pending 的 action 都被 ES ack，从而提供at-least-once。(Apache Nightlies)

你贴的代码示例是setBulkFlushMaxActions(1)（每条都 flush），这适合 demo，但线上通常会配合：

• setBulkFlushMaxActions
• setBulkFlushMaxSizeMb
• setBulkFlushInterval
• 以及失败重试的 backoff（指数/常量）(Apache Nightlies)

实战提醒：开启 backoff 会延长 checkpoint 时长，因为 checkpoint 也要等重试后的请求刷出去；指数退避+高重试次数在 ES 拥塞时会把 checkpoint 拉得很长。(Apache Nightlies)

4.2 Elasticsearch SQL Connector：更“声明式”，但要读懂 3 个关键点

SQL 侧建表非常直接：

CREATETABLEmyUserTable(user_id STRING,user_name STRING,uvBIGINT,pvBIGINT,PRIMARYKEY(user_id)NOTENFORCED)WITH('connector'='elasticsearch-7','hosts'='http://localhost:9200','index'='users');

核心要点：

1. Append vs Upsert
   定义了主键就进入 upsert（可消费 UPDATE/DELETE changelog）；没主键只能 append（仅 INSERT）。(Apache Nightlies)

2. 文档 ID 的生成规则
   ES 文档 id 由主键字段按顺序拼接而成，分隔符由document-id.key-delimiter控制；同时文档 id 有长度与格式限制（最多 512 bytes、不能有空白等），因此主键字段类型要选“字符串表示稳定”的。(Apache Nightlies)

3. 动态 Index 能力与限制
   index支持：

• 静态：users
• 动态：users-{log_ts|yyyy-MM-dd}
• 甚至now()格式化
  但注意：用系统时间生成动态 index 时，对于 changelog（同主键多次更新）无法保证每次落到同一个 index，因此这种写法只适合 append-only 流。(Apache Nightlies)

同时，SQL sink 的容错强相关参数是sink.flush-on-checkpoint：关掉后 checkpoint 不会等 ES ack，因此就别指望有强 at-least-once 语义了。(Apache Nightlies)

5. Kinesis Data Streams SQL Connector：版本分裂与迁移要小心

Kinesis SQL Connector 文档里把“版本分裂”讲得很直白：因为从旧的SourceFunction/SinkFunction迁移到新的Source/Sink，出现了不同发行物与 connector identifier 的组合，而且同一应用依赖里只能有一个 identifier 为kinesis的 TableFactory，否则会冲突。(Apache Nightlies)

另外迁移 v4 → v5（文档的说法）没有 state 兼容，需要你用AT_TIMESTAMP在停旧任务前后做衔接，可能会重放一小段数据。(Apache Nightlies)

实战选型建议（按文档语义总结）：

• 想用新 Source：优先kinesis（Source）发行物
• 需要 metadata VIRTUAL 列（到达时间、shard-id、sequence-number）时，文档提示当前kinesisSource 有已知 bug，临时用kinesis-legacy更稳。(Apache Nightlies)

6. MongoDB SQL Connector：Upsert、分片写入与 Lookup Cache

MongoDB SQL Connector 的关键能力点：

• 主键决定写入模式：有 PK → upsert（支持 UPDATE/DELETE），没 PK → append only。(Apache Nightlies)
• 幂等写：使用 upsert 模式（update(..., { upsert: true })）写入，并把主键组合成_id，天然具备失败恢复重放时的幂等性。(Apache Nightlies)
• 分片集合 upsert：如果 MongoDB 是 sharded collection，过滤条件里需要带 shard key；Flink SQL 通过PARTITIONED BY声明 shard key，从每条记录中取值填到 filter。(Apache Nightlies)
• 维表 Join 性能：支持 Lookup Cache（lookup.cache=PARTIAL）按 TaskManager 进程级缓存，配合 max-rows 与 expire-after-* 在“吞吐 vs 新鲜度”间权衡。(Apache Nightlies)

7. JDBC SQL Connector：最常用的“万能落库/维表”方案

JDBC SQL Connector 的主线逻辑和 MongoDB 很像：

• 有主键 → upsert；无主键 → append only，且底层库唯一键冲突会直接失败。(Apache Nightlies)
• 支持 Partitioned Scan：通过scan.partition.column/num/lower-bound/upper-bound做并行读（批处理提速很明显）。(Apache Nightlies)
• 维表 Join：同样是同步 lookup，支持 PARTIAL cache 与 TTL/容量控制。(Apache Nightlies)
• Upsert 语法随数据库方言变化（MySQL/PG/Oracle/SQLServer 等）。(Apache Nightlies)
• 还有 JdbcCatalog（目前重点支持 Postgres/MySQL Catalog 的部分能力）。(Apache Nightlies)

8. 一套通用的“可靠性与调参”经验

你贴的这些 connector 其实都在围绕同一组工程问题打转：

8.1 语义保证：优先幂等，其次依赖 checkpoint

• 各类 sink 大多默认at-least-once；真正“接近 exactly-once”的做法往往是：

  • 让下游写入幂等（MongoDB/JDBC 的 upsert、ES 的 deterministic id + upsert 思路）(Apache Nightlies)
  • 再配合 checkpoint 等待 flush（ES SQL 的sink.flush-on-checkpoint、ES DataStream 的 pending 等待机制）(Apache Nightlies)

8.2 背压三件套：batch、in-flight、buffer

几乎每个 sink builder 都有：

• maxBatchSize/maxBatchSizeInBytes
• maxInFlightRequests
• maxBufferedRequests
• maxTimeInBufferMS（或 flush interval）

建议你把它们当成一个整体去调：吞吐上不去时别只盯 batch，in-flight 太小也会“跑不满”；但 in-flight 太大又会把下游打爆并拖慢 checkpoint。

8.3 本地联调：优先 Localstack + endpoint override

DynamoDB/Firehose 文档都明确支持自定义 endpoint（常用于 Localstack 或 VPC endpoint），同时强调 region 用于签名。(Apache Nightlies)

9. 打包部署：uber-jar 或 Flink lib，两条路选一条

文档多次强调：不少 streaming connector / SQL jar不在二进制发行包里，要么：

• 你打成 uber-jar（把 connector 依赖都带上）
• 要么把 connector jar 放到 Flink 的lib/，让集群系统级可见(Apache Nightlies)

Kinesis 还要额外注意不要同时引入两个带kinesisTableFactory 的发行物，否则启动就会 factory 冲突。(Apache Nightlies)

10. 小结

• DynamoDB Streams 适合做“表变更 CDC”，并能在同主键范围内保证顺序性(Apache Nightlies)
• Kinesis Data Streams 更像“事件总线”，SQL connector 版本分裂要看清楚kinesis/kinesis-legacy与 TableFactory 冲突规则(Apache Nightlies)
• Firehose 是投递型 sink，靠 batch/in-flight/buffer 把吞吐打满(Apache Nightlies)
• Elasticsearch 写入既可走 DataStream（BulkProcessor）也可走 SQL（动态 index、flush-on-checkpoint），语义与调参点不同(Apache Nightlies)
• MongoDB/JDBC 作为维表与落库，核心在：主键决定 upsert、lookup cache 决定 join 性能、幂等决定恢复成本(Apache Nightlies)