【系统架构设计百科】数据迁移与版本化：在线不停机的数据演进

-- MySQL 5.6+ 的在线 DDL 语法
ALTER TABLE orders ADD COLUMN shipping_address_id BIGINT, ALGORITHM=INPLACE, LOCK=NONE;

-- 扩展阶段：添加新列
ALTER TABLE users ADD COLUMN province VARCHAR(50);
ALTER TABLE users ADD COLUMN city VARCHAR(50);
ALTER TABLE users ADD COLUMN district VARCHAR(100);
ALTER TABLE users ADD COLUMN street VARCHAR(200);

def update_user_address(user_id: int, province: str, city: str, district: str, street: str):
    full_address = f"{province}{city}{district}{street}"
    db.execute("""
        UPDATE users
        SET address = %s,
            province = %s,
            city = %s,
            district = %s,
            street = %s
        WHERE id = %s
    """, (full_address, province, city, district, street, user_id))

import re
from typing import Optional, Tuple

def parse_address(address: str) -> Tuple[str, str, str, str]:
    """解析地址字符串为省市区街道四个部分"""
    province_pattern = r'([\u4e00-\u9fa5]{2,}(?:省|自治区|市))'
    city_pattern = r'([\u4e00-\u9fa5]{2,}(?:市|自治州|地区|盟))'
    district_pattern = r'([\u4e00-\u9fa5]{2,}(?:区|县|市|旗))'

    province = re.search(province_pattern, address)
    city = re.search(city_pattern, address)
    district = re.search(district_pattern, address)

    province_str = province.group(0) if province else ''
    city_str = city.group(0) if city else ''
    district_str = district.group(0) if district else ''

    remaining = address
    for part in [province_str, city_str, district_str]:
        remaining = remaining.replace(part, '', 1)
    street_str = remaining.strip()

    return province_str, city_str, district_str, street_str

def backfill_addresses(batch_size: int = 1000):
    """批量回填地址数据"""
    last_id = 0
    total_migrated = 0
    total_failed = 0

    while True:
        rows = db.query("""
            SELECT id, address FROM users
            WHERE id > %s AND province IS NULL AND address IS NOT NULL
            ORDER BY id ASC
            LIMIT %s
        """, (last_id, batch_size))

        if not rows:
            break

        for row in rows:
            try:
                province, city, district, street = parse_address(row['address'])
                db.execute("""
                    UPDATE users
                    SET province = %s, city = %s, district = %s, street = %s
                    WHERE id = %s AND province IS NULL
                """, (province, city, district, street, row['id']))
                total_migrated += 1
            except Exception as e:
                logging.error(f"Failed to parse address for user {row['id']}: {e}")
                total_failed += 1

            last_id = row['id']

        logging.info(f"Migrated {total_migrated} rows, failed {total_failed}, last_id={last_id}")
        time.sleep(0.1)  # 控制回填速度，避免影响在线业务

    return total_migrated, total_failed

def verify_migration(sample_size: int = 10000) -> dict:
    """抽样校验迁移数据的正确性"""
    rows = db.query("""
        SELECT id, address, province, city, district, street
        FROM users
        WHERE province IS NOT NULL
        ORDER BY RAND()
        LIMIT %s
    """, (sample_size,))

    results = {'total': len(rows), 'match': 0, 'mismatch': 0, 'mismatches': []}

    for row in rows:
        reconstructed = f"{row['province']}{row['city']}{row['district']}{row['street']}"
        if reconstructed == row['address']:
            results['match'] += 1
        else:
            results['mismatch'] += 1
            if len(results['mismatches']) < 100:
                results['mismatches'].append({
                    'id': row['id'],
                    'original': row['address'],
                    'reconstructed': reconstructed
                })

    results['match_rate'] = results['match'] / results['total'] if results['total'] > 0 else 0
    return results

def get_user_address(user_id: int) -> dict:
    row = db.query_one("SELECT province, city, district, street FROM users WHERE id = %s", (user_id,))
    return {
        'province': row['province'],
        'city': row['city'],
        'district': row['district'],
        'street': row['street'],
        'full': f"{row['province']}{row['city']}{row['district']}{row['street']}"
    }

-- 收缩阶段：删除旧列（确认安全后执行）
ALTER TABLE users DROP COLUMN address;

