03. 消息与实时

目标：MySQL/PostgreSQL 的变更能通过 Flink 写入 Kafka。
本页所有资源都在 data namespace。

为什么实时链路要拆成 Kafka + Flink

源库不能直接被下游拖死：BI、湖仓、Doris 都直连业务库，查询一多就影响交易；Kafka 把变更先接住，下游按自己的节奏消费。
Flink 负责“有状态”的部分：CDC 要记录 binlog 位点、做 checkpoint、失败恢复；这是流引擎的强项，不是脚本能长期扛的。
Kafka 负责“解耦和回放”：下游 Iceberg 写入失败，可以重置 offset 重放；源库 binlog 不会为下游保留很久。
代价：多一套消息系统要运维，分区/副本/磁盘都要管；所以测试环境只保留 3 broker，生产要按峰值估。

1. Kafka 3 broker（KRaft）

为什么用 KRaft 而不是 ZooKeeper：Kafka 3.x 已经把元数据管理收进 controller quorum，少一套 ZK 就少一个故障点。
为什么是 3 broker：副本因子 3 + 最小 ISR 2，挂一台还能写；1 broker 的“3 副本”没有意义。
为什么 advertised.listeners 用 Pod IP：客户端拿到的地址必须能从别的 Pod 访问，写 0.0.0.0 会让客户端连回自己。

1.1 完整 manifest

保存为 30-kafka.yaml：

yaml

apiVersion: v1
kind: Service
metadata:
  name: kafka-headless
  namespace: data
