当前位置： 首页 > backend >正文

Flink架构概览，Flink DataStream API 的使用，FlinkCDC的使用

backend 2025/5/22 6:14:46

一、Flink与其他组件的协同

Flink 是一个分布式、高性能、始终可用、准确一次（Exactly-Once）语义的流处理引擎，广泛应用于大数据实时处理场景中。它与 Hadoop 生态系统中的组件可以深度集成，形成完整的大数据处理链路。下面我们从 Flink 的 核心架构 出发，结合与 Hadoop 组件协同方式，详细剖析 Flink 的作用。

1. Flink 核心架构详解

1）架构组件图概览

+-------------------------+
|       Client            |
+-------------------------+|v
+-------------------------+
|    JobManager (JM)      |  <-- Master 负责调度
+-------------------------+|v
+-------------------------+
|    TaskManagers (TM)    |  <-- Worker 执行算子任务
+-------------------------+|v
+-------------------------+
|         Slot            |  <-- 执行资源单位
+-------------------------+

2）核心组件职责

组件描述
Client提交作业到 Flink 集群，触发作业执行。
JobManager (JM)管理作业生命周期，负责调度任务、故障恢复、协调检查点（Checkpoint）等。
TaskManager (TM)具体执行作业的物理任务（算子），负责数据交换、状态管理等。
SlotTaskManager 内部的资源单位，用于任务部署。每个 TaskManager 有多个 Slot。

3）状态管理与容错

  • Checkpoint/Savepoint：可恢复一致性状态（Exactly Once）

  • State Backend：保存状态（如 RocksDB、FsStateBackend）

  • Recovery：通过重放 Checkpoint 恢复任务

2. Flink 与 Hadoop 各组件的协同关系

Flink 虽然是独立系统，但能与 Hadoop 生态的多个关键组件协同工作，构建完整的大数据平台。

1）与 HDFS（Hadoop Distributed File System）

协同方式描述
输入源Flink 可直接读取 HDFS 中的批量数据（如 ORC、Parquet、Text 等格式）
状态后端Flink Checkpoint/Savepoint 可存储到 HDFS 上，保证高可用与容灾
输出目标Flink 作业可以将计算结果输出到 HDFS，作为后续离线处理的数据
fs.defaultFS: hdfs://namenode:8020
state.backend: filesystem
state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints/

2）与 Hive

协同方式描述
读取表数据Flink 可通过 Hive Catalog 与 Hive 元数据打通，直接读取 Hive 表
写入表Flink SQL 可将流式数据写入 Hive（使用 INSERT INTO）
统一元数据Flink + Hive Catalog 支持表结构共享，便于湖仓一体实践
CREATE CATALOG my_hive WITH ('type' = 'hive','hive-conf-dir' = '/etc/hive/conf'
);

3）与 Kafka（实时采集）

协同方式描述
实时数据源Flink 通过 Kafka Source 接收实时数据流（如日志、订单等）
下游结果写入Flink 可将流式计算结果写入 Kafka（供下游消费）
Exactly Once 语义Flink + Kafka + Checkpoint 可实现端到端的精确一次语义
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
consumer.setStartFromGroupOffsets();
consumer.setCommitOffsetsOnCheckpoints(true);

4）与 HBase（实时查询）

协同方式描述
维表关联Flink 可使用 HBase 作为维表进行流批 Join，实时补充维度数据
实时写入计算结果可实时写入 HBase，支持下游查询系统使用（如用户画像等）
tableEnv.executeSql("CREATE TABLE hbase_dim (...) WITH ('connector' = 'hbase-2.2', ...)");

5）与 YARN

协同方式描述
资源调度Flink 可部署在 YARN 上，利用 Hadoop 的资源调度管理能力
Session / Per-Job 模式支持多租户资源隔离或每个作业独立资源隔离部署
flink run -m yarn-cluster -ynm my-flink-job myjob.jar

6）与 Zookeeper

协同方式描述
高可用 JobManager使用 Zookeeper 实现 JobManager 的 leader election
Checkpoint HA 元数据存储配合 HDFS 存储 Checkpoint 元数据路径
high-availability: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.storageDir: hdfs://namenode:8020/flink/ha/

3. Flink 的作用总结

模块Flink 的角色
实时数据处理核心组件，进行低延迟、高吞吐流处理计算
数据清洗与 ETL提供强大 SQL / DataStream API 进行多源数据处理与聚合
实时指标计算支持实时 KPI、UV/PV、订单流等分析
数据湖构建可作为流式数据入湖的计算引擎（结合 Hudi/Iceberg）
实时监控预警搭配 Kafka + Prometheus，构建告警与监控系统
实时数仓建设联合 Kafka + Hive + HDFS + HBase 构建流批一体数仓体系

