Oracle数仓迁移至StarRocks实施方案

文档版本控制

目录

1. 项目概述
2. 项目目标与范围
3. 技术选型与架构设计
4. 详细迁移方案
5. 数据迁移实施指南
6. 应用程序改造指南
7. 性能优化方案
8. 测试方案
9. 风险评估与应对措施
10. 项目管理计划
11. 培训与知识转移
12. 验收标准
13. 交付物清单
14. 附录

1. 项目概述

1.1 项目背景

本项目旨在将现有Oracle数据仓库系统迁移至自主可控的StarRocks数据库平台，以实现：

• 技术自主可控：降低对商业数据库的依赖
• 成本优化：降低数据库许可证和维护成本
• 性能提升：利用StarRocks的OLAP特性提升查询性能
• 云原生架构：适应现代化数据架构需求

1.2 项目规模

graph LR
    A[Oracle数仓] -->|迁移| B[StarRocks数仓]
    A -->|7000+对象| C[数据表/视图/存储过程等]
    A -->|2000+程序| D[ETL程序/函数/触发器]
    B --> E[完整功能等价系统]
    
    style A fill:#ff9999
    style B fill:#99ff99
    style E fill:#9999ff

迁移对象统计：

• 数据表：约5000+张
• 视图：约1500+个
• 存储过程：约800+个
• 函数：约500+个
• 触发器：约200+个
• ETL程序：约2000+个
• 总计：约10000+数据库对象

1.3 实施周期

本项目需满足VMTC项目计划要求，建议分阶段实施：

gantt
    title Oracle迁移StarRocks项目时间表
    dateFormat  YYYY-MM-DD
    section 准备阶段
    需求调研与分析       :done, prep1, 2025-10-10, 2w
    技术选型与POC       :done, prep2, after prep1, 2w
    详细方案设计        :active, prep3, after prep2, 3w
    
    section 开发阶段
    环境搭建           :dev1, after prep3, 1w
    数据模型转换        :dev2, after dev1, 4w
    ETL程序改造        :dev3, after dev1, 6w
    应用层改造         :dev4, after dev2, 4w
    
    section 测试阶段
    功能测试           :test1, after dev3, 3w
    性能测试           :test2, after test1, 2w
    数据一致性验证      :test3, after test1, 2w
    
    section 上线阶段
    试运行            :deploy1, after test3, 2w
    正式上线          :deploy2, after deploy1, 1w
    稳定期支持         :deploy3, after deploy2, 4w

2. 项目目标与范围

2.1 项目目标

2.1.1 核心目标

1. 完整迁移Oracle数仓内容

   • 迁移约7000+数据库对象（表、视图、存储过程、函数、触发器等）
   • 迁移约2000+ETL程序和业务逻辑
   • 确保数据完整性和准确性达到100%

2. 功能等价性保证

   • 确保迁移后数仓的操作功能与Oracle环境一致
   • 性能指标不低于Oracle环境（目标提升20%-50%）
   • 数据查询结果与Oracle完全一致

3. 知识转移与能力建设

   • 提供完整的迁移方案和技术文档
   • 进行系统的技术培训和知识转移
   • 确保团队具备StarRocks运维能力

2.1.2 预期成果

mindmap
  root((迁移成果))
    技术成果
      自主可控数据库
      现代化数据架构
      高性能OLAP引擎
    业务成果
      业务连续性保障
      功能完全等价
      查询性能提升
    管理成果
      完整技术文档
      运维能力建设
      风险可控可管

2.2 项目范围

2.2.1 数据迁移范围

包含内容：

1. 数据对象迁移

   • 所有业务表（含分区表、临时表）
   • 所有物化视图和普通视图
   • 表结构、索引、约束
   • 数据类型转换和适配

2. 程序对象迁移

   • 存储过程转换（PL/SQL → SQL/Python）
   • 函数转换
   • 触发器逻辑改造
   • ETL作业重构

3. 数据内容迁移

   • 全量历史数据迁移
   • 增量数据同步机制
   • 数据质量验证

不包含内容：

• Oracle数据库的物理备份
• 非业务相关的临时数据
• 已废弃不再使用的对象
• 第三方商业工具的迁移

2.2.2 技术支持范围

1. 项目实施期技术支持

   • 迁移方案设计与咨询
   • 技术问题解决
   • 性能调优指导
   • 代码审核与优化建议

2. 上线后技术支持

   • 提供3-6个月的稳定期支持
   • 紧急问题快速响应（4小时响应）
   • 定期巡检与优化建议
   • 版本升级指导

3. 培训与知识转移

   • StarRocks基础培训（2天）
   • 高级特性培训（2天）
   • 运维管理培训（2天）
   • 故障排查培训（1天）

3. 技术选型与架构设计

3.1 StarRocks技术特性

3.1.1 StarRocks vs Oracle对比

3.1.2 StarRocks核心优势

graph TD
    A[StarRocks核心优势] --> B[高性能]
    A --> C[易扩展]
    A --> D[实时性]
    A --> E[成本低]
    
    B --> B1[向量化执行引擎]
    B --> B2[CBO智能优化器]
    B --> B3[列式存储压缩]
    
    C --> C1[MPP并行架构]
    C --> C2[水平弹性扩展]
    C --> C3[存算分离可选]
    
    D --> D1[实时数据导入]
    D --> D2[秒级物化视图]
    D --> D3[流式更新支持]
    
    E --> E1[开源免费]
    E --> E2[硬件成本低]
    E --> E3[运维简单]
    
    style A fill:#4a90e2,color:#fff
    style B fill:#7ed321,color:#fff
    style C fill:#f5a623,color:#fff
    style D fill:#bd10e0,color:#fff
    style E fill:#50e3c2,color:#fff

3.2 目标架构设计

3.2.1 系统架构图

graph TB
    subgraph "数据源层"
        A1[业务系统]
        A2[Oracle数仓<br/>待迁移]
        A3[外部数据源]
    end
    
    subgraph "数据接入层"
        B1[Flink CDC]
        B2[DataX]
        B3[Broker Load]
        B4[Stream Load]
    end
    
    subgraph "StarRocks集群"
        C1[FE节点<br/>Frontend<br/>元数据+查询调度]
        C2[BE节点1<br/>Backend<br/>数据存储+计算]
        C3[BE节点2<br/>Backend<br/>数据存储+计算]
        C4[BE节点N<br/>Backend<br/>数据存储+计算]
    end
    
    subgraph "应用服务层"
        D1[BI报表系统]
        D2[数据分析平台]
        D3[API服务]
        D4[调度系统<br/>DolphinScheduler]
    end
    
    subgraph "监控运维层"
        E1[Prometheus监控]
        E2[Grafana可视化]
        E3[日志采集ELK]
    end
    
    A1 -->|CDC实时同步| B1
    A2 -->|批量迁移| B2
    A3 -->|数据导入| B3
    
    B1 --> C1
    B2 --> C1
    B3 --> C1
    B4 --> C1
    
    C1 -.管理.-> C2
    C1 -.管理.-> C3
    C1 -.管理.-> C4
    
    C1 --> D1
    C1 --> D2
    C1 --> D3
    C1 --> D4
    
    C1 -.监控.-> E1
    C2 -.监控.-> E1
    E1 --> E2
    C1 -.日志.-> E3
    
    style C1 fill:#4a90e2,color:#fff
    style C2 fill:#7ed321,color:#fff
    style C3 fill:#7ed321,color:#fff
    style C4 fill:#7ed321,color:#fff

3.2.2 StarRocks集群规划

推荐配置（基于中大型数仓场景）：

存储容量规划：

• Oracle原始数据大小 × 0.3（列式压缩比）× 1.5（副本系数）× 1.2（增长预留）

4. 详细迁移方案

4.1 迁移策略

4.1.1 迁移方式选择