public class DualWriteRepository {
    private final OldRepository oldRepo;
    private final NewRepository newRepo;
    private final MigrationConfig config;

    public void save(Order order) {
        // 先写旧库（主库）
        oldRepo.save(order);

        try {
            // 再写新库（目标库）
            newRepo.save(transform(order));
        } catch (Exception e) {
            // 新库写入失败不影响业务
            // 但需要记录到修复队列
            repairQueue.enqueue(new RepairTask(order.getId(), "save", e));
            metrics.increment("dual_write.new_store.failure");
        }
    }

    private NewOrder transform(Order order) {
        // 将旧数据模型转换为新数据模型
        return NewOrder.builder()
            .id(order.getId())
            .customerId(order.getCustomerId())
            .totalAmount(order.getTotalCents() / 100.0)
            .currency(order.getCurrency())
            .shippingAddress(AddressMapper.fromLegacy(order.getAddress()))
            .createdAt(order.getCreatedAt())
            .build();
    }
}

class AsyncDualWriter:
    def __init__(self, old_db, change_publisher):
        self.old_db = old_db
        self.change_publisher = change_publisher

    def save_order(self, order: Order):
        # 只写旧库
        self.old_db.save(order)

        # 发布变更事件，由消费者异步写入新库
        event = ChangeEvent(
            table='orders',
            operation='INSERT',
            data=order.to_dict(),
            timestamp=datetime.utcnow()
        )
        self.change_publisher.publish('order-changes', event)

class MigrationAwareRepository:
    def __init__(self, old_db, new_db, config):
        self.old_db = old_db
        self.new_db = new_db
        self.config = config

    def find_order(self, order_id: str) -> Order:
        # 始终从旧库读取
        order = self.old_db.find_order(order_id)

        # 如果开启了影子读取，同时从新库读取并比较
        if self.config.shadow_read_enabled:
            try:
                new_order = self.new_db.find_order(order_id)
                self._compare_and_report(order, new_order)
            except Exception as e:
                metrics.increment("shadow_read.failure")

        return order

    def _compare_and_report(self, old_order, new_order):
        """比较新旧库的数据，上报差异"""
        if new_order is None:
            metrics.increment("shadow_read.missing_in_new")
            return

        diffs = []
        for field in ['status', 'total_amount', 'customer_id']:
            old_val = getattr(old_order, field)
            new_val = getattr(new_order, field)
            if old_val != new_val:
                diffs.append(f"{field}: old={old_val}, new={new_val}")

        if diffs:
            metrics.increment("shadow_read.mismatch")
            logging.warning(f"Data mismatch for order {old_order.id}: {diffs}")
        else:
            metrics.increment("shadow_read.match")

# Debezium 连接器配置示例
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: orders-cdc-connector
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  config:
    database.hostname: old-mysql-primary
    database.port: "3306"
    database.user: cdc_reader
    database.password: "${CDC_PASSWORD}"
    database.server.id: "10001"
    database.server.name: "orders-db"
    database.include.list: "ecommerce"
    table.include.list: "ecommerce.orders,ecommerce.order_items"
    database.history.kafka.bootstrap.servers: "kafka:9092"
    database.history.kafka.topic: "schema-changes.orders"
    snapshot.mode: "initial"
    decimal.handling.mode: "string"
    tombstones.on.delete: "true"

