Data Migration — 本质、方法与数据传输过程

Q: Data Migration 使用哪些工具？

主要工具：AWS DMS、Azure Data Factory、用于 CDC 的 Debezium、用于模式迁移的 Flyway 和 Liquibase、用于 ETL 管道的 Apache NiFi。选择取决于源类型和目标平台。

Q: 如果迁移后数据丢失了怎么办？

立即停止向接收系统写入，根据回滚计划切换到源，并从备份中恢复数据。在分析失败原因后，重复测试迁移。回滚计划必须在开始前准备就绪。

作者： IT Sectr 发布日期: 2026-06-14 阅读时间: 10 分钟

Data Migration 是在存储系统、格式或软件版本之间传输数据的过程。在应用程序开发中，数据迁移在更新数据库、更换提供商或过渡到新的存储架构时是必需的。根据 Gartner（2025）的数据，60% 的数据迁移项目因测试不足和缺乏回滚策略而超出预算。规划良好的迁移可最大限度地减少停机时间并消除数据丢失。

要点

Data Migration — 在系统之间传输数据，保持完整性和可用性。
ETL 过程（Extract, Transform, Load）— 任何数据迁移的基本模型。
Big Bang — 在短停机窗口内一次性迁移。
Trickle — 无需停止系统的持续同步。
回滚 — 发生故障时返回初始状态的强制性计划。

什么是 Data Migration

Data Migration 是将数据从一个源传输到另一个源的过程，确保数据在完成后保持完整性、一致性和可用性。与简单复制不同，迁移包括格式转换、重复数据清理、参照完整性检查和结果验证。

数据迁移的需求出现在更新 DBMS（例如从 MySQL 5.7 到 MySQL 8.0）、更换云提供商、从单体架构迁移到微服务或从 schema 过渡到 NoSQL 时。根据 Stripe（2024）的数据，89% 的公司至少每两年遇到一次数据迁移，43% 的公司认为这是技术升级中最困难的阶段。

数据迁移的主要目标

第一个目标 — 通过过渡到更现代的存储解决方案来提高性能。第二个 — 通过更换基础设施提供商来降低运营成本。第三个 — 当数据必须存储在特定司法管辖区时，确保符合监管要求（GDPR、152-FZ）。

迁移与集成的区别

集成假设两个运行系统之间进行永久同步。迁移 — 一次性传输，随后关闭源。集成不会删除源中的数据，迁移则以将接收系统切换到 primary 状态而结束。这一根本性区别决定了工具和验证方法的选择。

数据迁移的 ETL 模型

任何 Data Migration 的基本架构都基于 ETL（Extract, Transform, Load）模型。Extract — 从源中提取数据。Transform — 转换为目标模式。Load — 加载到接收器中。每个阶段都有自己的质量控制方法。

Extract：数据提取

在提取阶段，数据从源数据库、文件存储或 API 中读取。增量导出（CDC — Change Data Capture）允许仅传输已更改的记录，从而减少流量。完整转储适用于小数据量，但对于 TB 级数据库，通过 Debezium 或 Kafka Connect 进行流复制是更优选择。

Transform：模式转换

转换包括重命名列、更改数据类型、标准化值和聚合。例如，在从 MySQL 迁移到 PostgreSQL 时，数字类型 DECIMAL 必须转换为 NUMERIC，日期格式转换为 ISO 8601。根据 Talend（2024）的数据，70% 的迁移时间花费在转换上，而不是传输上。

Load：加载到接收器

加载以批处理（batch insert）或流方式执行。为了消除重复，使用幂等性键。加载后必须进行验证步骤：比较记录数、计算校验和、检查业务规则。没有验证，Data Migration 被视为未完成。

数据迁移策略

Data Migration 策略的选择决定了系统停机时间、回滚复杂性和准备工作量。主要策略 — Big Bang、Trickle 和并行运行（Parallel Run）。每种策略适用于不同的场景。

策略	停机时间	复杂性	丢失风险
Big Bang	小时-天	低	高
Trickle	分钟	高	低
Parallel Run	无	非常高	最小

Big Bang 迁移

Big Bang — 一次性关闭旧系统、传输数据并启动新系统。适用于小数据量和简单模式。风险：如果发生故障，系统在从备份完全恢复之前不可用。2024 年，GitLab 使用 Big Bang 将 5 TB 数据从 AWS RDS 迁移到 GCP Cloud SQL，停机窗口为 14 小时。

Trickle 迁移