4. Flink 架构在 Hadoop 平台的实际部署图

                     +-------------+|  Flume/Nginx|+------+------+|Kafka集群|+-------------------+--------------------+|                                        |+---v---+                               +----v----+| Flink |--> 清洗 → 维表 Join → 计算    |  Spark  |+---+---+                               +----+----+|                                        |
+-------v---------+                     +--------v--------+
| HBase/Redis     |                     |  HDFS / Hive    |
+-----------------+                     +-----------------+

二、Flink DataStream API的使用

现在以 Flink DataStream API 为核心，深入剖析一个真实生产场景的 从 Kafka 到 Kafka 的流式处理全流程，包括：

  1. 项目结构与依赖

  2. 数据模型与清洗

  3. 水位线与乱序处理

  4. 异步维表查询（HBase/MySQL/Redis）

  5. 窗口聚合逻辑

  6. 数据下发（Kafka Sink）

  7. 容错机制与 Checkpoint 配置

1. 项目结构与依赖

1）Maven 依赖（pom.xml）

<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.17.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>1.17.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>3.0.1-1.17</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.14.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-hbase-2.2</artifactId><version>1.17.1</version></dependency>
</dependencies>

2. 数据模型定义

1）订单数据结构（OrderEvent）

public class OrderEvent {public String orderId;public String userId;public String productId;public double price;public int quantity;public long orderTime; // epoch millis
}

2） 商品维度（ProductInfo）

public class ProductInfo {public String productId;public String categoryId;public String productName;
}

3）聚合结果结构（OrderStat）

public class OrderStat {public String categoryId;public long windowStart;public long windowEnd;public double totalAmount;
}

3. Kafka Source + JSON 反序列化

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("kafka:9092").setTopics("order_events").setGroupId("flink-consumer").setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();DataStream<OrderEvent> orderStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "KafkaSource").map(json -> new ObjectMapper().readValue(json, OrderEvent.class)).returns(OrderEvent.class);

4. 水位线处理（乱序数据支持）

WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((event, ts) -> event.orderTime);DataStream<OrderEvent> orderStreamWithWM = orderStream.assignTimestampsAndWatermarks(watermarkStrategy);

5. 异步维表关联（以 HBase 为例）

使用 AsyncFunction 实现异步查询（支持 Redis/HBase/MySQL）

示例实现：AsyncProductEnrichmentFunction

public class AsyncProductEnrichmentFunction extends RichAsyncFunction<OrderEvent, Tuple2<OrderEvent, ProductInfo>> {private transient HBaseClient hBaseClient;@Overridepublic void open(Configuration parameters) throws Exception {hBaseClient = new HBaseClient("hbase.zookeeper.quorum");}@Overridepublic void asyncInvoke(OrderEvent input, ResultFuture<Tuple2<OrderEvent, ProductInfo>> resultFuture) {CompletableFuture.supplyAsync(() -> hBaseClient.queryProductInfo(input.productId)).thenAccept(productInfo -> resultFuture.complete(Collections.singletonList(Tuple2.of(input, productInfo))));}@Overridepublic void close() throws Exception {hBaseClient.close();}
}

应用异步函数

DataStream<Tuple2<OrderEvent, ProductInfo>> enrichedStream = AsyncDataStream.unorderedWait(orderStreamWithWM,new AsyncProductEnrichmentFunction(),5, TimeUnit.SECONDS, 100
);

6. 按类目 ID 滚动窗口聚合

DataStream<OrderStat> resultStream = enrichedStream.map(tuple -> new Tuple3<>(tuple.f1.categoryId, tuple.f0.orderTime, tuple.f0.price * tuple.f0.quantity)).returns(Types.TUPLE(Types.STRING, Types.LONG, Types.DOUBLE)).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, Long, Double>>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((t, ts) -> t.f1)).keyBy(t -> t.f0).window(TumblingEventTimeWindows.of(Time.minutes(1))).aggregate(new AggregateFunction<Tuple3<String, Long, Double>, Double, OrderStat>() {private long windowStart, windowEnd;private String categoryId;public Double createAccumulator() { return 0.0; }public Double add(Tuple3<String, Long, Double> value, Double acc) {categoryId = value.f0;return acc + value.f2;}public OrderStat getResult(Double acc) {return new OrderStat(categoryId, windowStart, windowEnd, acc);}public Double merge(Double acc1, Double acc2) {return acc1 + acc2;}}, new ProcessWindowFunction<OrderStat, OrderStat, String, TimeWindow>() {@Overridepublic void process(String key, Context context, Iterable<OrderStat> elements, Collector<OrderStat> out) {OrderStat stat = elements.iterator().next();stat.windowStart = context.window().getStart();stat.windowEnd = context.window().getEnd();out.collect(stat);}});