class ConsistencyChecker:
    def __init__(self, old_db, new_db, repair_queue):
        self.old_db = old_db
        self.new_db = new_db
        self.repair_queue = repair_queue

    def check_batch(self, table: str, start_id: int, end_id: int) -> dict:
        """批量比对一段 ID 范围内的数据"""
        old_rows = self.old_db.query(
            f"SELECT * FROM {table} WHERE id BETWEEN %s AND %s ORDER BY id",
            (start_id, end_id)
        )
        new_rows = self.new_db.query(
            f"SELECT * FROM {table} WHERE id BETWEEN %s AND %s ORDER BY id",
            (start_id, end_id)
        )

        old_map = {row['id']: row for row in old_rows}
        new_map = {row['id']: row for row in new_rows}

        stats = {'checked': 0, 'match': 0, 'mismatch': 0, 'missing_new': 0, 'extra_new': 0}

        # 检查旧库有但新库没有的记录
        for id, old_row in old_map.items():
            stats['checked'] += 1
            if id not in new_map:
                stats['missing_new'] += 1
                self.repair_queue.enqueue(RepairTask(table, id, 'missing_in_new'))
            else:
                if self._rows_match(old_row, new_map[id]):
                    stats['match'] += 1
                else:
                    stats['mismatch'] += 1
                    self.repair_queue.enqueue(RepairTask(table, id, 'mismatch'))

        # 检查新库有但旧库没有的记录
        for id in new_map:
            if id not in old_map:
                stats['extra_new'] += 1
                self.repair_queue.enqueue(RepairTask(table, id, 'extra_in_new'))

        return stats

    def _rows_match(self, old_row: dict, new_row: dict, ignored_fields=None) -> bool:
        """比较两行数据是否一致"""
        ignored = ignored_fields or {'updated_at', 'created_at'}
        for key in old_row:
            if key in ignored:
                continue
            if key not in new_row:
                return False
            if old_row[key] != new_row[key]:
                return False
        return True

class MigrationSwitch:
    """迁移切换控制器，支持灰度切换和紧急回切"""

    def __init__(self, feature_flag_client):
        self.ff = feature_flag_client

    def get_read_source(self, user_id: str) -> str:
        """决定读取来源：old / new / both"""
        phase = self.ff.get_string('migration.read_source', default='old')

        if phase == 'shadow':
            return 'both'  # 新旧都读，以旧库为准，新库用于校验

        if phase == 'new_with_fallback':
            return 'new_fallback_old'  # 优先新库，失败回退旧库

        if phase == 'new':
            return 'new'  # 只读新库

        return 'old'  # 默认只读旧库

    def get_write_target(self, user_id: str) -> str:
        """决定写入目标：old / both / new"""
        phase = self.ff.get_string('migration.write_target', default='both')
        return phase

    def emergency_rollback(self):
        """紧急回切：将所有读写切回旧库"""
        self.ff.set_string('migration.read_source', 'old')
        self.ff.set_string('migration.write_target', 'old')
        logging.critical("Migration emergency rollback triggered!")
        alerting.send("Migration rollback", severity="critical")

# gh-ost 的命令行示例
gh-ost \
  --host=mysql-primary \
  --database=ecommerce \
  --table=orders \
  --alter="ADD COLUMN shipping_address_id BIGINT" \
  --allow-on-master \
  --max-load="Threads_running=25" \
  --critical-load="Threads_running=100" \
  --chunk-size=1000 \
  --nice-ratio=0.5 \
  --throttle-query="SELECT IF(COUNT(*)>1000, 1, 0) FROM ecommerce.replication_lag_monitor" \
  --execute

-- 步骤 1: 创建一个占位表
CREATE TABLE _tablename_del LIKE tablename;

-- 步骤 2: 在一个会话中锁定原表（阻塞所有写入）
LOCK TABLES tablename WRITE, _tablename_del WRITE;

-- 步骤 3: 在锁定期间进行原子交换
-- 注意：RENAME TABLE 是原子操作
RENAME TABLE
  tablename TO _tablename_del,
  _tablename_gho TO tablename;

-- 步骤 4: 释放锁
UNLOCK TABLES;

-- 步骤 5: 清理占位表
DROP TABLE _tablename_del;

class StripeMigrationValidator:
    """Stripe 风格的迁移校验器"""

    def __init__(self, old_store, new_store, metrics):
        self.old_store = old_store
        self.new_store = new_store
        self.metrics = metrics

    def validate_continuous(self, table: str, interval_seconds: int = 60):
        """持续校验：每隔一段时间抽样比对"""
        while True:
            sample_ids = self.old_store.random_sample_ids(table, count=1000)
            results = self.validate_batch(table, sample_ids)

            self.metrics.gauge('migration.consistency_rate', results['match_rate'])
            self.metrics.counter('migration.checked_total', results['total'])
            self.metrics.counter('migration.mismatches_total', results['mismatch'])

            if results['match_rate'] < 0.999:
                alerting.warn(
                    f"Migration consistency below threshold: "
                    f"{results['match_rate']:.4f} for table {table}"
                )

            time.sleep(interval_seconds)

    def validate_batch(self, table: str, ids: list) -> dict:
        results = {'total': len(ids), 'match': 0, 'mismatch': 0, 'missing': 0}

        for id in ids:
            old_record = self.old_store.get(table, id)
            new_record = self.new_store.get(table, id)

            if old_record is None:
                continue

            if new_record is None:
                results['missing'] += 1
                continue

            if self._records_equal(old_record, new_record):
                results['match'] += 1
            else:
                results['mismatch'] += 1

        results['match_rate'] = (
            results['match'] / (results['total'] - results['missing'])
            if (results['total'] - results['missing']) > 0 else 1.0
        )
        return results

    def _records_equal(self, old, new) -> bool:
        """比较记录，忽略已知的差异字段"""
        ignored_keys = {'internal_id', 'migrated_at', 'legacy_field'}
        for key in old:
            if key in ignored_keys:
                continue
            if old.get(key) != new.get(key):
                return False
        return True