graph LR
    A[迁移方式] --> B[一次性切换]
    A --> C[灰度迁移<br/>推荐]
    A --> D[双轨并行]
    
    B --> B1[停机时间长<br/>风险高<br/>不推荐]
    C --> C1[分批迁移<br/>逐步切换<br/>风险可控<br/>✅推荐]
    D --> D1[长期并行<br/>成本高<br/>适合核心系统]
    
    style C fill:#7ed321,color:#fff
    style C1 fill:#7ed321,color:#fff

推荐策略：灰度迁移（分阶段实施）

阶段1：试点迁移（2-3周）

• 选择1-2个非核心业务主题域
• 完成数据迁移和验证
• 评估迁移效果和风险

阶段2：批量迁移（6-8周）

• 按业务主题域分批迁移
• 优先级：非核心 → 次核心 → 核心
• 每批迁移后进行充分验证

阶段3：核心系统迁移（4-6周）

• 迁移核心业务系统
• 双轨运行验证（1-2周）
• 最终切换

阶段4：优化与稳定（4周）

• 性能调优
• 问题修复
• 知识转移

4.1.2 迁移流程图

flowchart TD
    Start([开始迁移]) --> A1[环境准备]
    A1 --> A2[对象梳理与分类]
    A2 --> A3[DDL语句转换]
    A3 --> A4[表结构创建]
    
    A4 --> B1{存储过程/触发器?}
    B1 -->|是| B2[逻辑分析]
    B2 --> B3[改造方案设计]
    B3 --> B4[SQL改写/应用层实现]
    B4 --> B5[代码评审]
    B1 -->|否| C1[数据迁移准备]
    B5 --> C1
    
    C1 --> C2[全量数据迁移]
    C2 --> C3[数据一致性校验]
    C3 --> C4{校验通过?}
    C4 -->|否| C5[差异分析与修复]
    C5 --> C3
    C4 -->|是| D1[增量同步机制建立]
    
    D1 --> E1[功能测试]
    E1 --> E2[性能测试]
    E2 --> E3[压力测试]
    E3 --> E4{测试通过?}
    E4 -->|否| E5[问题修复]
    E5 --> E1
    E4 -->|是| F1[试运行]
    
    F1 --> F2[双轨对比验证]
    F2 --> F3{结果一致?}
    F3 -->|否| F4[问题排查]
    F4 --> F2
    F3 -->|是| G1[正式切换]
    
    G1 --> G2[监控观察期]
    G2 --> G3[稳定性评估]
    G3 --> End([迁移完成])
    
    style Start fill:#4a90e2,color:#fff
    style End fill:#7ed321,color:#fff
    style C4 fill:#f5a623,color:#000
    style E4 fill:#f5a623,color:#000
    style F3 fill:#f5a623,color:#000

4.2 迁移准备工作

4.2.1 对象清单梳理

创建详细的迁移清单：

-- Oracle: 导出对象清单脚本
-- 1. 统计所有表
SELECT 'TABLE' AS object_type, 
       owner, 
       table_name AS object_name,
       num_rows,
       blocks,
       ROUND(blocks * 8192 / 1024 / 1024, 2) AS size_mb
FROM dba_tables
WHERE owner NOT IN ('SYS', 'SYSTEM', 'OUTLN')
ORDER BY blocks DESC;

-- 2. 统计所有视图
SELECT 'VIEW' AS object_type,
       owner,
       view_name AS object_name,
       text_length
FROM dba_views
WHERE owner NOT IN ('SYS', 'SYSTEM');

-- 3. 统计存储过程和函数
SELECT object_type,
       owner,
       object_name,
       status,
       last_ddl_time
FROM dba_objects
WHERE object_type IN ('PROCEDURE', 'FUNCTION', 'PACKAGE', 'PACKAGE BODY')
  AND owner NOT IN ('SYS', 'SYSTEM')
ORDER BY object_type, object_name;

-- 4. 统计触发器
SELECT owner,
       trigger_name,
       table_name,
       triggering_event,
       status
FROM dba_triggers
WHERE owner NOT IN ('SYS', 'SYSTEM');

-- 5. 分析依赖关系
SELECT name,
       type,
       referenced_name,
       referenced_type
FROM dba_dependencies
WHERE owner = 'YOUR_SCHEMA'
ORDER BY name, referenced_name;

4.2.2 兼容性评估

创建兼容性评估表：

5. 数据迁移实施指南

5.1 数据类型映射

5.1.1 类型映射表

5.1.2 类型转换示例

-- Oracle DDL
CREATE TABLE orders (
    order_id NUMBER(18) PRIMARY KEY,
    order_no VARCHAR2(50) NOT NULL,
    customer_id NUMBER(18),
    order_amount NUMBER(18,2),
    order_date DATE,
    create_time TIMESTAMP(6),
    status CHAR(1),
    remark CLOB
);

-- StarRocks DDL（转换后）
CREATE TABLE orders (
    order_id BIGINT NOT NULL COMMENT 'Order ID',
    order_no VARCHAR(50) NOT NULL COMMENT 'Order Number',
    customer_id BIGINT COMMENT 'Customer ID',
    order_amount DECIMAL(18,2) COMMENT 'Order Amount',
    order_date DATETIME COMMENT 'Order Date',
    create_time DATETIME COMMENT 'Create Time',
    status CHAR(1) COMMENT 'Status: A-Active, C-Cancelled',
    remark STRING COMMENT 'Remark'
)
DUPLICATE KEY(order_id)
COMMENT "Orders table migrated from Oracle"
DISTRIBUTED BY HASH(order_id) BUCKETS 32
PROPERTIES (
    "replication_num" = "3",
    "storage_format" = "DEFAULT",
    "compression" = "LZ4"
);

5.2 表结构迁移

5.2.1 StarRocks表模型选择

graph TD
    A[选择表模型] --> B{查询模式?}
    B -->|明细查询<br/>保留所有数据| C[DUPLICATE KEY<br/>明细模型]
    B -->|去重查询<br/>主键唯一| D[UNIQUE KEY<br/>主键模型]
    B -->|聚合查询<br/>预聚合| E[AGGREGATE KEY<br/>聚合模型]
    
    C --> C1[适用场景:<br/>日志/明细表<br/>无主键约束]
    D --> D1[适用场景:<br/>用户表/订单表<br/>需要更新操作]
    E --> E1[适用场景:<br/>报表汇总表<br/>只增不改]
    
    style A fill:#4a90e2,color:#fff
    style C fill:#7ed321,color:#fff
    style D fill:#f5a623,color:#fff
    style E fill:#bd10e0,color:#fff

模型选择指南：

5.2.2 分区策略设计

-- Oracle: Range分区示例
CREATE TABLE sales_data (
    sale_id NUMBER,
    sale_date DATE,
    amount NUMBER(18,2)
)
PARTITION BY RANGE (sale_date) (
    PARTITION p_2023 VALUES LESS THAN (TO_DATE('2024-01-01', 'YYYY-MM-DD')),
    PARTITION p_2024_q1 VALUES LESS THAN (TO_DATE('2024-04-01', 'YYYY-MM-DD')),
    PARTITION p_2024_q2 VALUES LESS THAN (TO_DATE('2024-07-01', 'YYYY-MM-DD'))
);

-- StarRocks: 动态分区（推荐）
CREATE TABLE sales_data (
    sale_id BIGINT,
    sale_date DATE,
    amount DECIMAL(18,2)
)
DUPLICATE KEY(sale_id, sale_date)
PARTITION BY RANGE(sale_date) ()  -- 空分区列表，使用动态分区
DISTRIBUTED BY HASH(sale_id) BUCKETS 32
PROPERTIES (
    "replication_num" = "3",
    -- 动态分区配置
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-30",  -- 保留30天历史
    "dynamic_partition.end" = "3",      -- 提前创建3天分区
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "32",
    -- 过期数据自动删除
    "dynamic_partition.history_partition_num" = "0"  -- 保留所有历史分区，或设置具体数量
);

分区设计建议：

1. 按时间分区（推荐用于大部分事实表）

   • 日分区：数据量大（每天百万级以上）
   • 月分区：数据量适中（每月千万级以下）
   • 年分区：小表或历史归档表

