1. 项目背景与核心价值
去年在帮一家跨境电商做实时数仓改造时，我们遇到了一个典型的数据孤岛问题——用户行为数据存在Kinesis，交易日志在DynamoDB，而风控系统需要实时消费这些数据做联合分析。当时用Flink 1.13手动写了不少connector代码，现在Flink 2.2官方connector生态大幅增强后，终于可以扔掉那些定制化轮子了。
这个方案的核心价值在于：
- 一站式解决云原生环境下的实时数据通路问题
- 官方connector经过生产验证，稳定性远超自研实现
- 支持Exactly-Once语义，金融级数据一致性有保障
- 动态扩缩容能力应对业务高峰更从容
2. 技术架构解析
2.1 整体数据流向设计
典型的数据管道分为三层：
[Source Layer] -> [Processing Layer] -> [Sink Layer] AWS服务 Flink计算引擎 分析存储/服务实际业务中常见三种模式：
- Kinesis -> Flink -> ES：实时日志分析场景
- DynamoDB -> Flink -> MongoDB：订单状态同步场景
- Firehose -> Flink -> JDBC：审计数据归档场景
2.2 版本适配矩阵
| 组件 | Flink 2.2适配版本 | 关键特性 |
|---|---|---|
| Kinesis Connector | 2.4.0 | 支持Enhanced Fan-Out |
| DynamoDB Streams | 1.2.0 | 自动处理分片分裂 |
| ES Connector | 7.1.0 | 支持索引自动创建 |
| MongoDB Connector | 2.3.0 | Change Stream捕获 |
特别提示：避免混用不同大版本的connector，我们曾因ES connector版本冲突导致整夜数据延迟
3. 关键配置实战
3.1 Kinesis源配置示例
KinesisStreamSource<String> source = KinesisStreamSource.<String>builder() .setStreamName("user_click_stream") .setRegion("us-west-2") .setInitialPosition(InitialPosition.LATEST) .setDeserializer(new SimpleStringSchema()) .setConsumerConfig(Collections.singletonMap( "AWS_ACCESS_KEY_ID", "AKIAxxxxxxxx" )) .build();避坑指南：
- 生产环境一定要用IAM角色替代AK/SK
- 监控
MillisBehindLatest指标，超过1分钟要告警 - 建议开启Enhanced Fan-Out（每个消费者独立带宽）
3.2 DynamoDB CDC配置技巧
# flink-conf.yaml 必须配置 execution.checkpointing.interval: 30s execution.checkpointing.mode: EXACTLY_ONCE # 表定义需包含StreamSpecification aws.dynamodb.streams.record.format: "json"我们遇到过三个典型问题：
- 分片热点导致消费延迟 - 通过调整分区键解决
- TTL设置过短导致数据丢失 - 建议不少于7天
- 序列化冲突 - 强制指定Jackson版本2.12.x
3.3 多目标写入优化
当需要同时写入ES和MongoDB时：
DataStream<Event> stream = env.addSource(source); // 分支1：写入ES stream.addSink(new Elasticsearch7Sink<>( hosts, new EventElasticsearchSinkFunction() )); // 分支2：写入Mongo stream.addSink(MongoSink.sink( MongoSinkConfig.builder() .setDatabase("analytics") .setCollection("events") .build(), new EventMongoSerializer() ));性能调优参数：
- ES批量写入大小：建议500-1000条
- MongoDB写入超时：不少于30s
- 并行度设置：Kinesis分片数×2
4. 监控与运维要点
4.1 关键监控指标
| 指标类型 | 指标名称 | 健康阈值 |
|---|---|---|
| Source | currentEmitTimestampLag | < 60s |
| Sink | numRecordsOut | ≈ numRecordsIn |
| Checkpoint | lastCheckpointDuration | < checkpoint间隔 |
4.2 常见故障处理
案例1：ES写入拒绝
- 现象：
BulkProcessorException - 根因：索引mapping冲突
- 解决：预先创建带严格mapping的索引模板
案例2：DynamoDB流中断
- 现象：
ShardIteratorExpiredException - 根因：checkpoint间隔过长
- 解决：调整checkpoint间隔至30s内
5. 生产环境验证
在某跨境支付平台的实际运行数据：
- 日均处理：23亿事件
- 端到端延迟：98%在800ms内
- 峰值吞吐：12万事件/秒
- 连续运行：已稳定运行147天
关键成功因素：
- 使用Flink 2.2的增量checkpoint功能
- Kinesis消费者启用AGGREGATION
- ES使用bulk+refresh_interval=30s组合
6. 扩展应用场景
除了基础的数据同步，这套架构还能支持：
- 实时风控：DynamoDB变更触发规则计算
- 库存同步：Kinesis事件驱动库存缓存更新
- 用户画像：ES聚合结果回写MongoDB
最近我们还在尝试用JDBC connector实现：
- 实时数据归档到Snowflake
- 维表关联时自动拉取MySQL数据
- 将聚合结果写回OLTP数据库
