患者检查数据仓库实施指南

📋 实施步骤

阶段1：创建 Doris 表结构（第1步）

操作：执行建表 SQL

# 连接 Doris
mysql -h 127.0.0.1 -P 9030 -u root -p

# 使用数据库
USE bolin_prd;

# 按顺序执行建表语句
source E:/公司/BI1.0/doris_sync/sql/patient_exam/01_create_ods_tables.sql
source E:/公司/BI1.0/doris_sync/sql/patient_exam/02_create_dwd_tables.sql
source E:/公司/BI1.0/doris_sync/sql/patient_exam/03_create_ads_tables.sql

验证：

-- 查看创建的表
SHOW TABLES LIKE 'ods_%';
SHOW TABLES LIKE 'fact_%';

-- 查看表结构
DESC ods_patient_custome;
DESC fact_patient_exam_detail;
DESC fact_patient_exam_wide;

阶段2：测试 ODS 层同步（第2步）

操作：手动运行 ODS 同步任务

cd E:\公司\BI1.0\doris_sync

# 运行 ODS 层同步（同步昨天的数据）
python main.py run ods_patient_exam_sync

预期输出：

============================================================
[ODS同步] 开始同步: 2026-03-04
============================================================

开始同步: ods_patient_custome (increment)
查询到 50 条记录
已写入 50/50
✓ ods_patient_custome 同步完成: 50 条

开始同步: ods_clinic_clinic (increment)
查询到 100 条记录
已写入 100/100
✓ ods_clinic_clinic 同步完成: 100 条

...（其他表）

============================================================
[ODS同步] 同步完成，成功 9/9 个表
============================================================

验证：

-- 查看同步的数据
SELECT COUNT(*) FROM ods_patient_custome;
SELECT COUNT(*) FROM ods_clinic_clinic;
SELECT * FROM ods_patient_custome LIMIT 10;

阶段3：测试 DWD 层计算（第3步）

操作：手动运行 DWD 计算任务

# 运行 DWD 层计算
python main.py run dwd_patient_exam_build

预期输出：

============================================================
[DWD计算] 开始计算: 2026-03-04
============================================================
清理旧数据: 0 条
开始执行计算...
============================================================
[DWD计算] 计算完成: 500 条
============================================================

验证：

-- 查看计算的明细数据
SELECT COUNT(*) FROM fact_patient_exam_detail;
SELECT
    patient_name,
    goods_name,
    come_time,
    pickup_time
FROM fact_patient_exam_detail
LIMIT 10;

阶段4：测试 ADS 层计算（第4步）

操作：手动运行 ADS 计算任务

# 运行 ADS 层计算
python main.py run ads_patient_exam_build

预期输出：

============================================================
[ADS计算] 开始计算: 2026-03-04
============================================================
步骤1: 查询取镜患者...
找到 50 个患者
步骤2: 生成宽表数据...
步骤3: 写入宽表...
已写入 50/50
============================================================
[ADS计算] 计算完成: 50 条
============================================================

验证：

-- 查看生成的宽表数据
SELECT
    patient_id_str,
    patient_name,
    fee_class,
    last_pickup_date,
    baseline_right_se_val,
    m3_right_se_val,
    p3_right_se_val
FROM fact_patient_exam_wide
LIMIT 10;

阶段5：启用定时任务（第5步）

操作：修改 cron/tasks.py，启用任务

# 将 enabled=False 改为 enabled=True
register("ods_patient_exam_sync", ods_patient_exam_sync_run, "0 1 * * *", enabled=True)
register("dwd_patient_exam_build", dwd_build_run, "0 2 * * *", enabled=True)
register("ads_patient_exam_build", ads_build_run, "0 4 * * *", enabled=True)

验证任务状态：

# 查看任务列表
python main.py jobs

预期输出：

============================================================
初始化检测完成，所有连接正常
============================================================

任务列表:
  - clinic_log: 0 2 * * * [启用]
  - consumption: 0 3 * * * [启用]
  - clean_logs: 0 1 * * * [启用]
  - ods_patient_exam_sync: 0 1 * * * [启用]      # 新增
  - dwd_patient_exam_build: 0 2 * * * [启用]     # 新增
  - ads_patient_exam_build: 0 4 * * * [启用]     # 新增

阶段6：启动调度器（第6步）

操作：启动调度器

# 启动调度器（所有任务按 cron 表达式自动执行）
python main.py

预期日志：

==================================================
初始化检测完成，所有连接正常
==================================================

2026-03-05 01:00:00 - INFO - [ODS同步] 开始同步: 2026-03-04
2026-03-05 01:05:00 - INFO - [ODS同步] 同步完成，成功 9/9 个表

2026-03-05 02:00:00 - INFO - [DWD计算] 开始计算: 2026-03-04
2026-03-05 02:05:00 - INFO - [DWD计算] 计算完成: 500 条