2. 分区裁剪优化

   • 确保查询条件包含分区字段
   • 避免跨太多分区查询

5.2.3 分桶策略

-- 分桶数计算公式：
-- buckets_num = CEIL(data_size_gb * 1024 / 100)
-- 建议：每个bucket约100MB-1GB数据

-- 示例1：小表（< 1GB）
CREATE TABLE dim_customer (...)
DISTRIBUTED BY HASH(customer_id) BUCKETS 8;

-- 示例2：中等表（10GB-100GB）
CREATE TABLE fact_orders (...)
DISTRIBUTED BY HASH(order_id) BUCKETS 32;

-- 示例3：大表（> 100GB）
CREATE TABLE fact_transactions (...)
DISTRIBUTED BY HASH(trans_id) BUCKETS 64;

5.3 数据迁移工具与方法

5.3.1 工具选择矩阵

graph LR
    A[数据迁移工具] --> B[DataX<br/>批量迁移]
    A --> C[Flink CDC<br/>实时同步]
    A --> D[Stream Load<br/>程序导入]
    A --> E[Broker Load<br/>HDFS导入]
    
    B --> B1[离线全量迁移<br/>TB级数据<br/>✅ 推荐]
    C --> C1[增量实时同步<br/>毫秒级延迟<br/>✅ 推荐]
    D --> D1[程序控制<br/>灵活性高]
    E --> E1[海量数据<br/>从HDFS导入]
    
    style B fill:#7ed321,color:#fff
    style C fill:#7ed321,color:#fff

5.3.2 DataX批量迁移（推荐方案）

步骤1：安装配置DataX

# 下载DataX
wget https://github.com/alibaba/DataX/releases/download/v3.0/datax.tar.gz
tar -zxvf datax.tar.gz
cd datax

# 测试安装
python bin/datax.py job/job.json

步骤2：创建迁移配置文件

{
  "job": {
    "setting": {
      "speed": {
        "channel": 4,
        "byte": 10485760
      },
      "errorLimit": {
        "record": 100,
        "percentage": 0.1
      }
    },
    "content": [
      {
        "reader": {
          "name": "oraclereader",
          "parameter": {
            "username": "your_oracle_user",
            "password": "your_oracle_password",
            "column": ["order_id", "order_no", "customer_id", "order_amount", "order_date", "create_time", "status", "remark"],
            "connection": [
              {
                "table": ["orders"],
                "jdbcUrl": ["jdbc:oracle:thin:@192.168.1.100:1521:orcl"]
              }
            ],
            "splitPk": "order_id",
            "where": "order_date >= TO_DATE('2024-01-01', 'YYYY-MM-DD')"
          }
        },
        "writer": {
          "name": "starrockswriter",
          "parameter": {
            "username": "root",
            "password": "your_starrocks_password",
            "database": "dw",
            "table": "orders",
            "column": ["order_id", "order_no", "customer_id", "order_amount", "order_date", "create_time", "status", "remark"],
            "preSql": [],
            "postSql": [],
            "jdbcUrl": "jdbc:mysql://192.168.1.200:9030/",
            "loadUrl": ["192.168.1.200:8030"],
            "loadProps": {
              "format": "json",
              "strip_outer_array": true,
              "max_filter_ratio": "0.1"
            }
          }
        }
      }
    ]
  }
}

步骤3：批量生成迁移脚本

# generate_datax_jobs.py
import cx_Oracle
import json
import os

def generate_datax_config(table_name, columns, split_pk):
    """Generate DataX config for a table"""
    config = {
        "job": {
            "setting": {
                "speed": {"channel": 4},
                "errorLimit": {"record": 100}
            },
            "content": [{
                "reader": {
                    "name": "oraclereader",
                    "parameter": {
                        "username": "${ORACLE_USER}",
                        "password": "${ORACLE_PASSWORD}",
                        "column": columns,
                        "connection": [{
                            "table": [table_name],
                            "jdbcUrl": ["${ORACLE_JDBC_URL}"]
                        }],
                        "splitPk": split_pk
                    }
                },
                "writer": {
                    "name": "starrockswriter",
                    "parameter": {
                        "username": "${SR_USER}",
                        "password": "${SR_PASSWORD}",
                        "database": "${SR_DATABASE}",
                        "table": table_name.lower(),
                        "column": columns,
                        "jdbcUrl": "${SR_JDBC_URL}",
                        "loadUrl": ["${SR_LOAD_URL}"]
                    }
                }
            }]
        }
    }
    return config

# Connect to Oracle and generate configs
conn = cx_Oracle.connect('user/password@host:port/service')
cursor = conn.cursor()

# Get all tables
cursor.execute("""
    SELECT table_name 
    FROM user_tables 
    WHERE table_name NOT LIKE '%$%'
    ORDER BY table_name
""")

for (table_name,) in cursor:
    # Get columns
    cursor.execute(f"""
        SELECT column_name 
        FROM user_tab_columns 
        WHERE table_name = '{table_name}'
        ORDER BY column_id
    """)
    columns = [col[0].lower() for col in cursor]
    
    # Get primary key for splitPk
    cursor.execute(f"""
        SELECT column_name
        FROM user_cons_columns
        WHERE constraint_name = (
            SELECT constraint_name
            FROM user_constraints
            WHERE table_name = '{table_name}' AND constraint_type = 'P'
        )
        AND rownum = 1
    """)
    result = cursor.fetchone()
    split_pk = result[0].lower() if result else columns[0]
    
    # Generate config file
    config = generate_datax_config(table_name, columns, split_pk)
    with open(f'jobs/{table_name.lower()}.json', 'w') as f:
        json.dump(config, f, indent=2)
    
    print(f"Generated config for {table_name}")

cursor.close()
conn.close()

步骤4：执行迁移

#!/bin/bash
# batch_migrate.sh

# 配置环境变量
export ORACLE_USER="your_user"
export ORACLE_PASSWORD="your_password"
export ORACLE_JDBC_URL="jdbc:oracle:thin:@192.168.1.100:1521:orcl"
export SR_USER="root"
export SR_PASSWORD="your_password"
export SR_DATABASE="dw"
export SR_JDBC_URL="jdbc:mysql://192.168.1.200:9030/"
export SR_LOAD_URL="192.168.1.200:8030"

# 日志目录
LOG_DIR="logs/$(date +%Y%m%d)"
mkdir -p $LOG_DIR

