数据迁移步骤清单

以下为从 MySQL 迁移数据到 PostgreSQL 的标准步骤与要点，覆盖评估、架构转换、全量/增量迁移、验证与切换。可按需裁剪为离线一次性迁移或零停机迁移。

一、评估与规划
- 清点与分析
  - 库与表数量、数据量、最大表行数与体积、增长率。
  - 长事务、热点表、触发器/事件、存储过程、视图、外键、全文索引、分区等。
  - 字符集与排序规则（MySQL: utf8mb4/latin1 与 collation；PostgreSQL: cluster 初始化时确定编码/排序）。
- 版本与兼容性
  - MySQL 5.7/8.0 到 PostgreSQL 12+（建议 14+）。
  - 工具支持（pgloader、AWS DMS、pg_chameleon、Debezium + Kafka Connect）。
- 性能与窗口
  - 是否需要零停机（CDC 同步），可接受的只读或停机窗口。
  - 目标库资源与网络带宽评估，批量加载参数规划。
- 安全与网络
  - 凭据、VPC/防火墙、最小权限账户，TLS。

二、目标 PostgreSQL 准备
- 初始化数据库
  - 创建数据库与 schema，编码 UTF8，选择合适的排序规则与 ICU（如需稳定排序/多语言需求）。
  - 建议参数：shared_buffers、work_mem、maintenance_work_mem、wal_compression、checkpoint_timeout 等按数据量调整（大表加载时提高 maintenance_work_mem）。
- 扩展
  - 依据功能启用：uuid-ossp 或 pgcrypto（UUID），citext（不区分大小写文本），postgis（空间数据），pg_trgm（模糊搜索）。
- 账户与权限
  - 迁移专用用户（仅所需权限），业务用户与角色分离。

三、架构转换（DDL 层）
- 数据类型与语义映射（示例）
  - TINYINT(1) → boolean
  - 其他 TINYINT/SMALLINT/INT → smallint/integer/bigint（注意 UNSIGNED → 上推一档或加 CHECK 约束）
  - DECIMAL/NUMERIC → numeric(p,s)
  - FLOAT/DOUBLE → double precision（注意精度差异）
  - CHAR/VARCHAR/TEXT → text/varchar（PostgreSQL 无长度显示宽度语义）
  - BINARY/VARBINARY/BLOB → bytea
  - DATE/DATETIME/TIMESTAMP → date/timestamp/timestamptz（如存储 UTC，建议 timestamptz）
  - JSON → jsonb（功能更强，索引支持更好）
  - ENUM → PostgreSQL enum 类型或 text + CHECK（推荐 enum 或 domain 以维持约束）
  - SET → text[]（数组），或分离多对多关联
  - GEOMETRY → PostGIS 几何类型
- 约束与索引
  - 主键/唯一约束、外键、检查约束迁移；MySQL 隐式索引与前缀索引需要重设策略。
  - FULLTEXT 索引 → tsvector + GIN/GIST；需构造 to_tsvector 列或表达式索引。
- 自增/默认值
  - AUTO_INCREMENT → GENERATED BY DEFAULT AS IDENTITY（或 legacy SERIAL）。
  - MySQL 的 ON UPDATE CURRENT_TIMESTAMP → 在 PostgreSQL 用触发器维护 updated_at（默认值不会在更新时自动刷新）。
- 语义差异处理
  - 0 日期（'0000-00-00'、'0000-00-00 00:00:00'）在 PostgreSQL 非法：需转 NULL 或哨兵值。
  - 区分大小写与保留字：统一小写蛇形命名，避免引用标识符。
  - 排序/比较：MySQL 不同 collation 的大小写比较行为与 PostgreSQL 不同，必要时用 citext 或定义一致 collation。
  - 布尔型：MySQL 用 0/1，迁移为 boolean 后注意应用侧参数化。

建议使用工具自动生成/转换 DDL，并进行人工审校与定制。

四、生成与部署目标 DDL
- 方案 A：使用 pgloader 由源库反射创建目标表
  - 优点：自动化；缺点：复杂场景需自定义 CAST 与后置修正。
- 方案 B：手工整理 DDL
  - 使用脚本/模板统一命名、类型与约束策略，更可控。

建议先在空库执行 DDL，暂不创建外键/次级索引（或以 NOT VALID 形式延后），以便加速后续数据加载。

五、全量数据迁移（初次装载）
- 工具选择
  - pgloader：MySQL → PostgreSQL 一体化（复制时用 COPY，加速，带类型映射与零日期处理）。
  - 或导出为 CSV 并在 PostgreSQL 用 COPY 导入（需自行处理转义与类型转换）。