7. 写入 Kafka Sink

KafkaSink<OrderStat> kafkaSink = KafkaSink.<OrderStat>builder().setBootstrapServers("kafka:9092").setRecordSerializer(KafkaRecordSerializationSchema.builder().setTopic("order_stats").setValueSerializationSchema(stat -> {ObjectMapper mapper = new ObjectMapper();return mapper.writeValueAsBytes(stat);}).build()).setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE).build();resultStream.sinkTo(kafkaSink);

8.  容错与 HA 配置（关键）

1）Checkpoint 配置

env.enableCheckpointing(60000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.setStateBackend(new RocksDBStateBackend("hdfs://namenode/flink/checkpoints"));

2）高可用配置（flink-conf.yaml）

high-availability: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181
state.checkpoints.dir: hdfs://namenode/flink/checkpoints
state.savepoints.dir: hdfs://namenode/flink/savepoints

9. 运行命令（on YARN）

flink run -m yarn-cluster -c com.company.OrderRealtimeJob your-job.jar

10. 监控与排障建议

工具功能
Flink Web UI监控 Task、Checkpoint、Watermark
Prometheus指标采集
Grafana可视化
AlertManager告警配置
Savepoint容错恢复点

三、FlinkCDC实时采集数据入湖

解析Flink CDC（Change Data Capture）在大数据体系中的使用方法，并结合 Kafka、Hudi、Iceberg、Hive、HDFS 等大数据组件，提供一套 可落地、可执行、可扩展的完整集成方案。

1. Flink CDC 简介

Flink CDC 是 Apache Flink + Debezium 的组合，用于实时采集 MySQL/PostgreSQL 等数据库的变更数据（INSERT/UPDATE/DELETE），并以 流式方式传递到下游系统（Kafka、Hudi、Iceberg、HBase 等）。

2. 典型架构场景：Flink CDC + Hudi + Hive 实时数据湖方案

           +-------------+               +---------------------+| MySQL/Postgres            |                     ||    Source DB   +-------->  | Flink CDC Connector |+-------------+             |                     |+----------+----------+|| Row-level ChangeLogv+----------+----------+|     Flink Job       ||   (数据清洗/处理)   |+----------+----------+|v+----------+----------+| Hudi Sink (Flink)   |+----------+----------+|v+-------------+-------------+| Hive / Presto / Trino     || 实时查询（支持 ACID）      |+---------------------------+

3. 方案目标

  • 实时采集 MySQL 数据（基于 Binlog）

  • 支持变更（Insert/Update/Delete）语义

  • 数据存入 Hudi 表（支持 MOR/COW 格式）

  • Hive/Presto 端可直接查询

4. 组件版本建议

组件版本建议
Flink1.17.x 或 1.18.x
Flink CDC2.4.1
Debezium内嵌于 Flink CDC
Hudi0.13.1+
Hive2.3.x / 3.1.x
Hadoop/HDFS3.x

5. 部署准备

1）安装 Kafka（可选）

用于做 CDC 中转（可选，支持 Flink 直接接 Hudi）

2）安装 Hive Metastore + Hadoop HDFS

用于管理 Hudi 表元数据和 HDFS 存储

3）准备 MySQL 源数据库

配置 binlog，设置 binlog_format = ROW，并开启 server_id、binlog_row_image = full

6. 关键配置代码与步骤

1）添加 Maven 依赖

<dependencies><!-- Flink CDC --><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.4.1</version></dependency><!-- Hudi Sink --><dependency><groupId>org.apache.hudi</groupId><artifactId>hudi-flink-bundle_2.12</artifactId><version>0.13.1</version></dependency>
</dependencies>

2）Flink SQL 示例（CDC → Hudi）

-- 1. 源表：MySQL CDC 表
CREATE TABLE ods_orders (id STRING,user_id STRING,amount DOUBLE,ts TIMESTAMP(3),PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc','hostname' = 'mysql-host','port' = '3306','username' = 'flink','password' = 'flink123','database-name' = 'srm','table-name' = 'orders','scan.startup.mode' = 'initial'
);-- 2. 目标表：Hudi 表（MOR 模式）
CREATE TABLE dwd_orders (id STRING PRIMARY KEY NOT ENFORCED,user_id STRING,amount DOUBLE,ts TIMESTAMP(3)
) PARTITIONED BY (`user_id`)
WITH ('connector' = 'hudi','path' = 'hdfs://namenode/data/hudi/dwd_orders','table.type' = 'MERGE_ON_READ','hoodie.datasource.write.recordkey.field' = 'id','write.tasks' = '4','compaction.async.enabled' = 'true','hive_sync.enabled' = 'true','hive_sync.mode' = 'hms','hive_sync.metastore.uris' = 'thrift://hive-metastore:9083','hive_sync.db' = 'ods','hive_sync.table' = 'dwd_orders'
);-- 3. 实时写入
INSERT INTO dwd_orders
SELECT * FROM ods_orders;

7. 关键功能说明

功能配置字段说明
主键变更支持PRIMARY KEY ... NOT ENFORCED支持 upsert
增量采集模式scan.startup.mode = initial首次全量 + 后续增量
实时 compactioncompaction.async.enabled = trueMOR 表性能保障
Hive 数据同步hive_sync.enabled = trueHudi 自动注册 Hive 元数据

8. 整合优化建议

1）多表 CDC 同步统一处理

使用 Flink CDC 的 schema-name.table-name通配符：

'database-name' = 'srm',
'table-name' = '.*',

配合 Flink SQL Catalog + Dynamic Table Factory，可实现一拖 N 的多表处理逻辑。

2）增加清洗逻辑（如空值过滤、转换）

SELECTid,user_id,amount * 1.13 AS amount_tax,ts
FROM ods_orders
WHERE amount IS NOT NULL;

3）写入 Kafka（替代 Hudi） → 用于事件总线或下游消费

CREATE TABLE kafka_sink (id STRING,user_id STRING,amount DOUBLE,ts TIMESTAMP(3)
) WITH ('connector' = 'kafka','topic' = 'ods.orders','properties.bootstrap.servers' = 'kafka:9092','format' = 'json','scan.startup.mode' = 'latest-offset'
);

9. Flink CDC 整合场景汇总

场景描述推荐组件
实时数据入湖MySQL → HudiFlink CDC + Hudi
数据仓库加速Oracle → IcebergFlink CDC + Iceberg
数据中台构建MySQL → Kafka → 多下游Flink CDC + Kafka
数据回流校验Kafka → Flink → MySQLFlink SQL + JDBC Sink
DWD建模ODS → DWD/DWM → ADSFlink SQL + 维表 JOIN

10. 可视化监控

工具功能
Flink UICheckpoint、Watermark、吞吐
Prometheus指标采集
Grafana监控仪表盘
HiveSQL 查询验证

四、自定义 Flink CDC Job 的完整实现

自定义 Flink CDC Job 的完整实现，采用 Java DataStream API 编写，支持：

  • 多表接入（MySQL 为例）

  • 自定义清洗、转换逻辑

  • 支持写入 Kafka、Hudi、Iceberg 等下游系统

  • 可部署为标准 Flink 应用（flink run 执行）

1. 自定义 Flink CDC Job 场景说明

目标：

  • 从 MySQL 采集订单表 srm.orders

  • 做清洗（如金额换算、字段过滤）

  • 输出到 Hudi 表（或 Kafka/Console）

2. 依赖配置（Maven）

<dependencies><!-- Flink CDC --><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.4.1</version></dependency><!-- Flink 通用 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>1.17.2</version></dependency><!-- 可选：Sink 依赖，如 Kafka、Hudi、Iceberg -->
</dependencies>

3. 完整代码示例：CustomCdcJob.java

public class CustomCdcJob {public static void main(String[] args) throws Exception {// 1. 创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 2. 配置 CDC 源：MySQLMySqlSource<Order> mysqlSource = MySqlSource.<Order>builder().hostname("mysql-host").port(3306).databaseList("srm").tableList("srm.orders").username("flink").password("flink123").deserializer(new OrderDeserializationSchema()) // 自定义反序列化.build();// 3. 接入 SourceDataStreamSource<Order> orderStream = env.fromSource(mysqlSource,WatermarkStrategy.noWatermarks(),"MySQL CDC Source");// 4. 数据清洗/转换SingleOutputStreamOperator<Order> cleaned = orderStream.filter(order -> order.amount > 0).map(order -> {order.amount = order.amount * 1.13; // 加税return order;});// 5. Sink：控制台 / Kafka / Hudicleaned.print();env.execute("Custom Flink CDC Job");}
}

4. 自定义反序列化器：OrderDeserializationSchema

public class OrderDeserializationSchema implements DebeziumDeserializationSchema<Order> {@Overridepublic void deserialize(SourceRecord sourceRecord, Collector<Order> collector) {Struct value = (Struct) sourceRecord.value();if (value == null) return;Struct after = value.getStruct("after");if (after != null) {Order order = new Order();order.id = after.getString("id");order.userId = after.getString("user_id");order.amount = after.getFloat64("amount");order.ts = Instant.ofEpochMilli(after.getInt64("ts")).atZone(ZoneId.of("UTC")).toLocalDateTime();collector.collect(order);}}@Overridepublic TypeInformation<Order> getProducedType() {return TypeInformation.of(Order.class);}
}

5. 定义 POJO 类：Order.java

public class Order implements Serializable {public String id;public String userId;public Double amount;public LocalDateTime ts;@Overridepublic String toString() {return String.format("[Order] id=%s, user=%s, amt=%.2f, ts=%s",id, userId, amount, ts.toString());}
}

6. Sink 可选方案

1）控制台输出（开发调试）

cleaned.print();

2）Kafka Sink（事件总线）

KafkaSink<String> kafkaSink = KafkaSink.<String>builder().setBootstrapServers("kafka:9092").setRecordSerializer(KafkaRecordSerializationSchema.builder().setTopic("srm.orders.cdc").setValueSerializationSchema(new SimpleStringSchema()).build()).build();cleaned.map(order -> JSON.toJSONString(order)).sinkTo(kafkaSink);

3）写入 Hudi 表（通过 Flink Hudi Sink）

cleaned.addSink(HudiSinkUtil.getSink());

自定义 Hudi Sink 工具类可基于 HoodieSink 封装。

七、打包部署方式

1）使用 maven-shade-plugin 打 fat-jar：

mvn clean package -DskipTests

输出：custom-cdc-job-1.0-SNAPSHOT.jar

2）提交到 Flink 集群

flink run -m yarn-cluster -c com.example.CustomCdcJob custom-cdc-job.jar

8. 扩展功能（可选）

功能实现方式
多表同步.tableList("srm.orders,srm.invoice")
动态 schema 推导使用 JsonDebeziumDeserializationSchema
维表 joinFlink SQL / Broadcast Join
自定义状态存储Flink KeyedState
exactly-once 写入 Kafka/Hudi使用 checkpoint 支持

http://www.xdnf.cn/news/7865.html

相关文章：

  • 手机充电协议
  • 目标检测135个前沿算法模型汇总（附源码）！
  • rocketmq优先级控制 + 并发度控制
  • 85本适合AI入门的人工智能书籍合集免费资源
  • 游戏引擎学习第301天:使用精灵边界进行排序
  • 数据湖和数据仓库的区别
  • 线程、线程池、异步
  • 人脸识别，使用 deepface + api + flask, 改写 + 调试
  • 【沉浸式求职学习day46】【华为5.7暑期机试题目讲解】
  • 广东省省考备考（第十六天5.21）—言语：语句排序题（听课后强化）
  • Mcu_Bsdiff_Upgrade
  • 数据结构与算法——堆
  • ThreadPoolTaskExecutor 和 ThreadPoolExecutor 的使用场景
  • (vue)前端实现下载后端提供的URL文件
  • 设计模式1 ——单例模式
  • 前后端的双精度浮点数精度不一致问题解决方案，自定义Spring的消息转换器处理JSON转换
  • LeetCode117_填充每个结点的下一个右侧结点指针Ⅱ
  • WPS深度适配鸿蒙电脑折叠形态，国产替代下的未来何在？
  • L53.【LeetCode题解】二分法习题集2
  • 关于收集 Android Telephony 网络信息的设计思考2
  • WinForms 应用中集成 OpenCvSharp 实现基础图像处理
  • 基于AI大语言模型的历史文献分析在气候与灾害重建中的技术-以海南岛千年台风序列重建为例
  • C++初阶-vector的模拟实现2
  • 前端（小程序）学习笔记（CLASS 1）：组件
  • 强化学习入门：RL开发框架Gym简介
  • App 出海：全渠道营销如何通过性能监控与精准归因实现增长
  • 【209. 长度最小的子数组】
  • shell脚本之函数详细解释及运用
  • 【深度估计 Depth Estimation】数据集介绍
  • [Java实战]Spring Boot整合Seata：分布式事务一致性解决方案（三十一）