# 遍历所有配置文件
for config_file in jobs/*.json; do
    table_name=$(basename "$config_file" .json)
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] Starting migration for table: $table_name"
    
    # 替换环境变量
    envsubst < "$config_file" > "/tmp/datax_job.json"
    
    # 执行迁移
    python datax/bin/datax.py /tmp/datax_job.json \
        > "$LOG_DIR/${table_name}.log" 2>&1
    
    if [ $? -eq 0 ]; then
        echo "[$(date '+%Y-%m-%d %H:%M:%S')] ✅ Success: $table_name"
    else
        echo "[$(date '+%Y-%m-%d %H:%M:%S')] ❌ Failed: $table_name"
    fi
done

echo "[$(date '+%Y-%m-%d %H:%M:%S')] Migration completed"

5.3.3 Flink CDC实时同步（增量方案）

// Flink CDC Oracle to StarRocks
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class OracleToStarRocksCDC {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000); // 1分钟checkpoint
        
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        
        // 创建Oracle CDC源表
        tableEnv.executeSql(
            "CREATE TABLE oracle_source_orders (" +
            "  order_id BIGINT," +
            "  order_no STRING," +
            "  customer_id BIGINT," +
            "  order_amount DECIMAL(18,2)," +
            "  order_date TIMESTAMP," +
            "  PRIMARY KEY (order_id) NOT ENFORCED" +
            ") WITH (" +
            "  'connector' = 'oracle-cdc'," +
            "  'hostname' = '192.168.1.100'," +
            "  'port' = '1521'," +
            "  'username' = 'your_user'," +
            "  'password' = 'your_password'," +
            "  'database-name' = 'ORCL'," +
            "  'schema-name' = 'YOUR_SCHEMA'," +
            "  'table-name' = 'ORDERS'" +
            ")"
        );
        
        // 创建StarRocks目标表
        tableEnv.executeSql(
            "CREATE TABLE starrocks_sink_orders (" +
            "  order_id BIGINT," +
            "  order_no STRING," +
            "  customer_id BIGINT," +
            "  order_amount DECIMAL(18,2)," +
            "  order_date TIMESTAMP," +
            "  PRIMARY KEY (order_id) NOT ENFORCED" +
            ") WITH (" +
            "  'connector' = 'starrocks'," +
            "  'jdbc-url' = 'jdbc:mysql://192.168.1.200:9030'," +
            "  'load-url' = '192.168.1.200:8030'," +
            "  'database-name' = 'dw'," +
            "  'table-name' = 'orders'," +
            "  'username' = 'root'," +
            "  'password' = 'your_password'," +
            "  'sink.properties.format' = 'json'," +
            "  'sink.properties.strip_outer_array' = 'true'" +
            ")"
        );
        
        // 执行同步
        tableEnv.executeSql(
            "INSERT INTO starrocks_sink_orders " +
            "SELECT * FROM oracle_source_orders"
        );
    }
}

5.4 数据一致性校验

5.4.1 校验策略

graph TD
    A[数据校验] --> B[行数校验]
    A --> C[数据抽样对比]
    A --> D[聚合值对比]
    A --> E[业务规则校验]
    
    B --> B1[COUNT对比<br/>必须100%一致]
    C --> C2[随机抽取1-5%<br/>逐行对比]
    D --> D1[SUM/MAX/MIN对比<br/>关键字段验证]
    E --> E1[业务逻辑验证<br/>如金额平衡]
    
    B1 --> F{通过?}
    C1 --> F
    D1 --> F
    E1 --> F
    
    F -->|是| G[校验通过]
    F -->|否| H[差异分析]
    H --> I[增量补偿]
    I --> A
    
    style G fill:#7ed321,color:#fff
    style H fill:#ff6b6b,color:#fff

5.4.2 校验脚本示例

-- 1. 行数对比
-- Oracle
SELECT COUNT(*) AS row_count FROM orders;

-- StarRocks
SELECT COUNT(*) AS row_count FROM orders;

-- 2. 聚合值对比
-- Oracle
SELECT 
    COUNT(*) AS total_rows,
    COUNT(DISTINCT order_id) AS unique_orders,
    SUM(order_amount) AS total_amount,
    MAX(order_date) AS max_date,
    MIN(order_date) AS min_date,
    AVG(order_amount) AS avg_amount
FROM orders;

-- StarRocks（应得到完全相同的结果）
SELECT 
    COUNT(*) AS total_rows,
    COUNT(DISTINCT order_id) AS unique_orders,
    SUM(order_amount) AS total_amount,
    MAX(order_date) AS max_date,
    MIN(order_date) AS min_date,
    AVG(order_amount) AS avg_amount
FROM orders;

-- 3. 分区数据校验
-- Oracle
SELECT 
    TRUNC(order_date, 'MM') AS month,
    COUNT(*) AS row_count,
    SUM(order_amount) AS total_amount
FROM orders
GROUP BY TRUNC(order_date, 'MM')
ORDER BY month;

-- StarRocks
SELECT 
    DATE_TRUNC('month', order_date) AS month,
    COUNT(*) AS row_count,
    SUM(order_amount) AS total_amount
FROM orders
GROUP BY DATE_TRUNC('month', order_date)
ORDER BY month;

-- 4. 数据抽样对比（Python脚本）

# data_validation.py
import cx_Oracle
import pymysql
import pandas as pd
from hashlib import md5

def connect_oracle():
    return cx_Oracle.connect('user/password@host:port/service')

def connect_starrocks():
    return pymysql.connect(
        host='192.168.1.200',
        port=9030,
        user='root',
        password='password',
        database='dw'
    )

def validate_table(table_name, sample_rate=0.05):
    """Validate table data consistency"""
    oracle_conn = connect_oracle()
    sr_conn = connect_starrocks()
    
    # 1. Row count validation
    oracle_count = pd.read_sql(f"SELECT COUNT(*) as cnt FROM {table_name}", oracle_conn)['cnt'][0]
    sr_count = pd.read_sql(f"SELECT COUNT(*) as cnt FROM {table_name}", sr_conn)['cnt'][0]
    
    print(f"Table: {table_name}")
    print(f"  Oracle rows: {oracle_count}")
    print(f"  StarRocks rows: {sr_count}")
    print(f"  Match: {'✅' if oracle_count == sr_count else '❌'}")
    
    if oracle_count != sr_count:
        return False
    
    # 2. Sample data validation
    sample_size = int(oracle_count * sample_rate)
    
    # Get primary key column
    pk_query = f"""
        SELECT column_name
        FROM user_cons_columns
        WHERE constraint_name = (
            SELECT constraint_name
            FROM user_constraints
            WHERE table_name = '{table_name.upper()}' AND constraint_type = 'P'
        )
        AND rownum = 1
    """
    pk_col = pd.read_sql(pk_query, oracle_conn).iloc[0, 0].lower()
    
    # Random sample
    sample_query = f"""
        SELECT * FROM {table_name}
        WHERE rownum <= {sample_size}
        ORDER BY DBMS_RANDOM.VALUE
    """
    oracle_sample = pd.read_sql(sample_query, oracle_conn)
    
    # Get same records from StarRocks
    pk_values = ','.join([str(v) for v in oracle_sample[pk_col].tolist()])
    sr_sample = pd.read_sql(
        f"SELECT * FROM {table_name} WHERE {pk_col} IN ({pk_values})",
        sr_conn
    )
    
    # Compare
    oracle_sample = oracle_sample.sort_values(pk_col).reset_index(drop=True)
    sr_sample = sr_sample.sort_values(pk_col).reset_index(drop=True)
    
    # Convert to hashable format for comparison
    oracle_hash = md5(oracle_sample.to_csv().encode()).hexdigest()
    sr_hash = md5(sr_sample.to_csv().encode()).hexdigest()
    
    print(f"  Sample validation ({sample_size} rows): {'✅' if oracle_hash == sr_hash else '❌'}")
    
    oracle_conn.close()
    sr_conn.close()
    
    return oracle_hash == sr_hash

# Run validation
tables = ['orders', 'customers', 'products']  # Add your tables
for table in tables:
    validate_table(table)

6. 应用程序改造指南

6.1 存储过程改造

6.1.1 改造策略

graph TD
    A[存储过程分析] --> B{复杂度?}
    B -->|简单<br/>单表操作| C[改写为SQL]
    B -->|中等<br/>多表JOIN| D[拆分为多个SQL<br/>应用层编排]
    B -->|复杂<br/>业务逻辑重| E[迁移到应用层<br/>Java/Python实现]
    
    C --> F[StarRocks SQL]
    D --> G[应用服务]
    E --> G
    
    G --> H[调度工具<br/>DolphinScheduler]
    
    style C fill:#7ed321,color:#fff
    style D fill:#f5a623,color:#fff
    style E fill:#ff6b6b,color:#fff

改造优先级：

1. 高优先级：核心业务流程的存储过程
2. 中优先级：定时调度的批处理过程
3. 低优先级：很少使用的辅助过程

6.1.2 典型案例：数据汇总过程

Oracle存储过程示例：

-- Oracle PL/SQL
CREATE OR REPLACE PROCEDURE proc_daily_sales_summary(
    p_date IN DATE,
    p_status OUT VARCHAR2
) AS
    v_total_amount NUMBER;
    v_total_orders NUMBER;
BEGIN
    -- Delete existing summary
    DELETE FROM daily_sales_summary WHERE summary_date = p_date;
    
    -- Calculate summary
    INSERT INTO daily_sales_summary (
        summary_date,
        total_orders,
        total_amount,
        avg_amount,
        max_amount,
        customer_count
    )
    SELECT 
        TRUNC(order_date) AS summary_date,
        COUNT(*) AS total_orders,
        SUM(order_amount) AS total_amount,
        AVG(order_amount) AS avg_amount,
        MAX(order_amount) AS max_amount,
        COUNT(DISTINCT customer_id) AS customer_count
    FROM orders
    WHERE TRUNC(order_date) = p_date
    GROUP BY TRUNC(order_date);
    
    -- Get counts for logging
    SELECT total_orders, total_amount 
    INTO v_total_orders, v_total_amount
    FROM daily_sales_summary
    WHERE summary_date = p_date;
    
    -- Update status
    UPDATE etl_job_log
    SET status = 'SUCCESS',
        rows_processed = v_total_orders,
        amount_processed = v_total_amount,
        end_time = SYSDATE
    WHERE job_name = 'daily_sales_summary'
      AND job_date = p_date;
    
    COMMIT;
    
    p_status := 'SUCCESS';
    
EXCEPTION
    WHEN OTHERS THEN
        ROLLBACK;
        p_status := 'ERROR: ' || SQLERRM;
        
        INSERT INTO etl_error_log (
            job_name, job_date, error_message, error_time
        ) VALUES (
            'daily_sales_summary', p_date, SQLERRM, SYSDATE
        );
        COMMIT;
END;
/

改造方案1：拆分为SQL（推荐用于简单过程）

-- StarRocks SQL脚本
-- 1. 清理已存在的汇总数据
DELETE FROM daily_sales_summary 
WHERE summary_date = '${date}';

-- 2. 插入新的汇总数据
INSERT INTO daily_sales_summary (
    summary_date,
    total_orders,
    total_amount,
    avg_amount,
    max_amount,
    customer_count
)
SELECT 
    DATE_TRUNC('day', order_date) AS summary_date,
    COUNT(*) AS total_orders,
    SUM(order_amount) AS total_amount,
    AVG(order_amount) AS avg_amount,
    MAX(order_amount) AS max_amount,
    COUNT(DISTINCT customer_id) AS customer_count
FROM orders
WHERE DATE_TRUNC('day', order_date) = '${date}'
GROUP BY DATE_TRUNC('day', order_date);

-- 3. 更新作业日志
UPDATE etl_job_log
SET status = 'SUCCESS',
    rows_processed = (SELECT total_orders FROM daily_sales_summary WHERE summary_date = '${date}'),
    amount_processed = (SELECT total_amount FROM daily_sales_summary WHERE summary_date = '${date}'),
    end_time = NOW()
WHERE job_name = 'daily_sales_summary'
  AND job_date = '${date}';

改造方案2：迁移到应用层（推荐用于复杂过程）

# daily_sales_summary.py
import pymysql
from datetime import datetime
import logging

class DailySalesSummary:
    """Daily sales summary job"""
    
    def __init__(self, sr_config):
        self.config = sr_config
        self.logger = logging.getLogger(__name__)
        
    def get_connection(self):
        """Get StarRocks connection"""
        return pymysql.connect(
            host=self.config['host'],
            port=self.config['port'],
            user=self.config['user'],
            password=self.config['password'],
            database=self.config['database']
        )
    
    def execute(self, summary_date):
        """
        Execute daily sales summary
        
        Args:
            summary_date: Date to summarize (YYYY-MM-DD format)
            
        Returns:
            dict: Execution result
        """
        conn = None
        cursor = None
        
        try:
            conn = self.get_connection()
            cursor = conn.cursor()
            
            # Step 1: Delete existing summary
            self.logger.info(f"Deleting existing summary for {summary_date}")
            delete_sql = """
                DELETE FROM daily_sales_summary 
                WHERE summary_date = %s
            """
            cursor.execute(delete_sql, (summary_date,))
            
            # Step 2: Insert new summary
            self.logger.info(f"Calculating summary for {summary_date}")
            insert_sql = """
                INSERT INTO daily_sales_summary (
                    summary_date,
                    total_orders,
                    total_amount,
                    avg_amount,
                    max_amount,
                    customer_count
                )
                SELECT 
                    DATE_TRUNC('day', order_date) AS summary_date,
                    COUNT(*) AS total_orders,
                    SUM(order_amount) AS total_amount,
                    AVG(order_amount) AS avg_amount,
                    MAX(order_amount) AS max_amount,
                    COUNT(DISTINCT customer_id) AS customer_count
                FROM orders
                WHERE DATE_TRUNC('day', order_date) = %s
                GROUP BY DATE_TRUNC('day', order_date)
            """
            cursor.execute(insert_sql, (summary_date,))
            
            # Step 3: Get result for logging
            select_sql = """
                SELECT total_orders, total_amount
                FROM daily_sales_summary
                WHERE summary_date = %s
            """
            cursor.execute(select_sql, (summary_date,))
            result = cursor.fetchone()
            
            if result:
                total_orders, total_amount = result
            else:
                total_orders, total_amount = 0, 0
            
            # Step 4: Update job log
            update_sql = """
                UPDATE etl_job_log
                SET status = 'SUCCESS',
                    rows_processed = %s,
                    amount_processed = %s,
                    end_time = NOW()
                WHERE job_name = 'daily_sales_summary'
                  AND job_date = %s
            """
            cursor.execute(update_sql, (total_orders, total_amount, summary_date))
            
            # Commit transaction
            conn.commit()
            
            self.logger.info(f"Summary completed: {total_orders} orders, ${total_amount}")
            
            return {
                'status': 'SUCCESS',
                'total_orders': total_orders,
                'total_amount': total_amount
            }
            
        except Exception as e:
            self.logger.error(f"Error in daily sales summary: {str(e)}")
            
            if conn:
                conn.rollback()
            
            # Log error
            try:
                error_sql = """
                    INSERT INTO etl_error_log (
                        job_name, job_date, error_message, error_time
                    ) VALUES (%s, %s, %s, NOW())
                """
                cursor.execute(error_sql, ('daily_sales_summary', summary_date, str(e)))
                conn.commit()
            except:
                pass
            
            return {
                'status': 'ERROR',
                'error_message': str(e)
            }
            
        finally:
            if cursor:
                cursor.close()
            if conn:
                conn.close()

# Usage
if __name__ == '__main__':
    import sys
    from datetime import date, timedelta
    
    config = {
        'host': '192.168.1.200',
        'port': 9030,
        'user': 'root',
        'password': 'your_password',
        'database': 'dw'
    }
    
    # Get date parameter (default: yesterday)
    if len(sys.argv) > 1:
        summary_date = sys.argv[1]
    else:
        summary_date = (date.today() - timedelta(days=1)).strftime('%Y-%m-%d')
    
    # Execute job
    job = DailySalesSummary(config)
    result = job.execute(summary_date)
    
    print(f"Job result: {result}")
    sys.exit(0 if result['status'] == 'SUCCESS' else 1)

6.1.3 函数改造

常用Oracle函数映射：

SQL改写示例：

-- Oracle SQL
SELECT 
    order_id,
    NVL(customer_name, 'Unknown') AS customer_name,
    DECODE(status, 'A', 'Active', 'C', 'Cancelled', 'Other') AS status_desc,
    TO_CHAR(order_date, 'YYYY-MM-DD') AS order_date_str,
    TRUNC(order_date, 'MM') AS order_month
FROM orders
WHERE ROWNUM <= 100;

-- StarRocks SQL
SELECT 
    order_id,
    IFNULL(customer_name, 'Unknown') AS customer_name,
    CASE status
        WHEN 'A' THEN 'Active'
        WHEN 'C' THEN 'Cancelled'
        ELSE 'Other'
    END AS status_desc,
    DATE_FORMAT(order_date, '%Y-%m-%d') AS order_date_str,
    DATE_TRUNC('month', order_date) AS order_month
FROM (
    SELECT *, ROW_NUMBER() OVER() AS rn
    FROM orders
) t
WHERE rn <= 100;

6.2 触发器改造

6.2.1 改造策略

StarRocks不支持触发器，需要将触发器逻辑改造到应用层或使用其他机制：

graph TD
    A[Oracle触发器] --> B{触发器类型?}
    B -->|BEFORE INSERT<br/>数据校验| C[应用层实现<br/>数据校验逻辑]
    B -->|AFTER INSERT<br/>日志记录| D[使用审计功能<br/>或应用层记录]
    B -->|BEFORE UPDATE<br/>自动填充| E[应用层设置<br/>默认值]
    B -->|级联更新| F[应用层实现<br/>或使用物化视图]
    
    style C fill:#f5a623,color:#fff
    style D fill:#f5a623,color:#fff
    style E fill:#f5a623,color:#fff
    style F fill:#f5a623,color:#fff

示例：审计触发器改造

-- Oracle触发器
CREATE OR REPLACE TRIGGER trg_orders_audit
AFTER INSERT OR UPDATE OR DELETE ON orders
FOR EACH ROW
BEGIN
    IF INSERTING THEN
        INSERT INTO orders_audit_log (
            order_id, operation, old_amount, new_amount, change_time, change_user
        ) VALUES (
            :NEW.order_id, 'INSERT', NULL, :NEW.order_amount, SYSDATE, USER
        );
    ELSIF UPDATING THEN
        INSERT INTO orders_audit_log (
            order_id, operation, old_amount, new_amount, change_time, change_user
        ) VALUES (
            :NEW.order_id, 'UPDATE', :OLD.order_amount, :NEW.order_amount, SYSDATE, USER
        );
    ELSIF DELETING THEN
        INSERT INTO orders_audit_log (
            order_id, operation, old_amount, new_amount, change_time, change_user
        ) VALUES (
            :OLD.order_id, 'DELETE', :OLD.order_amount, NULL, SYSDATE, USER
        );
    END IF;
END;
/

改造方案：应用层实现

# order_service.py
class OrderService:
    """Order service with audit logging"""
    
    def __init__(self, db_conn):
        self.conn = db_conn
        
    def insert_order(self, order_data, user_id):
        """Insert order with audit log"""
        cursor = self.conn.cursor()
        
        try:
            # Insert order
            insert_sql = """
                INSERT INTO orders (order_id, order_no, customer_id, order_amount)
                VALUES (%(order_id)s, %(order_no)s, %(customer_id)s, %(order_amount)s)
            """
            cursor.execute(insert_sql, order_data)
            
            # Insert audit log
            audit_sql = """
                INSERT INTO orders_audit_log (
                    order_id, operation, old_amount, new_amount, change_time, change_user
                ) VALUES (
                    %(order_id)s, 'INSERT', NULL, %(order_amount)s, NOW(), %(user_id)s
                )
            """
            cursor.execute(audit_sql, {
                'order_id': order_data['order_id'],
                'order_amount': order_data['order_amount'],
                'user_id': user_id
            })
            
            self.conn.commit()
            return True
            
        except Exception as e:
            self.conn.rollback()
            raise e
        finally:
            cursor.close()
    
    def update_order(self, order_id, new_amount, user_id):
        """Update order with audit log"""
        cursor = self.conn.cursor()
        
        try:
            # Get old amount
            cursor.execute("SELECT order_amount FROM orders WHERE order_id = %s", (order_id,))
            old_amount = cursor.fetchone()[0]
            
            # Update order
            update_sql = "UPDATE orders SET order_amount = %s WHERE order_id = %s"
            cursor.execute(update_sql, (new_amount, order_id))
            
            # Insert audit log
            audit_sql = """
                INSERT INTO orders_audit_log (
                    order_id, operation, old_amount, new_amount, change_time, change_user
                ) VALUES (
                    %s, 'UPDATE', %s, %s, NOW(), %s
                )
            """
            cursor.execute(audit_sql, (order_id, old_amount, new_amount, user_id))
            
            self.conn.commit()
            return True
            
        except Exception as e:
            self.conn.rollback()
            raise e
        finally:
            cursor.close()

6.3 ETL作业改造

6.3.1 调度工具选择：DolphinScheduler

graph LR
    A[DolphinScheduler] --> B[任务编排]
    A --> C[依赖管理]
    A --> D[监控告警]
    A --> E[多种任务类型]
    
    B --> B1[DAG工作流]
    C --> C1[上下游依赖]
    D --> D1[失败重试<br/>邮件告警]
    E --> E1[Shell/SQL<br/>Python/Spark]
    
    style A fill:#4a90e2,color:#fff

6.3.2 ETL作业迁移示例

Oracle DBMS_SCHEDULER作业：

-- Oracle定时作业
BEGIN
    DBMS_SCHEDULER.CREATE_JOB (
        job_name        => 'daily_etl_job',
        job_type        => 'STORED_PROCEDURE',
        job_action      => 'proc_daily_etl',
        start_date      => SYSTIMESTAMP,
        repeat_interval => 'FREQ=DAILY; BYHOUR=2; BYMINUTE=0',
        enabled         => TRUE
    );
END;
/

DolphinScheduler作业配置：

# dolphinscheduler_workflow.yaml
workflow:
  name: "daily_etl_workflow"
  description: "Daily ETL job migrated from Oracle"
  schedule: "0 2 * * *"  # 每天凌晨2点执行
  
  tasks:
    - name: "extract_data"
      type: "SHELL"
      script: |
        #!/bin/bash
        python /opt/etl/extract_orders.py --date ${date}
      
    - name: "transform_data"
      type: "PYTHON"
      script: |
        from etl import transform_orders
        transform_orders('${date}')
      depends_on: ["extract_data"]
      
    - name: "load_to_starrocks"
      type: "SQL"
      datasource: "starrocks_prod"
      sql: |
        INSERT INTO fact_orders
        SELECT * FROM staging_orders
        WHERE etl_date = '${date}'
      depends_on: ["transform_data"]
      
    - name: "data_quality_check"
      type: "PYTHON"
      script: |
        from etl import quality_check
        quality_check('fact_orders', '${date}')
      depends_on: ["load_to_starrocks"]
      
  alerts:
    - type: "EMAIL"
      on_failure: true
      recipients: ["dba@company.com", "dev@company.com"]

7. 性能优化方案

7.1 性能优化策略

mindmap
  root((性能优化))
    表设计优化
      合理选择表模型
      分区分桶设计
      列顺序优化
      数据类型选择
    索引优化
      前缀索引
      Bitmap索引
      倒排索引
    查询优化
      SQL改写
      JOIN优化
      物化视图
      CBO优化器
    系统优化
      集群配置
      内存调优
      并发控制
      监控告警

7.2 表设计性能优化

7.2.1 列顺序优化原则

最佳实践：

1. KEY列：放在最前面（用于数据排序和过滤）
2. 频繁查询列：紧跟KEY列
3. 低基数列：相对靠前（便于压缩）
4. 大字段：放在最后（STRING, TEXT等）

-- ❌ 不推荐的列顺序
CREATE TABLE orders_bad (
    remark STRING,           -- 大字段在前
    order_amount DECIMAL(18,2),
    order_status CHAR(1),
    order_date DATE,
    order_id BIGINT          -- KEY列在最后
) ...;

-- ✅ 推荐的列顺序
CREATE TABLE orders_good (
    order_id BIGINT,         -- KEY列：主键
    order_date DATE,         -- KEY列：分区字段
    order_status CHAR(1),    -- 常用过滤字段，低基数
    customer_id BIGINT,      -- 常用JOIN字段
    order_amount DECIMAL(18,2),
    order_no VARCHAR(50),
    remark STRING            -- 大字段放最后
)
DUPLICATE KEY(order_id, order_date)
PARTITION BY RANGE(order_date) ()
DISTRIBUTED BY HASH(order_id) BUCKETS 32;

7.2.2 索引设计

StarRocks支持的索引类型：

-- 1. Bitmap索引（适用于状态、类型等低基数列）
CREATE INDEX idx_order_status ON orders (order_status) USING BITMAP;
CREATE INDEX idx_payment_type ON orders (payment_type) USING BITMAP;

-- 2. Bloom Filter（适用于JOIN字段）
CREATE TABLE orders (
    ...
) ...
PROPERTIES (
    "bloom_filter_columns" = "customer_id, product_id"
);

7.3 查询优化

7.3.1 SQL改写技巧

**技巧1：避免SELECT ***

-- ❌ 不推荐
SELECT * FROM orders WHERE order_date = '2024-01-01';

-- ✅ 推荐：只查询需要的列
SELECT order_id, customer_id, order_amount 
FROM orders 
WHERE order_date = '2024-01-01';

技巧2：充分利用分区裁剪

-- ❌ 不推荐：无法利用分区裁剪
SELECT * FROM orders 
WHERE DATE_FORMAT(order_date, '%Y-%m') = '2024-01';

-- ✅ 推荐：直接使用分区字段
SELECT * FROM orders 
WHERE order_date >= '2024-01-01' AND order_date < '2024-02-01';

技巧3：JOIN优化

-- ✅ 推荐：小表（维度表）在前，让优化器选择Broadcast Join
SELECT o.*, c.customer_name
FROM dim_customer c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.order_date = '2024-01-01';

7.3.2 物化视图加速查询

-- 创建物化视图：日订单汇总
CREATE MATERIALIZED VIEW mv_daily_orders_summary
PARTITION BY order_date
DISTRIBUTED BY HASH(order_date) BUCKETS 10
REFRESH ASYNC START('2024-01-01 00:00:00') EVERY(INTERVAL 1 HOUR)
AS
SELECT 
    DATE_TRUNC('day', order_date) AS order_date,
    customer_id,
    COUNT(*) AS order_count,
    SUM(order_amount) AS total_amount
FROM orders
GROUP BY DATE_TRUNC('day', order_date), customer_id;

7.4 监控与诊断

-- 查看正在执行的查询
SELECT 
    query_id,
    query_start_time,
    TIMESTAMPDIFF(SECOND, query_start_time, NOW()) AS running_seconds,
    `user`,
    state,
    LEFT(query, 200) AS query_preview
FROM information_schema.query_statistics
WHERE state = 'RUNNING'
ORDER BY query_start_time;

-- 查看慢查询历史
SELECT 
    query_id,
    query_start_time,
    query_time / 1000 AS query_time_seconds,
    scan_rows,
    scan_bytes / 1024 / 1024 AS scan_mb,
    LEFT(query, 200) AS query_preview
FROM information_schema.query_statistics
WHERE query_time > 10000  -- 超过10秒的查询
ORDER BY query_time DESC
LIMIT 50;

8. 测试方案

8.1 测试策略

graph LR
    A[测试阶段] --> B[单元测试]
    A --> C[功能测试]
    A --> D[性能测试]
    A --> E[数据一致性测试]
    A --> F[回归测试]
    
    B --> B1[SQL单元测试<br/>表结构验证]
    C --> C1[业务功能验证<br/>端到端测试]
    D --> D1[性能基准测试<br/>压力测试]
    E --> E1[数据对比验证<br/>100%一致]
    F --> F1[版本升级验证<br/>变更回归]
    
    style E fill:#ff6b6b,color:#fff
    style E1 fill:#ff6b6b,color:#fff

8.2 功能测试

测试用例模板：

8.3 性能测试

性能基准测试：

8.4 数据一致性测试

核心验证点：

• ✅ 行数100%一致
• ✅ 聚合值100%一致（SUM/AVG/MAX/MIN）
• ✅ 随机抽样数据逐行对比
• ✅ 业务规则验证（如金额平衡）

9. 风险评估与应对措施

9.1 风险识别

mindmap
  root((项目风险))
    技术风险
      兼容性问题
      性能不达标
      数据丢失风险
      功能缺失
    进度风险
      人力不足
      技术难度估计不足
      依赖方延期
    业务风险
      业务中断
      数据错误影响决策
      用户使用不适应
    管理风险
      需求变更频繁
      沟通不畅
      测试不充分

9.2 风险清单与应对措施

9.3 应急预案

9.3.1 数据迁移失败预案

graph TD
    A[数据迁移失败] --> B{严重程度?}
    B -->|轻微<br/>部分数据| C[增量补偿]
    B -->|严重<br/>大量错误| D[回滚到Oracle]
    
    C --> C1[1. 识别差异数据]
    C1 --> C2[2. 增量导入]
    C2 --> C3[3. 重新校验]
    
    D --> D1[1. 停止StarRocks服务]
    D1 --> D2[2. 切换应用到Oracle]
    D2 --> D3[3. 问题分析]
    D3 --> D4[4. 修复后重新迁移]
    
    style A fill:#ff6b6b,color:#fff
    style D fill:#ff6b6b,color:#fff

回滚时间目标：

• 决策时间：<30分钟
• 执行回滚：<1小时
• 业务恢复：<2小时

10. 项目管理计划

10.1 项目组织架构

graph TD
    A[项目指导委员会] --> B[项目经理]
    B --> C[技术组]
    B --> D[测试组]
    B --> E[运维组]
    B --> F[业务组]
    
    C --> C1[架构师]
    C --> C2[数据库工程师]
    C --> C3[ETL开发工程师]
    
    D --> D1[测试负责人]
    D --> D2[测试工程师]
    
    E --> E1[运维负责人]
    E --> E2[DBA]
    
    F --> F1[业务代表]
    F --> F2[需求分析师]
    
    style A fill:#4a90e2,color:#fff
    style B fill:#7ed321,color:#fff

10.2 项目里程碑

10.3 沟通管理

10.3.1 会议机制

10.4 质量管理

10.4.1 质量标准

10.5 变更管理

10.5.1 变更流程

graph TD
    A[变更请求提交] --> B[变更评估]
    B --> C{影响评估}
    C -->|重大变更| D[指导委员会审批]
    C -->|一般变更| E[项目经理审批]
    C -->|轻微变更| F[技术负责人审批]
    
    D --> G[变更实施]
    E --> G
    F --> G
    
    G --> H[变更验证]
    H --> I[变更记录归档]
    
    style A fill:#4a90e2,color:#fff
    style G fill:#7ed321,color:#fff

11. 培训与知识转移

11.1 培训计划

gantt
    title 培训计划时间表
    dateFormat  YYYY-MM-DD
    section DBA培训
    StarRocks基础         :train1, 2025-11-01, 2d
    集群管理与运维         :train2, after train1, 2d
    性能调优              :train3, after train2, 2d
    故障排查              :train4, after train3, 1d
    
    section 开发培训
    SQL差异与改写         :train5, 2025-11-01, 2d
    数据导入方式          :train6, after train5, 1d
    应用开发最佳实践       :train7, after train6, 2d
    
    section 业务培训
    系统功能介绍          :train8, 2025-11-15, 1d
    报表使用指导          :train9, after train8, 1d

11.2 培训内容

11.2.1 DBA培训（7天）

Day 1-2: StarRocks基础

• StarRocks架构原理
• FE/BE节点职责
• 表模型详解（DUPLICATE/UNIQUE/AGGREGATE）
• 数据类型和函数
• 实操：创建表、导入数据

Day 3-4: 集群管理与运维

• 集群部署与扩容
• 用户权限管理
• 数据备份与恢复
• 监控指标解读
• 实操：集群扩容演练

Day 5-6: 性能调优

• 表设计优化
• 索引策略
• 物化视图使用
• 查询优化技巧
• 实操：慢查询优化

Day 7: 故障排查

• 常见问题诊断
• 日志分析
• 问题定位方法
• 应急处理流程
• 实操：故障演练

11.2.2 开发人员培训（5天）

Day 1-2: SQL差异与改写

• Oracle vs StarRocks SQL差异
• 常用函数映射
• 存储过程改造方法
• 事务处理差异
• 实操：SQL改写练习

Day 3: 数据导入

• Stream Load使用
• Broker Load使用
• Flink CDC实时同步
• 实操：数据导入实践

Day 4-5: 应用开发最佳实践

• JDBC连接配置
• 连接池优化
• 错误处理
• 性能优化技巧
• 实操：应用改造实践

11.3 培训材料

11.3.1 交付材料清单

• StarRocks架构设计文档
• 表设计规范文档
• SQL改写指南（Oracle → StarRocks）
• 数据导入操作手册
• 集群运维手册
• 故障排查手册
• 性能优化手册
• 培训视频录像（每个培训主题）
• 实操练习题库
• FAQ常见问题解答

11.4 技能认证

认证体系：

12. 验收标准

12.1 验收总体原则

graph LR
    A[验收标准] --> B[数据完整性]
    A --> C[功能正确性]
    A --> D[性能达标]
    A --> E[文档完整性]
    A --> F[培训完成度]
    
    B --> B1[100%一致]
    C --> C1[100%功能可用]
    D --> D1[≥Oracle的80%性能]
    E --> E1[100%交付]
    F --> F1[100%通过认证]
    
    style A fill:#4a90e2,color:#fff
    style B1 fill:#7ed321,color:#fff
    style C1 fill:#7ed321,color:#fff
    style D1 fill:#7ed321,color:#fff
    style E1 fill:#7ed321,color:#fff
    style F1 fill:#7ed321,color:#fff

12.2 详细验收标准

12.2.1 数据完整性验收

12.2.2 功能正确性验收

12.2.3 性能验收

12.2.4 文档完整性验收

交付文档清单：

• 项目实施方案（本文档）
• 详细设计文档
• 数据模型设计文档
• ETL程序改造文档
• 数据迁移报告
• 功能测试报告
• 性能测试报告
• 数据一致性验证报告
• 集群运维手册
• 故障排查手册
• 应急预案文档
• 培训教材（PPT+视频）
• 操作手册（用户端）
• 项目总结报告

12.3 验收流程

graph TD
    A[项目组自检] --> B{自检通过?}
    B -->|否| C[问题修复]
    C --> A
    B -->|是| D[提交验收申请]
    
    D --> E[验收准备]
    E --> F[验收会议]
    
    F --> G[数据验收]
    F --> H[功能验收]
    F --> I[性能验收]
    F --> J[文档验收]
    
    G --> K{验收结果}
    H --> K
    I --> K
    J --> K
    
    K -->|有问题| L[问题清单]
    L --> M[整改]
    M --> D
    
    K -->|通过| N[签署验收报告]
    N --> O[项目移交]
    O --> P[进入质保期]
    
    style A fill:#4a90e2,color:#fff
    style N fill:#7ed321,color:#fff
    style P fill:#7ed321,color:#fff

12.4 验收标准总结

pie title 验收权重分布
    "数据完整性" : 35
    "功能正确性" : 30
    "性能达标" : 20
    "文档完整性" : 10
    "培训完成度" : 5

验收通过条件：

• 总分 ≥ 90分
• 数据完整性和功能正确性必须100%达标
• 无一级缺陷（阻塞性问题）

13. 交付物清单

13.1 技术文档

• Oracle迁移StarRocks实施方案（本文档）
• 详细设计文档
• 技术选型报告
• POC测试报告
• 表结构对照表
• SQL改写文档
• 存储过程改造文档
• ETL作业改造文档

13.2 测试文档

• 测试计划
• 测试用例
• 性能测试报告
• 数据一致性验证报告
• 测试总结报告

13.3 运维文档

• 集群部署手册
• 日常运维操作手册
• 监控配置手册
• 故障排查手册
• 应急预案

13.4 工具与脚本

• 表结构转换工具
• 数据迁移脚本
• 数据校验工具
• 集群管理脚本
• 监控脚本

13.5 培训材料

• 培训PPT
• 培训视频
• 练习题库
• 认证考试题库

13.6 项目管理文档

• 项目计划
• 风险管理表
• 问题跟踪表
• 项目周报
• 项目总结报告

14. 附录

14.1 附录A：Oracle与StarRocks SQL差异速查表

14.2 附录B：常见问题FAQ

Q1: StarRocks是否支持事务？

A: StarRocks的UNIQUE KEY和PRIMARY KEY模型支持部分事务特性：

• 支持单表的原子性写入
• Stream Load和Broker Load保证事务性
• 不支持跨表事务
• 建议在应用层实现事务逻辑

Q2: 如何处理Oracle的SEQUENCE？

A: 三种替代方案：

1. 使用StarRocks的AUTO_INCREMENT列（推荐）
2. 应用层生成ID（如雪花算法）
3. 使用外部ID生成服务（如Redis）

Q3: 大字段（CLOB/BLOB）如何迁移？

A:

• CLOB → STRING类型（最大1MB）
• 超过1MB的大文本：建议存储在对象存储（OSS/S3），数据库只存URL
• BLOB：不推荐存储在OLAP数仓，建议存储在对象存储

Q4: 迁移过程中如何保证数据不丢失？

A:

1. 迁移前完整备份Oracle数据
2. 分批迁移，每批验证
3. 迁移期间保持Oracle环境运行
4. 使用增量同步技术（Flink CDC）
5. 详细的数据校验流程

Q5: StarRocks的查询性能真的比Oracle快吗？

A:

• OLAP场景：StarRocks通常快3-10倍（列式存储+向量化引擎）
• OLTP场景：Oracle更快（StarRocks不适合高并发点查）
• 复杂分析：StarRocks明显优势
• 实际性能取决于数据量、查询模式、表设计等

Q6: 数据导入失败如何排查？

A: 排查步骤：

1. 查看导入任务状态

   SHOW LOAD WHERE LABEL = 'your_label';
   

2. 查看错误日志

   curl http://be_host:8040/api/error_log/your_label
   

3. 常见原因：
   • 数据格式不匹配
   • 列数不一致
   • 超过配置的错误率
   • 网络超时

14.3 附录C：参考资料

14.3.1 官方文档

• StarRocks官方文档：https://docs.starrocks.io/
• StarRocks GitHub：https://github.com/StarRocks/starrocks
• DataX文档：https://github.com/alibaba/DataX
• Flink CDC文档：https://ververica.github.io/flink-cdc-connectors/

14.3.2 社区资源

• StarRocks中文社区：https://forum.starrocks.io/
• 技术博客：https://www.starrocks.io/blog

14.4 附录D：术语表

15. 总结

15.1 项目成功关键因素

mindmap
  root((成功关键))
    充分准备
      详细的需求分析
      全面的风险评估
      合理的项目计划
    技术保障
      POC充分验证
      分阶段实施
      完善的测试
      数据校验机制
    团队能力
      技术培训
      知识转移
      专家支持
    管理控制
      进度把控
      质量管理
      变更管理
      沟通机制

15.2 经验教训

DO's（应该做的）：

1. ✅ 充分的POC验证，确保技术可行性
2. ✅ 分阶段灰度迁移，降低风险
3. ✅ 严格的数据校验，确保100%一致
4. ✅ 详细的文档和培训，确保可维护
5. ✅ 完善的监控和告警，快速发现问题
6. ✅ 双轨并行验证，确保切换安全
7. ✅ 充分的应急预案，快速回滚

DON'T（不应该做的）：

1. ❌ 不要一次性切换，风险太高
2. ❌ 不要跳过数据校验，数据一致性是底线
3. ❌ 不要忽视性能测试，性能问题影响体验
4. ❌ 不要低估存储过程改造工作量
5. ❌ 不要忽视培训，运维能力很重要
6. ❌ 不要在生产高峰期切换
7. ❌ 不要没有回滚预案就上线

15.3 持续改进

项目上线后，持续优化：

1. 性能优化

   • 定期分析慢查询
   • 优化表结构和索引
   • 调整系统参数

2. 功能增强

   • 根据用户反馈优化
   • 新增便捷功能
   • 提升用户体验

3. 运维优化

   • 自动化运维工具
   • 完善监控告警
   • 优化备份策略

4. 知识积累

   • 故障案例总结
   • 最佳实践沉淀
   • 团队能力提升

文档变更历史

文档结束

如有任何问题或需要进一步说明，请联系项目组。