Trickle — 以小批量持续同步。旧系统和新系统并行运行，更改实时复制。数据稳定后，旧源被关闭。这种方法需要双向同步和冲突解决。在 Continuous Delivery 中用于在不停机的情况下更新数据库模式。

Parallel Run

Parallel Run — 两个系统同时运行，应用程序从两个源写入和读取。在接收器中验证数据后，旧源被关闭。这是最安全但也是最昂贵的策略 — 需要维护两个基础设施。适用于关键金融系统的迁移。

数据迁移类型

在应用程序开发中，根据传输对象和上下文区分多种 Data Migration 类型。每种类型都有自己的方法论和工具。理解类型是选择正确策略的第一步。

数据库迁移（Database Migration）

Database Migration — 在不同供应商的 DBMS 之间传输：从 Oracle 到 PostgreSQL，从 SQL Server 到 MySQL，从 MongoDB 到 DynamoDB。复杂性在于数据类型、SQL 方言和索引机制的不兼容性。工具：AWS DMS、Debezium、Liquibase。

应用程序数据迁移（Application Data Migration）

同一应用程序不同版本之间的数据传输 — 例如，在更新代码并更改实体结构时。通常伴随着以应用程序语言执行迁移脚本：Ruby on Rails 中的 Active Record Migrations、Java 的 Flyway、.NET 中的 Entity Framework Migrations。这些脚本依次转换模式和数据。

云迁移（Cloud Data Migration）

Cloud Data Migration — 从本地基础设施到云或云之间的数据传输。AWS Snowball、Azure Data Box 和 Google Transfer Appliance 用于 TB 级数据的物理运输。对于在线迁移，使用 VPN 隧道和复制。根据 Gartner 的数据，到 2027 年，70% 的迁移将在混合云环境中执行。

数据迁移规划

没有计划的数据迁移 — 必然失败。规划包括当前模式审计、数据剖析、策略选择、环境准备、测试和回滚计划批准。根据 McKinsey（2024）的研究，54% 的失败迁移是由于缺乏正式计划造成的。

审计和数据剖析

在迁移之前，必须对源系统进行审计：确定数据量、表数量、实体之间的依赖关系、可空字段类型和重复项的存在。数据剖析可发现异常 — 关键字段中的 NULL 值、格式不匹配、损坏的引用。这些数据构成了完整转储的基线。

测试环境准备

测试迁移在主运行之前对生产数据的副本进行。目的 — 检查 ETL 管道的性能、转换的正确性和加载速度。建议在 Big Bang 之前至少进行三个完整周期的测试。每个周期包括完整传输、验证和回滚。

回滚计划

回滚 — 当检测到关键错误时返回到原始系统。回滚计划包括：开始前源的完整备份、模式恢复脚本、启动旧系统的分步说明以及与用户的沟通计划。未经批准的回滚计划，Data Migration 不应在生产中启动。

数据迁移代码示例

让我们看一个 Kotlin 中与 Flyway 结合使用的 Data Migration 实际示例。V1 迁移脚本创建 users 表并从旧格式传输数据。Flyway 自动跟踪应用的迁移并保证幂等性。

kotlin

// V1__Migrate_Users.kt — 从旧格式进行数据迁移
import org.flywaydb.core.api.migration.BaseJavaMigration
import java.sql.Connection
import java.sql.PreparedStatement

class V1__MigrateUsers : BaseJavaMigration() {
    override fun migrate(connection: Connection) {
        val legacyUsers = connection.prepareStatement(
            "SELECT id, name, legacy_role FROM users_legacy"
        ).executeQuery()

        val insertStmt: PreparedStatement = connection.prepareStatement(
            "INSERT INTO users (id, name, role, migrated_at) VALUES (?, ?, ?, NOW())"
        )

        while (legacyUsers.next()) {
            insertStmt.setInt(1, legacyUsers.getInt("id"))
            insertStmt.setString(2, legacyUsers.getString("name"))
            val role = mapLegacyRole(legacyUsers.getString("legacy_role"))
            insertStmt.setString(3, role)
            insertStmt.executeUpdate()
        }
    }
}

使用 SQLAlchemy 在 Python 中将数据从 CSV 迁移到 PostgreSQL 的示例。脚本执行从文件中提取、转换类型并加载目标表。

python

# migrate_data.py — 将 CSV 加载到 PostgreSQL 并转换
import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql://user:pass@host/db")
df = pd.read_csv("legacy_orders.csv")

df["order_date"] = pd.to_datetime(df["order_date"])
df["amount"] = df["amount"].astype("float")
df.to_sql("orders", engine, if_exists="append", index=False)

print("Data Migration 完成")