// 版本 1：初始定义
syntax = "proto3";

message Order {
  int64 id = 1;
  string customer_id = 2;
  int64 total_cents = 3;
  string currency = 4;
  string status = 5;
  int64 created_at = 6;
}

// 版本 2：添加新字段
message Order {
  int64 id = 1;
  string customer_id = 2;
  int64 total_cents = 3;
  string currency = 4;
  string status = 5;
  int64 created_at = 6;

  // 新增字段，使用新的字段编号
  string shipping_address = 7;
  int64 updated_at = 8;
  repeated string tags = 9;
}

// 版本 3：删除字段
message Order {
  reserved 5;  // 原来的 status 字段已废弃
  reserved "status";

  int64 id = 1;
  string customer_id = 2;
  int64 total_cents = 3;
  string currency = 4;
  // status 字段已删除
  int64 created_at = 6;
  string shipping_address = 7;
  int64 updated_at = 8;
  repeated string tags = 9;

  // 用新字段替代旧字段
  OrderStatus order_status = 10;
}

enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_PAID = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_CANCELLED = 4;
}

// 错误示范：以下变更会导致数据损坏

// 1. 修改字段编号
message Order {
  int64 id = 1;
  string customer_id = 3;  // 从 2 改为 3，旧数据中编号 2 的字段会被错误解析
}

// 2. 修改字段类型（不兼容的类型转换）
message Order {
  int64 id = 1;
  int64 customer_id = 2;  // 从 string 改为 int64，旧数据无法正确解码
}

// 3. 复用已删除字段的编号
message Order {
  int64 id = 1;
  string email = 2;  // 复用了 customer_id 的编号 2，旧数据会被错误解读为 email
}

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_cents", "type": "long"},
    {"name": "currency", "type": "string", "default": "USD"},
    {"name": "status", "type": "string", "default": "pending"}
  ]
}

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_cents", "type": "long"},
    {"name": "currency", "type": "string", "default": "USD"},
    {"name": "status", "type": "string", "default": "pending"},
    {"name": "shipping_address", "type": ["null", "string"], "default": null}
  ]
}

# 向 Schema Registry 注册新版本 Schema
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"total_cents\",\"type\":\"long\"},{\"name\":\"shipping_address\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}' \
  http://schema-registry:8081/subjects/orders-value/versions

# 检查兼容性
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "..."}' \
  http://schema-registry:8081/compatibility/subjects/orders-value/versions/latest

// 定义存储接口
type OrderRepository interface {
    FindByID(ctx context.Context, id string) (*Order, error)
    FindByCustomerID(ctx context.Context, customerID string, opts QueryOptions) ([]*Order, error)
    Save(ctx context.Context, order *Order) error
    Update(ctx context.Context, order *Order) error
    Delete(ctx context.Context, id string) error
}

// MySQL 实现
type MySQLOrderRepository struct {
    db *sql.DB
}

func (r *MySQLOrderRepository) FindByID(ctx context.Context, id string) (*Order, error) {
    row := r.db.QueryRowContext(ctx, "SELECT id, customer_id, total_cents, status FROM orders WHERE id = ?", id)
    order := &Order{}
    err := row.Scan(&order.ID, &order.CustomerID, &order.TotalCents, &order.Status)
    if err == sql.ErrNoRows {
        return nil, ErrNotFound
    }
    return order, err
}