2026-03-05 04:00:00 - INFO - [ADS计算] 开始计算: 2026-03-04
2026-03-05 04:10:00 - INFO - [ADS计算] 计算完成: 50 条

⚠️ 注意事项

1. 任务执行顺序

01:00 - ODS 层同步（MySQL → Doris，轻量级）
02:00 - DWD 层计算（Doris 内部关联）
04:00 - ADS 层计算（Doris 内部生成宽表）

重要：必须按顺序执行，因为下一层依赖上一层的数据。

2. 资源占用评估

任务	MySQL 压力	Doris 计算	耗时
ODS 层同步	✅ 轻（只查增量）	无	5-10 分钟
DWD 层计算	无	✅ 重（关联）	10-20 分钟
ADS 层计算	无	✅ 重（宽表）	20-30 分钟

3. 数据量监控

-- 查看 ODS 层数据量
SELECT
    TABLE_NAME,
    TABLE_ROWS
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'bolin_prd'
  AND TABLE_NAME LIKE 'ods_%'
ORDER BY TABLE_ROWS DESC;

-- 查看 DWD/ADS 层数据量
SELECT
    TABLE_NAME,
    TABLE_ROWS
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'bolin_prd'
  AND TABLE_NAME LIKE 'fact_%'
ORDER BY TABLE_ROWS DESC;

4. 故障排查

问题1：ODS 同步失败

# 查看日志
tail -f logs/ods-$(date +%Y-%m-%d).log

# 手动测试单表同步
python -c "
from jobs.patient_exam.ods_sync import sync_table
from db.pool import get_mysql_conn, get_doris_conn

mysql_conn = get_mysql_conn()
doris_conn = get_doris_conn()

sync_table(mysql_conn, doris_conn, {
    'name': 'ods_patient_custome',
    'source_table': 'patient_custome',
    'time_field': 'update_time',
    'sync_mode': 'increment',
    'columns': 'id, patientId, name, sex, birthday, mobile, customer_optometrist_staff_id, create_time, update_time'
}, '2026-03-04', '2026-03-04')
"

问题2：DWD 计算失败

-- 检查 ODS 层数据是否完整
SELECT COUNT(*) FROM ods_clinic_clinic WHERE come_time >= '2026-03-04';
SELECT COUNT(*) FROM ods_patient_custome;
SELECT COUNT(*) FROM ods_exam_examresult;

问题3：ADS 宽表数据不准确

这是正常的！因为周期数据计算逻辑复杂，需要多次迭代优化
建议：先运行一段时间，每周手动验证一次数据准确性
优化：参考 export_goods.py 中的逻辑，完善周期数据计算

📊 性能优化建议

优化1：使用分区表（数据量 > 100万时）

-- 按数据日期分区
ALTER TABLE fact_patient_exam_wide
ADD PARTITION BY RANGE(`data_date`) (
    PARTITION p202401 VALUES LESS THAN ('2024-02-01'),
    PARTITION p202402 VALUES LESS THAN ('2024-03-01'),
    PARTITION p202403 VALUES LESS THAN ('2024-04-01'),
    PARTITION p202404 VALUES LESS THAN ('2024-05-01')
);

优化2：创建 Rollup（加速常用查询）

ALTER TABLE fact_patient_exam_wide
ADD ROLLUP recent_data (
    patient_id, patient_name, mobile,
    baseline_right_se_val, baseline_left_se_val,
    p3_right_se_val, p3_left_se_val,
    p6_right_se_val, p6_left_se_val
);

优化3：调整 Doris 内存配置

# conf/fe.conf
mem_limit = 80%  # 使用 80% 内存

# conf/be.conf
write_buffer_size = 200MB  # 写缓冲区

✅ 验收标准

Doris 表创建成功
ODS 层能正常同步 MySQL 数据
DWD 层能正常计算明细数据
ADS 层能正常生成宽表
定时任务能自动执行
数据准确性与原始导出脚本一致（±5%）

📞 技术支持

如有问题，查看：

日志文件：logs/sync-YYYY-MM-DD.log
任务状态：python main.py jobs
手动运行：python main.py run <任务名称>

实施指南.md 8.4 KB Pysyvä linkki Historia Raaka

患者检查数据仓库实施指南

📋 实施步骤

阶段1：创建 Doris 表结构（第1步）

阶段2：测试 ODS 层同步（第2步）

阶段3：测试 DWD 层计算（第3步）

阶段4：测试 ADS 层计算（第4步）

阶段5：启用定时任务（第5步）

阶段6：启动调度器（第6步）

⚠️ 注意事项

1. 任务执行顺序

2. 资源占用评估

3. 数据量监控

4. 故障排查

📊 性能优化建议

优化1：使用分区表（数据量 > 100万时）

优化2：创建 Rollup（加速常用查询）

优化3：调整 Doris 内存配置

✅ 验收标准

📞 技术支持

实施指南.md 8.4 KB

Pysyvä linkki Historia Raaka