- pgloader 示例（简化）
  - 配置文件示例：
    LOAD DATABASE
      FROM mysql://user:pass@mysql-host:3306/src_db
      INTO postgresql://user:pass@pg-host:5432/dst_db
    WITH
      include drop, create tables, create indexes, reset sequences,
      workers = 8, batch rows = 50000, prefetch rows = 50000,
      on error stop
    SET
      maintenance_work_mem to '2GB',
      work_mem to '128MB',
      synchronous_commit to 'off'
    CAST
      type datetime to timestamptz,
      type date using zero-dates-to-null,
      type json to jsonb
    BEFORE LOAD DO
      $$ SET session_replication_role = 'replica'; $$;
    AFTER LOAD DO
      $$ SET session_replication_role = 'origin'; $$;
  - 说明：
    - 根据需要增加对 tinyint(1)→boolean、enum→enum/text 的映射规则。
    - session_replication_role=replica 会禁用用户触发器与外键触发器，加速导入但会跳过外键校验，需在导入后补做一致性校验与再验证。
- CSV + COPY 流程（备选）
  - mysqldump/SELECT ... INTO OUTFILE 导出 CSV（注意字符集统一为 UTF-8，NULL 表示，转义与分隔符）。
  - 在 PostgreSQL 使用 COPY 表加载（单事务，synchronous_commit=off，增大 maintenance_work_mem）。
- 性能建议
  - 暂不创建或临时禁用二级索引与外键，待数据装载后再建。
  - 批量加载完成后 VACUUM ANALYZE 以刷新统计信息。

六、增量同步/变更数据捕获（CDC，零停机可选）
- 方案 A：pg_chameleon
  - 使用 MySQL binlog 行级复制到 PostgreSQL，适合 MySQL→PG 单向迁移。
  - 要求 MySQL 开启 binlog，binlog_format=ROW，binlog_row_image=FULL，设置唯一 server_id。
- 方案 B：Debezium MySQL Connector + Kafka Connect JDBC Sink
  - Debezium 从 MySQL binlog 捕获变更，Kafka Connect 下游写入 PostgreSQL。
  - 适合高吞吐与可观测性需求；需 Kafka 基础设施与 schema/类型映射规则定制。
- 方案 C：AWS DMS
  - 支持全量 + CDC，托管化，适合云上场景。
- CDC 步骤
  1) 在 MySQL 启用 binlog（ROW 模式），创建最小权限复制账号。
  2) 全量装载完成后开启 CDC 订阅，持续追赶增量。
  3) 监控延迟，直至延迟趋近于 0，为切换做准备。
  4) 切换窗口将源库置只读，等待 CDC 清空剩余增量。

七、重建约束与索引、修复序列
- 索引
  - 为大表并发创建索引：CREATE INDEX CONCURRENTLY，减少锁阻塞。
- 外键
  - 可先 ADD CONSTRAINT ... NOT VALID，再 VALIDATE CONSTRAINT，降低写阻塞（注意验证期间仍有共享锁，但不阻塞读）。
- 序列/标识列
  - 如使用 SERIAL：SELECT setval(pg_get_serial_sequence('tbl','id'), (SELECT max(id) FROM tbl), true);
  - 使用 IDENTITY 一般无需手动对齐，但建议验证。
- 触发器
  - 如需模拟 MySQL 的 ON UPDATE 行为，为 updated_at 添加 BEFORE UPDATE 触发器。

八、数据验证与质量校验
- 数量级校验
  - 按库/表/分区比对行数，按条件分片比对。
- 取样与校验和
  - 关键表按主键范围抽样，对比字段哈希/聚合（注意不同类型/时区规范化）。
- 约束与一致性
  - VALIDATE CONSTRAINT 后确认无违例；业务关键唯一性核验。
- 性能基线
  - 收集并比较典型查询的执行计划与耗时，必要时补充索引或调整统计目标（ALTER TABLE ... SET (autovacuum_analyze_scale_factor/targets)）。

九、切换与回滚
- 切换
  - 冻结源库写入（短暂只读）、等待 CDC 清空、停止连接源、切换应用连接到 PostgreSQL。
  - 清理迁移专用权限与参数，开启监控与告警。
- 回滚预案
  - 切换前保留源库只读与快照；在明确验证期内可快速回切。
  - 记录所有变更脚本与参数，确保可重复迁移。

十、常见问题与处理建议
- 0000-00-00 日期：迁移前清洗或在工具层映射为 NULL。
- UNSIGNED 溢出风险：映射到更大整型或添加 CHECK (col >= 0)。
- 字符集/表情符：MySQL utf8 与 utf8mb4 差异，确保源端为 utf8mb4，目标端为 UTF8，并验证四字节字符。
- 全文索引差异：替换为 tsvector + GIN，并按语言选择合适的配置（simple/english/zhparser）。
- JSON 语义：迁移为 jsonb 并替换函数/运算符（->、->>、@>、索引表达式）与应用层查询。
- 时间戳与时区：统一存