spec:
  clusterIP: None
  publishNotReadyAddresses: true
  selector:
    app: kafka
  ports:
    - name: plaintext
      port: 9092
    - name: controller
      port: 9093
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: data
spec:
  serviceName: kafka-headless
  replicas: 3
  podManagementPolicy: Parallel
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchLabels:
                  app: kafka
              topologyKey: kubernetes.io/hostname
      containers:
        - name: kafka
          image: docker.io/apache/kafka:3.9.0
          imagePullPolicy: IfNotPresent
          securityContext:
            runAsUser: 0
          command: ["bash", "-c"]
          args:
            - |
              set -e
              NODE_ID=${HOSTNAME##*-}
              printf '%s\n' \
                "process.roles=broker,controller" \
                "node.id=${NODE_ID}" \
                "controller.quorum.voters=0@kafka-0.kafka-headless.data.svc.cluster.local:9093,1@kafka-1.kafka-headless.data.svc.cluster.local:9093,2@kafka-2.kafka-headless.data.svc.cluster.local:9093" \
                "listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://${POD_IP}:9093" \
                "advertised.listeners=PLAINTEXT://${POD_IP}:9092" \
                "controller.listener.names=CONTROLLER" \
                "listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT" \
                "inter.broker.listener.name=PLAINTEXT" \
                "log.dirs=/var/lib/kafka/data" \
                "num.partitions=3" \
                "offsets.topic.replication.factor=3" \
                "transaction.state.log.replication.factor=3" \
                "transaction.state.log.min.isr=2" \
                > /tmp/server.properties
              mkdir -p /var/lib/kafka/data
              if [ ! -f /var/lib/kafka/data/meta.properties ]; then
                /opt/kafka/bin/kafka-storage.sh format -t "$CLUSTER_ID" -c /tmp/server.properties
              fi
              exec /opt/kafka/bin/kafka-server-start.sh /tmp/server.properties
          ports:
            - containerPort: 9092
              name: plaintext
            - containerPort: 9093
              name: controller
          env:
            - name: CLUSTER_ID
              value: "MkU3OEVBNTcwNTJENDM2Qk"
            - name: POD_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
          volumeMounts:
            - name: data
              mountPath: /var/lib/kafka/data
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: ["ReadWriteOnce"]
        storageClassName: local-path
        resources:
          requests:
            storage: 20Gi

字段说明：

字段/配置	作用
`Service.clusterIP: None`	headless Service，给每个 broker 稳定 DNS
`publishNotReadyAddresses`	broker 启动前也能解析，方便 quorum 组网
`replicas: 3`	3 broker，配合 RF=3
`podManagementPolicy: Parallel`	三个 broker 同时起，不按顺序等
`podAntiAffinity`	尽量把 3 broker 分散到 3 个节点
`process.roles=broker,controller`	每个节点同时是 broker 和 controller（KRaft）
`node.id`	从 StatefulSet 序号取 0/1/2
`controller.quorum.voters`	3 个 controller 的地址，必须互相可达
`advertised.listeners`	客户端实际连的地址，必须用 Pod IP
`log.dirs`	数据落 PVC，不随 Pod 删除
`offsets.topic.replication.factor=3`	内部 offset topic 也 3 副本

关键点：

CONTROLLER 和 advertised.listeners 都用 Pod IP，不能用 0.0.0.0；
CLUSTER_ID 三台必须一致，可用 kafka-storage random-uuid 生成后写死；
反亲和确保 3 broker 分散到 3 台。

1.2 部署与验证

bash

kubectl apply -f 30-kafka.yaml
kubectl -n data get pods -l app=kafka -o wide

预期三个 broker 分别在 node1/2/3，Running。

建 topic：

bash

kubectl -n data exec kafka-0 -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-0.kafka-headless.data.svc.cluster.local:9092 \
  --create --topic cdc.orders --partitions 3 --replication-factor 3

验证：

bash

kubectl -n data exec kafka-0 -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-0.kafka-headless.data.svc.cluster.local:9092 \
  --describe --topic cdc.orders

预期 PartitionCount: 3，每个 partition 的 Isr 有 3 个 broker。

2. MySQL（CDC 源）

CDC 不是“定时查表”，而是读 binlog。
log-bin 打开变更日志，binlog-format=ROW 让 Flink 能拿到每一行的前后值；binlog-row-image=FULL 保证 UPDATE 能拿到完整新行。
没有这些配置，Flink 只能做全量快照，无法持续增量。

2.1 完整 manifest

保存为 20-mysql.yaml，替换密码：

yaml

apiVersion: v1
kind: Secret
metadata:
  name: mysql-secret
  namespace: data
type: Opaque
stringData:
  root-password: "<MYSQL_PASS>"
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: mysql-cnf
  namespace: data
data:
  my.cnf: |
    [mysqld]
    server-id=1
    log-bin=mysql-bin
    binlog-format=ROW
    binlog-row-image=FULL
    gtid-mode=ON
    enforce-gtid-consistency=ON
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: mysql-init
  namespace: data
data:
  init.sql: |
    CREATE DATABASE IF NOT EXISTS demo;
    CREATE TABLE IF NOT EXISTS demo.orders (
      id BIGINT PRIMARY KEY,
      user_id BIGINT,
      amount DECIMAL(10,2),
      status VARCHAR(20),
      updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
    );
    INSERT INTO demo.orders (id, user_id, amount, status) VALUES
      (1, 1001, 99.90, 'NEW'),
      (2, 1002, 199.00, 'PAID')
    ON DUPLICATE KEY UPDATE amount=VALUES(amount);
    CREATE USER IF NOT EXISTS 'flink'@'%' IDENTIFIED BY '<MYSQL_PASS>';
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink'@'%';
    FLUSH PRIVILEGES;
---
apiVersion: v1
kind: Service
metadata:
  name: mysql
  namespace: data
spec:
  selector:
    app: mysql
  ports:
    - name: mysql
      port: 3306
      targetPort: 3306
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: mysql
  namespace: data
spec:
  serviceName: mysql
  replicas: 1
  selector:
    matchLabels:
      app: mysql
  template:
    metadata:
      labels:
        app: mysql
    spec:
      securityContext:
        fsGroup: 999
      containers:
        - name: mysql
          image: docker.io/library/mysql:8.0
          imagePullPolicy: IfNotPresent
          env:
            - name: MYSQL_ROOT_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: mysql-secret
                  key: root-password
          ports:
            - containerPort: 3306
              name: mysql
          volumeMounts:
            - name: data
              mountPath: /var/lib/mysql
            - name: cnf
              mountPath: /etc/mysql/conf.d
            - name: init
              mountPath: /docker-entrypoint-initdb.d
      volumes:
        - name: cnf
          configMap:
            name: mysql-cnf
        - name: init
          configMap:
            name: mysql-init
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: ["ReadWriteOnce"]
        storageClassName: local-path
        resources:
          requests:
            storage: 10Gi

字段说明：

字段	作用
`mysql-secret.root-password`	root 初始密码，从 Secret 注入
`my.cnf.server-id`	MySQL 实例唯一 ID，CDC 复制拓扑靠它区分
`my.cnf.log-bin`	开启 binlog，CDC 才能读变更
`my.cnf.binlog-format=ROW`	行级变更，Flink 能拿到具体行
`my.cnf.binlog-row-image=FULL`	UPDATE 时记录完整新行
`gtid-mode/enforce-gtid-consistency`	用 GTID 定位复制位点
`mysql-init.init.sql`	首次启动建库、建表、建 `flink` 用户并授权
`securityContext.fsGroup: 999`	让 PVC 目录对 mysql 用户可写
`volumeMounts.cnf/init`	挂配置和初始化 SQL
`volumeClaimTemplates`	数据目录 10Gi PVC

2.2 部署与验证

bash

kubectl apply -f 20-mysql.yaml
kubectl -n data get pods -l app=mysql

验证 binlog：

bash

kubectl -n data exec mysql-0 -- mysql -uroot -p"$MYSQL_PASS" -e \
  "SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format';"

预期：

text

log_bin       ON
binlog_format ROW

3. PostgreSQL CDC 配置

PostgreSQL 的逻辑复制依赖 WAL。
wal_level=logical 才会记录行级变更；max_replication_slots 限制同时存在的复制槽；max_wal_senders 限制复制连接数。
这些参数要重启 PG 才生效，所以放在装 Flink 之前做。

wal_level 必须为 logical：

bash

kubectl -n data exec postgres-0 -- psql -U postgres -c \
  "ALTER SYSTEM SET wal_level='logical';"
kubectl -n data exec postgres-0 -- psql -U postgres -c \
  "ALTER SYSTEM SET max_replication_slots=10;"
kubectl -n data exec postgres-0 -- psql -U postgres -c \
  "ALTER SYSTEM SET max_wal_senders=10;"
kubectl -n data rollout restart statefulset/postgres

验证：

bash

kubectl -n data exec postgres-0 -- psql -U postgres -c \
  "SHOW wal_level; SHOW max_replication_slots;"

4. Flink 镜像

为什么把 connector 打进镜像，而不是挂宿主机目录：
hostPath 让 Pod 和某个节点绑定，换节点就缺 jar；
镜像不可变，JM/TM 拉同一个 tag，环境一致；
回滚只需要换 tag，不用登录节点改文件。
代价：改 connector 要重新 build/push，不能热改。

4.1 Dockerfile

dockerfile

FROM flink:1.20.5-scala_2.12

COPY connectors/flink-sql-connector-kafka-3.3.0-1.20.jar /opt/flink/lib/
COPY connectors/flink-sql-connector-mysql-cdc-3.2.1.jar /opt/flink/lib/
COPY connectors/flink-sql-connector-postgres-cdc-3.2.1.jar /opt/flink/lib/
COPY connectors/mysql-connector-j-8.4.0.jar /opt/flink/lib/
COPY connectors/postgresql-42.7.4.jar /opt/flink/lib/

构建并推到 Harbor（Harbor 在下一节部署；也可以先推到任意 registry）：

bash

docker build -t <REGISTRY>/flink:1.20.5-connectors .
docker push <REGISTRY>/flink:1.20.5-connectors

4.2 JM/TM

JM 管作业调度和 checkpoint 协调，TM 真正跑算子。
blob.server.port 必须固定并暴露：TM 启动时要向 JM 下载任务 jar，端口随机会让 TM 一直卡在 DEPLOYING。
taskmanager.numberOfTaskSlots 决定能并行几个 task；slot 不够时新作业会排队，不会自动扩容。

保存为 40-flink.yaml：

yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-jobmanager
  namespace: data
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink
      component: jobmanager
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
        - name: jobmanager
          image: <REGISTRY>/flink:1.20.5-connectors
          imagePullPolicy: IfNotPresent
          args: ["jobmanager"]
          env:
            - name: FLINK_PROPERTIES
              value: |
                jobmanager.rpc.address: flink-jobmanager
                jobmanager.rpc.port: 6123
                jobmanager.memory.process.size: 1600m
                blob.server.port: 6124
                blob.service.port: 6124
                taskmanager.numberOfTaskSlots: 2
                taskmanager.memory.process.size: 2048m
                parallelism.default: 2
          ports:
            - containerPort: 8081
              name: ui
            - containerPort: 6123
              name: rpc
            - containerPort: 6124
              name: blob
---
apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
  namespace: data
spec:
  selector:
    app: flink
    component: jobmanager
  ports:
    - name: ui
      port: 8081
      targetPort: 8081
    - name: rpc
      port: 6123
      targetPort: 6123
    - name: blob
      port: 6124
      targetPort: 6124
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
  namespace: data
spec:
  replicas: 2
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
        - name: taskmanager
          image: <REGISTRY>/flink:1.20.5-connectors
          imagePullPolicy: IfNotPresent
          args: ["taskmanager"]
          env:
            - name: FLINK_PROPERTIES
              value: |
                jobmanager.rpc.address: flink-jobmanager
                jobmanager.rpc.port: 6123
                blob.server.port: 6124
                blob.service.port: 6124
                taskmanager.numberOfTaskSlots: 2
                taskmanager.memory.process.size: 2048m
                parallelism.default: 2

字段说明：

字段	作用
`image`	自定义 Flink 镜像，已内置 connector
`args: ["jobmanager"]`	官方 entrypoint 的启动角色
`FLINK_PROPERTIES`	只对官方 entrypoint 生效，直接跑 `taskmanager.sh` 不读
`jobmanager.rpc.address`	TM 通过它找 JM
`blob.server.port/service.port`	固定 6124，TM 下载任务包用；不固定会卡 DEPLOYING
`taskmanager.numberOfTaskSlots`	每个 TM 的并行槽位；slot 不够作业排队
`taskmanager.memory.process.size`	TM 总内存，含 JVM/堆外
`parallelism.default`	默认并行度，作业未显式指定时用它
`service.ports`	8081 UI、6123 RPC、6124 blob，缺一不可

部署：

bash

kubectl apply -f 40-flink.yaml
kubectl -n data get pods -l app=flink

预期 JM 1/1、TM 2/2 Running。

5. CDC SQL

这一段是整条实时链路的核心：
mysql-cdc 读 binlog，先全量快照，再持续增量；
upsert-kafka 支持按主键更新/删除，普通 kafka sink 只能 append，遇到 UPDATE/DELETE 会报错；
execution.checkpointing.interval 必须开，否则 Flink 不记录 binlog 位点，重启只能从头再来。
server-id 每个 CDC 作业要唯一，否则 MySQL 会把两个作业当成同一个 slave。

5.1 MySQL → Kafka

sql

-- 每 10 秒 checkpoint 一次，Flink 会记录 binlog 位点；不开则重启后无法续传
SET 'execution.checkpointing.interval' = '10s';

-- 源表：Flink 从这里读 MySQL binlog，先全量快照再增量
CREATE TABLE mysql_orders (
  id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'mysql.data.svc.cluster.local',
  'port' = '3306',
  'username' = 'flink',
  'password' = '<MYSQL_PASS>',
  'database-name' = 'demo',
  'table-name' = 'orders',
  'server-id' = '5400-5404',
  'scan.incremental.snapshot.enabled' = 'true'
);

-- 目标表：upsert-kafka 支持按主键 update/delete，普通 kafka 只能 append
CREATE TABLE kafka_orders (
  id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'cdc.orders',
  'properties.bootstrap.servers' = 'kafka-headless.data.svc.cluster.local:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

-- 提交为长期运行的流作业；SQL client 只负责提交，作业由 JM 管理
INSERT INTO kafka_orders SELECT * FROM mysql_orders;

提交：

bash

kubectl -n data exec deploy/flink-jobmanager -- \
  /opt/flink/bin/sql-client.sh -f /opt/sql/mysql2kafka.sql

SQL 文件可以通过 ConfigMap 挂载到 /opt/sql，也可以在镜像里内置。

5.2 PostgreSQL → Kafka

sql

-- 源表：postgres-cdc 读取 WAL；slot.name 每个作业唯一
CREATE TABLE pg_orders (
  id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'postgres.data.svc.cluster.local',
  'port' = '5432',
  'username' = 'flink_pg',
  'password' = '<PG_PASS>',
  'database-name' = 'demo',
  'schema-name' = 'public',
  'table-name' = 'pg_orders',
  'slot.name' = 'flink_pg_slot',
  'decoding.plugin.name' = 'pgoutput'
);

-- 目标表：同样必须是 upsert-kafka，否则 PG 的 UPDATE/DELETE 无法写入
CREATE TABLE kafka_pg (
  id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'pg.cdc.orders',
  'properties.bootstrap.servers' = 'kafka-headless.data.svc.cluster.local:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

-- 提交 PG CDC 流作业
INSERT INTO kafka_pg SELECT * FROM pg_orders;

5.3 验证

bash

# 消费 topic
kubectl -n data exec kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-0.kafka-headless.data.svc.cluster.local:9092 \
  --topic cdc.orders --from-beginning --max-messages 5

Flink UI 或 REST 验证作业 RUNNING：

bash

kubectl -n data port-forward svc/flink-jobmanager 8081:8081
# 浏览器打开 http://localhost:8081

6. 本阶段验收

bash

kubectl -n data get pods -l app=kafka
kubectl -n data get pods -l app=mysql
kubectl -n data get pods -l app=flink
kubectl -n data exec kafka-0 -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-0.kafka-headless.data.svc.cluster.local:9092 --list

要求：

Kafka 3/3 Running；
MySQL Running，binlog=ROW；
Flink JM/TM Running；
cdc.orders、pg.cdc.orders 有数据；
Flink 作业 RUNNING。

下一步：查询与调度。

01.Mybatis

02.Spring

03.SpringMVC

04.SSM

05.MybatisPlus

06.SpringBoot

07.SpringBootWeb

08.Swagger

09.SpringCloud

10.SpringCloudAlibaba

11.Others

01.基础

02.爬虫

03. 选型决策

04. 部署实施

05. 实战与验收

5.1 实战教程

03. 消息与实时

为什么实时链路要拆成 Kafka + Flink

1. Kafka 3 broker（KRaft）

1.1 完整 manifest

1.2 部署与验证

2. MySQL（CDC 源）

2.1 完整 manifest

2.2 部署与验证

3. PostgreSQL CDC 配置

4. Flink 镜像

4.1 Dockerfile

4.2 JM/TM

5. CDC SQL

5.1 MySQL → Kafka

5.2 PostgreSQL → Kafka

5.3 验证

6. 本阶段验收

5.1 实战教程

03. 消息与实时 ​

为什么实时链路要拆成 Kafka + Flink ​

1. Kafka 3 broker（KRaft） ​

1.1 完整 manifest ​

1.2 部署与验证 ​

2. MySQL（CDC 源） ​

2.1 完整 manifest ​

2.2 部署与验证 ​

3. PostgreSQL CDC 配置 ​

4. Flink 镜像 ​

4.1 Dockerfile ​

4.2 JM/TM ​

5. CDC SQL ​

5.1 MySQL → Kafka ​

5.2 PostgreSQL → Kafka ​

5.3 验证 ​

6. 本阶段验收 ​

03. 消息与实时

为什么实时链路要拆成 Kafka + Flink

1. Kafka 3 broker（KRaft）

1.1 完整 manifest

1.2 部署与验证

2. MySQL（CDC 源）

2.1 完整 manifest

2.2 部署与验证

3. PostgreSQL CDC 配置

4. Flink 镜像

4.1 Dockerfile

4.2 JM/TM

5. CDC SQL

5.1 MySQL → Kafka

5.2 PostgreSQL → Kafka

5.3 验证

6. 本阶段验收