func (r *MySQLOrderRepository) Save(ctx context.Context, order *Order) error {
    _, err := r.db.ExecContext(ctx,
        "INSERT INTO orders (id, customer_id, total_cents, status, created_at) VALUES (?, ?, ?, ?, ?)",
        order.ID, order.CustomerID, order.TotalCents, order.Status, order.CreatedAt)
    return err
}

type GradualSwitchRepository struct {
    old       OrderRepository
    new       OrderRepository
    flagStore FeatureFlagStore
}

func (r *GradualSwitchRepository) FindByID(ctx context.Context, id string) (*Order, error) {
    readPercent := r.flagStore.GetInt("migration.read_new_percent")

    if shouldUseNew(id, readPercent) {
        order, err := r.new.FindByID(ctx, id)
        if err != nil {
            // 新存储失败，降级到旧存储
            metrics.Increment("read.new.fallback")
            return r.old.FindByID(ctx, id)
        }
        metrics.Increment("read.new.success")
        return order, nil
    }

    metrics.Increment("read.old")
    return r.old.FindByID(ctx, id)
}

func shouldUseNew(id string, percent int) bool {
    // 基于 ID 的哈希值决定是否使用新存储
    // 确保同一 ID 在同一百分比下的决策一致
    hash := crc32.ChecksumIEEE([]byte(id))
    return int(hash%100) < percent
}

# 关系型数据库：一条 SQL 搞定
result = db.query("""
    SELECT o.id, o.total_cents, c.name, c.email
    FROM orders o
    JOIN customers c ON o.customer_id = c.id
    WHERE o.status = 'pending'
""")

# NoSQL 数据库：需要在应用层拼接
pending_orders = dynamo.query(
    TableName='orders',
    IndexName='status-index',
    KeyConditionExpression='#s = :status',
    ExpressionAttributeNames={'#s': 'status'},
    ExpressionAttributeValues={':status': {'S': 'pending'}}
)

customer_ids = [o['customer_id']['S'] for o in pending_orders['Items']]
customers = dynamo.batch_get_item(
    RequestItems={
        'customers': {
            'Keys': [{'id': {'S': cid}} for cid in customer_ids]
        }
    }
)

# 在应用层做 JOIN
customer_map = {c['id']['S']: c for c in customers['Responses']['customers']}
result = []
for order in pending_orders['Items']:
    customer = customer_map.get(order['customer_id']['S'])
    result.append({
        'order_id': order['id']['S'],
        'total_cents': int(order['total_cents']['N']),
        'customer_name': customer['name']['S'] if customer else None,
        'customer_email': customer['email']['S'] if customer else None,
    })

class MigrationMetrics:
    """迁移过程的核心监控指标"""

    def __init__(self, metrics_client):
        self.m = metrics_client

    def record_backfill_progress(self, table: str, migrated: int, total: int):
        """回填进度"""
        self.m.gauge(f'migration.backfill.progress.{table}', migrated / total)
        self.m.gauge(f'migration.backfill.remaining.{table}', total - migrated)

    def record_consistency(self, table: str, match_rate: float):
        """数据一致性率"""
        self.m.gauge(f'migration.consistency.{table}', match_rate)

    def record_dual_write_latency(self, table: str, old_ms: float, new_ms: float):
        """双写延迟对比"""
        self.m.histogram(f'migration.write_latency.old.{table}', old_ms)
        self.m.histogram(f'migration.write_latency.new.{table}', new_ms)
        self.m.histogram(f'migration.write_latency.overhead.{table}', new_ms - old_ms)

    def record_read_source(self, table: str, source: str):
        """读取来源分布"""
        self.m.increment(f'migration.read.{source}.{table}')

    def record_repair(self, table: str, repair_type: str, success: bool):
        """修复操作"""
        status = 'success' if success else 'failure'
        self.m.increment(f'migration.repair.{repair_type}.{status}.{table}')

groups:
  - name: migration_alerts
    rules:
      - alert: MigrationConsistencyLow
        expr: migration_consistency < 0.999
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "数据一致性低于 99.9%"

      - alert: MigrationDualWriteFailureHigh
        expr: rate(migration_dual_write_failure_total[5m]) > 0.01
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: "双写失败率超过 1%"

      - alert: MigrationCDCLagHigh
        expr: migration_cdc_lag_seconds > 60
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "CDC 延迟超过 60 秒"