以下步骤面向“Kafka 集群 A 迁移到 Kafka 集群 B”的场景，覆盖规划、工具选择、在线镜像、位点迁移与业务切换。默认优先使用开源 MirrorMaker 2（MM2）。如使用 Confluent Platform/Cloud，可用 Cluster Linking，步骤类似但命令不同。

一、迁移前评估与规划
- 明确迁移目标与约束：是否零停机、是否保留消费者位点、能否短暂停写、允许的重放窗口。
- 盘点与基线：
  - 主题清单：名称、分区数、副本因子、配置（cleanup.policy、retention.ms、min.insync.replicas、compression.type 等）。
  - 流量画像：平均/峰值 TPS、字节流量、最大记录大小、滞后容忍。
  - 客户端与消费者组：语言/库版本、序列化方式（Avro/Protobuf/JSON Schema）、是否使用事务或 EOS。
- 目标集群容量与配置：Broker 数量、磁盘/网络、JVM、同等或更高的 RF、分区上限、消息大小限制。
- 网络与安全：
  - 连通性与带宽：开放 9092/9093 等端口，跨机房/云网络打通。
  - TLS/SASL：证书、SASL 机制（SCRAM/GSSAPI/OAUTHBEARER），账号/ACL 最小权限。
- Schema 与序列化：
  - 使用 Schema Registry 的，先迁移 Schema（导出/导入），确保目标兼容策略一致（BACKWARD/FORWARD/FULL）。
- 主题映射与命名策略：
  - 保持同名（IdentityReplicationPolicy）或加前缀（默认 {源}.{topic}）。建议使用同名，便于切换。
- 排除内部主题：不要迁移 __consumer_offsets、__transaction_state、_schemas 等内部主题。
- 切换与回滚预案：明确 T+0 切流步骤与失败回退路径（恢复到源集群消费/写入）。

二、方案选择
- 在线实时：MirrorMaker 2（开源，Kafka Connect 架构；可镜像主题、ACL、配置、消费者位点）。
- 在线实时（Confluent）：Cluster Linking（跨集群零拷贝镜像，受制于版本/许可）。
- 批量+增量：Connect Sink 导出到对象存储（S3/HDFS/OSS）后再回放，适用于离线大体量+可停机或双写场景。
- 自研消费/生产：适合复杂改造或强定制，但需自行处理顺序、重复与位点。

三、MirrorMaker 2 在线迁移步骤（推荐开源）
1) 准备运行环境
- 部署 Kafka Connect 分布式集群（2 台以上，JVM 与网络靠近源或目标集群），为 MM2 提供运行时。
- 确保 Connect 到两端集群的安全与网络连通，准备凭据与 ACL。

2) 准备 MM2 配置文件（示例）
- 使用 Kafka 自带的 connect-mirror-maker.sh 启动，单文件配置即可。
- 参考配置片段（关键项，按需修改）：
  - clusters = A, B
  - source.cluster.alias = A
  - target.cluster.alias = B
  - A.bootstrap.servers = <A-bootstrap>
  - B.bootstrap.servers = <B-bootstrap>
  - A.security.protocol = SASL_SSL
  - A.sasl.mechanism = SCRAM-SHA-512
  - A.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="userA" password="***";
  - B.security.protocol = SASL_SSL
  - B.sasl.mechanism = SCRAM-SHA-512
  - B.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="userB" password="***";
  - topics = .*
  - groups = .*
  - replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy
  - refresh.topics.enabled = true
  - refresh.topics.interval.seconds = 60
  - sync.topic.configs.enabled = true
  - sync.topic.acls.enabled = false
  - emit.checkpoints.interval.seconds = 10
  - replication.factor = 3
  - producer.acks = all
  - producer.enable.idempotence = true
  - producer.compression.type = lz4
  - consumer.max.partition.fetch.bytes = 10485760
  - consumer.fetch.max.bytes = 52428800
说明：
- IdentityReplicationPolicy 保持同名主题，易于切换。
- sync.topic.configs.enabled=true 会将源主题配置同步至目标（不包括内部主题）。
- 通过 producer.* 和 consumer.* 传递 Connect 内部生产者/消费者参数。
- 不建议同步 ACL 到生产目标，通常按最小权限独立设置。

3) 启动 MM2
- 命令：bin/connect-mirror-maker.sh -clusters <mm2.properties>
- 确认三个子任务启动：MirrorSourceConnector、MirrorHeartbeatConnector、MirrorCheckpointConnector。
- 观察日志无报错，目标集群自动创建镜像主题（分区数默认与源一致，必要时在配置中固定 RF）。

4) 监控与限速
- 监控指标：端到端延迟（record-age-ms/replication-latency-ms）、任务失败、重启次数、目标端入流量。
- 控制速率：
  - 通过 Kafka Broker/用户/客户端配额限速（字节/秒、请求/秒）。
  - 调整 consumer.fetch/producer.batch.size/linger.ms 以平滑带宽。
- 验证数据：抽样比对分区末端位移、消息计数或基于业务键的去重校验。

5) 迁移消费者位点（offset）
- MirrorCheckpointConnector 会在目标集群生成 checkpoint 信息，用于从源位点映射到目标位点。
- 建议流程：
  - 在业务切换窗口暂停源端消费者，等待 MM2 把最新数据复制到位（目标端主题对比滞后≈0）。
  - 使用 MM2 提供的 Offset 翻译能力（org.apache.kafka.connect.mirror.tools.MirrorClient 或等价 API）将源组位点映射为目标组位点。
  - 将映射结果应用到目标集群的消费者组（可用管理 API 或 kafka-consumer-groups 的重置位点功能，按目标分区写入具体 offset）。
说明：
- MM2 能保留分区内顺序，但不能保留源端事务边界；目标端为普通写入。
- 位点迁移要在停消费或只读窗口内进行，确保不丢不重。

6) 业务切换（Cutover）
- 方案一（短暂停写，强一致）：
  - 暂停源端生产者写入。
  - 等待 MM2 复制至零滞后。
  - 迁移消费者位点至目标。
  - 切换生产者与消费者到目标集群继续运行。
- 方案二（双写+回放，近零停机）：
  - 应用端短期双写 A 和 B。
  - 等待 B 补齐历史，切消费者到 B。
  - 停止对 A 的写入，观察稳定后停止双写。
- 切换后观察一段时间（至少一个高峰周期）再停止 MM2。

7) 收尾与清理
- 关闭 MM2，收回临时 ACL/账号。
- 销毁或降配源集群的主题（按保留策略）。
- 完成文档归档与监控规则切换。

四、Cluster Linking（如使用 Confluent Platform/Cloud）
- 在目标集群创建指向源集群的 Link（需具备合适的版本与许可）。
- 通过 Link 选择要镜像的主题，配置保留、前缀/同名策略及 ACL 同步策略。
- 监控 Link Lag，完成位点迁移与切流流程（与上面类似）。
说明：Cluster Linking 是 Confluent 功能，非 Apache Kafka 开源特性。

五、批量导出/导入（离线+增量）
- 使用 Kafka Connect Sink（S3/HDFS/File）从源集群批量导出历史数据。
- 使用相应 Source 连接器将历史回放到目标集群。
- 同时以 MM2 或应用双写方式承接增量数据，再在切换窗口合并位点并切流。
- 适用于大规模历史数据与可接受一定延迟的场景。

六、校验与常见注意事项
- 校验项：
  - 分区与副本：目标主题分区数与源一致（或按规划变更），RF 与 ISR 满足 min.insync.replicas。
  - 末端位移与滞后：逐主题逐分区对比 end offset，滞后接近 0 再切流。
  - 采样消息校验：按 key 抽样比对 checksum/业务字段。
- 注意事项：
  - 不迁移内部主题；事务在目标端不会保留原事务语义。
  - 确认消息大小限制（message.max.bytes、max.message.bytes）在目标端不更小。
  - Schema 先行迁移并保证兼容模式一致；避免直接镜像 _schemas。
  - 跨地域带宽受限时务必开启压缩并设置配额，防止对生产流量造成抢占。
  - 版本兼容：MM2 作为客户端对接两端，一般无需 broker 同版本，但要确保客户端与 broker 兼容矩阵成立。

七、最小可执行清单（MM2）
- 打通网络与安全；创建目标集群账号与 ACL。
- 准备并审核 MM2 配置（同名策略、主题/组正则、RF、压缩、限速）。
- 启动 MM2，观察 24–48 小时，确认稳定与滞后可控。
- 计划切流窗口：停消费/短暂停写，位点翻译与应用。
- 切换生产者与消费者至目标；观察验证。
- 停止 MM2，清理与收尾。

以上步骤覆盖从 Kafka 到 Kafka 的主流迁移路径与操作要点，可按环境与合规要求裁剪执行。需要可复用的配置模板或自动化脚本时，可进一步细化到环境变量、密钥管理与 CI/CD 流程。