DolphinScheduler集成方案.md 42 KB
DolphinScheduler 集成方案完整讨论记录
📋 讨论背景
用户提出的问题：
"现在的配置文件是config.yaml文件，你也要同时支持.env文件，.env文件优先级更高"
随后在实施患者检查数据仓库项目时，用户进一步提出：
"用DolphinScheduler能实现不？现在的每次写代码什么的，很繁琐"
核心问题：
- 每次新增同步任务都要写 Python 代码很繁琐
- 修改逻辑需要改代码、重启
- 运维人员无法操作，必须找开发人员
- 调度逻辑在代码中，不直观
🎯 初步方案：DolphinScheduler 实现架构
我最初提出的完整架构设计
┌─────────────────────────────────────────────────────────┐
│ MySQL（生产库） │
│ 只做一件事：增量同步原始表到 Doris ODS 层 │
│ ✅ 简单 SELECT，不 JOIN、不计算 │
└─────────────────────────────────────────────────────────┘
↓ 轻量级同步
┌─────────────────────────────────────────────────────────┐
│ Doris ODS 层（原始数据层） │
│ MySQL 的镜像表，结构完全一致 │
└─────────────────────────────────────────────────────────┘
↓ 内部计算
┌─────────────────────────────────────────────────────────┐
│ Doris DWD 层（明细数据层） │
│ 在 Doris 内部做表关联、数据清洗 │
└─────────────────────────────────────────────────────────┘
↓ 宽表计算
┌─────────────────────────────────────────────────────────┐
│ Doris ADS 层（应用数据层） │
│ 在 Doris 内部生成宽表，MySQL 无压力 │
└─────────────────────────────────────────────────────────┘
同步策略
方案一：混合策略（推荐）
┌─────────────────────────────────────────────────────────┐
│ 每日同步策略 │
├─────────────────────────────────────────────────────────┤
│ 1. 每天凌晨2点：增量同步新取镜患者 │
│ - 同步昨天有取镜记录的患者 │
│ - 新增到宽表（或更新已存在的患者） │
│ │
│ 2. 每周日凌晨3点：增量更新近期患者的周期数据 │
│ - 更新最近90天取镜患者的周期检查数据 │
│ - 因为他们的周期数据可能还在补充 │
│ │
│ 3. 每月1号凌晨4点：全量重建宽表 │
│ - 完全重新计算所有数据 │
│ - 修正历史数据，确保准确性 │
└─────────────────────────────────────────────────────────┘
任务配置：
# 每日增量同步新患者
register(
"patient_exam_daily", # 每日增量
patient_exam_daily_run, # 同步昨天的新取镜患者
"0 2 * * *", # 每天凌晨2点
enabled=True
)
# 每周更新近期患者
register(
"patient_exam_weekly", # 每周更新
patient_exam_weekly_run, # 更新最近90天患者
"0 3 * * 0", # 每周日凌晨3点
enabled=True
)
# 每月全量重建
register(
"patient_exam_monthly", # 每月全量
patient_exam_monthly_run, # 全量重建
"0 4 1 * *", # 每月1号凌晨4点
enabled=True
)
😱 用户提出的质疑
用户的担心：
"但是用DolphinScheduler实现之前的那些任务，是不是就是写sql，然后就变的非常复杂了。。。。"
用户的洞察非常准确！
我承认的设计错误
把 Python 逻辑改成 SQL，不仅不简单，反而会更复杂：
原 Python 逻辑（export_goods.py）：
# 1. 解析 JSON 检查结果
result_data = json.loads(result)
# 2. 计算基线时间（15天内最近的）
if 0 < diff <= 15 and diff < min_diff:
closest_exam_date = exam_date
# 3. 补充缺失的 SE 数据
if has_al and missing_se:
right_se = get_near_se_data(...)
# 4. 处理日期时间格式
if isinstance(pay_time, str):
pay_date = pay_time.split(" ")[0]
改成 SQL 会变成：
-- JSON 解析
JSON_EXTRACT(result, '$.r_sph') as r_sph
-- 窗口函数找最近的
FIRST_VALUE(exam_date) OVER (
PARTITION BY patient_id
ORDER BY ABS(DATEDIFF(come_time, pickup_time))
ROWS BETWEEN 15 PRECEDING AND CURRENT ROW
)
-- 自关联补充数据
LEFT JOIN exam_table e2 ON
e1.patient_id = e2.patient_id
AND e1.come_date BETWEEN e2.come_date - INTERVAL 15 DAY AND e2.come_date + INTERVAL 15 DAY
-- 日期处理
DATE_FORMAT(COALESCE(pay_time, '0000-00-00'), '%Y-%m-%d')
SQL 会变得超级复杂，难维护！
💡 修正后的方案：混合模式
正确的分工原则
┌─────────────────────────────────────────────────────┐
│ 简单的任务 → DolphinScheduler SQL 任务 │
│ - 表同步（INSERT ... SELECT） │
│ - 数据统计（COUNT, SUM, GROUP BY） │
│ - 数据清洗（UPDATE, DELETE） │
└─────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────┐
│ 复杂的任务 → DolphinScheduler Shell 任务调用 Python │
│ - JSON 解析和计算 │
│ - 复杂的业务逻辑（周期数据计算） │
│ - 多步数据处理流程 │
│ - 需要调试和日志记录的逻辑 │
└─────────────────────────────────────────────────────┘
推荐的混合架构
DolphinScheduler 工作流：患者检查数据每日同步
│
├─ 【SQL任务】ODS-患者表同步 → 简单 INSERT SELECT
├─ 【SQL任务】ODS-就诊表同步 → 简单 INSERT SELECT
├─ 【SQL任务】ODS-检查表同步 → 简单 INSERT SELECT
├─ 【SQL任务】ODS-取镜表同步 → 简单 INSERT SELECT
├─ 【SQL任务】ODS-商品表同步 → 简单 INSERT SELECT（全量）
├─ 【SQL任务】DWD-明细关联 → 简单多表 JOIN
├─ 【Shell任务】数据量检查 → Python 脚本检查数据量
├─ 【Shell任务】ADS-宽表计算 → Python 脚本处理复杂逻辑
└─ 【Shell任务】数据质量报告 → Python 脚本生成质量报告并发送通知
具体代码对比
方案A：全部用 SQL（复杂）❌
-- ADS 宽表计算 SQL（几百行，难维护）
INSERT INTO fact_patient_exam_wide
SELECT
patient_id,
(SELECT JSON_EXTRACT(result, '$.r_sph') FROM ...) as r_sph,
(SELECT ... FROM ... WHERE ... AND ... GROUP BY ...) as baseline_se,
(SELECT ... FROM ... WHERE ... AND ... GROUP BY ...) as m1_se,
...
(SELECT ... FROM ... WHERE ... AND ... GROUP BY ...) as y2_se
FROM ...
方案B：SQL + Python 混合（简单）✅
-- DolphinScheduler: SQL 任务（简单）
INSERT INTO ods_patient_custome
SELECT * FROM mysql_source.patient_custome
WHERE update_time >= '${bizdate}'
# DolphinScheduler: Shell 任务调用 Python（处理复杂逻辑）
# /app/scripts/patient_exam_wide.py
def run(date):
# 1. 查询数据（简单查询）
patients = query_patients_from_doris(date)
# 2. 复杂的业务逻辑（Python 处理）
for patient in patients:
# 解析 JSON
exam_data = parse_exam_json(patient['exam_result'])
# 计算基线时间
baseline = find_baseline(exam_data, patient['pickup_time'])
# 计算各个周期数据
periods = calculate_periods(exam_data, baseline)
# 3. 写入 Doris（简单插入）
insert_to_doris(patient, periods)
🔄 集成方案讨论
用户的问题
"DolphinScheduler能嵌入现在的这个doris_sync项目吗？"
我的解释
DolphinScheduler 不能"嵌入"到 Python 项目，但可以协作使用：
┌─────────────────────────────────────────────┐
│ doris_sync 项目（Python） │
│ - jobs/ 目录下的 Python 任务代码 │
│ - db/pool.py（数据库连接池） │
│ - config/（配置管理） │
│ - cron/scheduler.py（简单的调度器） │
└─────────────────────────────────────────────┘
↑ 被调用
│
┌─────────────────────────────────────────────┐
│ DolphinScheduler（调度系统） │
│ - Web UI 配置工作流 │
│ - 定时调度、任务依赖 │
│ - 通过 Shell 任务调用 Python 脚本 │
└─────────────────────────────────────────────┘
三种集成方案
方案A：完全替换（推荐）⭐
特点：
- ✅ 用 DolphinScheduler 完全替换
cron/scheduler.py - ✅ 保留所有 Python 业务代码
- ✅ 调度更专业，监控更完善
实施方式：
# DolphinScheduler Shell 任务
cd /app/doris_sync && python main.py run clinic_log ${bizdate}
# 或者直接调用任务模块
cd /app/doris_sync && python -c "
from jobs.clinic_log import run
run(start_date='${bizdate}', end_date='${bizdate}')
"
目录结构：
doris_sync/
├── jobs/ # 保留：Python 任务代码
│ ├── clinic_log.py
│ ├── consumption.py
│ └── patient_exam/
├── config/ # 保留：配置管理
├── db/ # 保留：数据库连接池
├── logger/ # 保留：日志模块
├── main.py # 保留：手动运行入口
└── scripts/ # 新增：独立脚本（供 DS 调用）
└── run_job.py # 统一的任务运行脚本
方案B：并存（过渡期）
特点：
- 🔄 doris_sync 用于开发和测试
- 🔄 DolphinScheduler 用于生产调度
- 🔄 两者调用同样的 jobs/ 代码
使用场景：
开发环境：用 doris_sync
python main.py run clinic_log 2024-01-01生产环境：用 DolphinScheduler
# DS 自动调用 python scripts/run_job.py --job clinic_log --start-date ${bizdate}
方案C：混合使用
特点：
- ✅ 简单的定时任务用 doris_sync
- ✅ 复杂的工作流用 DolphinScheduler
- ✅ 各取所长
📁 scripts/ 目录讨论
用户的问题
"scripts/ # 新增：供 DolphinScheduler 调用的脚本 啊意思？"
我的解释
为什么要 scripts/ 目录？
我之前的考虑是：
- 关注点分离：调度入口和业务代码分开
- 灵活性：可以有不同的调用方式
- 可维护性：调度相关逻辑集中管理
但实际上：
- ❌ 增加了复杂度
- ❌ 多一层目录结构
- ❌ 对实际帮助不大
对比两种方案
方案A：不用 scripts/ 目录
doris_sync/
├── jobs/ # 任务代码（保持不变）
│ ├── clinic_log.py
│ └── consumption.py
├── main.py # 调度入口 + 手动运行入口
└── cron/scheduler.py # 调度器
# DolphinScheduler 调用
python main.py run clinic_log # 复用了 main.py
方案B：创建 scripts/ 目录
doris_sync/
├── jobs/ # 任务代码（纯业务逻辑）
│ ├── clinic_log.py
│ └── consumption.py
├── scripts/ # 调用入口（新增）
│ └── run_job.py # 统一的任务运行脚本
├── main.py # 手动运行入口（保留）
└── cron/scheduler.py # 调度器（可能废弃）
# DolphinScheduler 调用
python scripts/run_job.py --job clinic_log --start-date 2024-03-04
✅ 最终建议：不创建 scripts/ 目录
理由
1. 直接改造 main.py 就够了
现有的 main.py 已经支持命令行调用：
# main.py
def main():
if len(sys.argv) > 1:
cmd = sys.argv[1]
if cmd == "jobs":
list_jobs()
elif cmd == "run":
job_name = sys.argv[2]
start_date = sys.argv[3] if len(sys.argv) > 3 else None
end_date = sys.argv[4] if len(sys.argv) > 4 else start_date
run_job(job_name, start_date, end_date)
DolphinScheduler 调用：
# 直接用现有的 main.py
python main.py run clinic_log 2024-03-04
# 或者默认昨天的数据
python main.py run clinic_log
2. 最简单的目录结构
doris_sync/
├── jobs/ # 任务代码（保持不变）
│ ├── clinic_log.py
│ ├── consumption.py
│ └── patient_exam/
│ ├── __init__.py
│ ├── ods_sync.py
│ ├── dwd_build.py
│ └── ads_build.py
├── config/ # 配置管理
├── db/ # 数据库连接池
├── logger/ # 日志模块
├── main.py # 统一的调用入口
│ ├── 手动运行：python main.py run clinic_log
│ ├── DS调度：python main.py run clinic_log 2024-03-04
│ └── 定时调度：python main.py（启动 scheduler）
└── cron/scheduler.py # 如果用 DS，这个可以废弃
3. DolphinScheduler 配置示例
# DolphinScheduler 工作流配置
工作流名称: 患者检查数据每日同步
任务1: ODS层同步
类型: SQL
数据源: MySQL → Doris
SQL: INSERT INTO ods_patient_custome SELECT * FROM ...
定时: 每天 01:00
任务2: DWD层计算
类型: Shell
脚本: cd /app/doris_sync && python main.py run ods_patient_exam_sync 2024-03-04
定时: 每天 02:00
前置任务: 任务1
任务3: ADS层计算
类型: Shell
脚本: cd /app/doris_sync && python main.py run ads_patient_exam_build 2024-03-04
定时: 每天 04:00
前置任务: 任务2
🎯 完整的实施方案
第1步：创建 Doris 表结构
mysql -h 127.0.0.1 -P 9030 -u root -p bolin_prd
source sql/patient_exam/01_create_ods_tables.sql
source sql/patient_exam/02_create_dwd_tables.sql
source sql/patient_exam/03_create_ads_tables.sql
第2步：测试任务执行
# 测试 ODS 层同步
python main.py run ods_patient_exam_sync 2024-03-04
# 测试 DWD 层计算
python main.py run dwd_patient_exam_build 2024-03-04
# 测试 ADS 层计算
python main.py run ads_patient_exam_build 2024-03-04
第3步：安装和配置 DolphinScheduler
# Docker 部署（推荐）
git clone https://github.com/apache/dolphinscheduler.git
cd dolphinscheduler/docker
docker-compose up -d
# 访问 Web UI
http://localhost:12345/dolphinscheduler
默认账号: admin/dolphinscheduler123
第4步：在 DolphinScheduler 配置工作流
创建数据源：
- MySQL 数据源（连接生产库）
- Doris 数据源（连接 Doris）
创建工作流：
工作流名称: 患者检查数据ETL
定时调度: 0 1 * * *
任务列表:
├─ SQL任务: ODS-患者表同步
├─ SQL任务: ODS-就诊表同步
├─ SQL任务: DWD-明细关联
└─ Shell任务: ADS-宽表计算
脚本: cd /app/doris_sync && python main.py run ads_patient_exam_build ${bizdate}
第5步：验证和监控
查看执行日志：
- DolphinScheduler Web UI → 工作流实例 → 查看日志
- doris_sync/logs/ 目录下也有日志
数据验证：
-- 查看同步的数据量
SELECT COUNT(*) FROM ods_patient_custome WHERE data_date = '2024-03-04';
SELECT COUNT(*) FROM fact_patient_exam_detail WHERE data_date = '2024-03-04';
SELECT COUNT(*) FROM fact_patient_exam_wide WHERE data_date = '2024-03-04';
📊 DolphinScheduler vs 纯Python方案对比
功能对比
| 功能特性 | 纯Python（doris_sync） | DolphinScheduler |
|---|---|---|
| 任务定义 | 写 Python 代码 | Web UI 拖拽配置 |
| 定时调度 | cron 表达式 | cron 表达式 + Web UI |
| 任务依赖 | 代码实现逻辑 | 可视化依赖连线 |
| 失败重试 | 需要写代码 | 勾选框配置 |
| 日志查看 | 查看日志文件 | Web UI 实时查看 |
| 参数传递 | 代码逻辑 | 参数化配置 |
| 运维操作 | 需要开发人员 | 运维人员自主操作 |
| 代码修改 | 改代码重启 | Web UI 修改 |
| 学习成本 | 低（已熟悉） | 中（需要学习） |
| 监控告警 | 需要自己开发 | 内置支持 |
使用场景建议
场景1：简单项目
推荐：纯 Python（doris_sync）
- 任务少（< 10个）
- 逻辑简单
- 人员少（1-2个开发人员）
场景2：中等项目
推荐：混合方案
- 简单同步用 doris_sync
- 复杂工作流用 DolphinScheduler
场景3：大型项目
推荐：DolphinScheduler
- 任务多（> 20个）
- 多人协作
- 需要完善的监控和告警
💡 最佳实践建议
1. 渐进式迁移
阶段1（第1-2周）：
- 保留 doris_sync，用于开发和测试
- 安装 DolphinScheduler，熟悉操作
阶段2（第3-4周）：
- 在 DolphinScheduler 配置简单的工作流
- 1-2个任务开始用 DolphinScheduler 调度
- 验证稳定性和准确性
阶段3（第2个月）：
- 逐步迁移所有任务到 DolphinScheduler
- 保留 doris_sync 作为备份和手动测试
阶段4（第3个月）：
- 完全切换到 DolphinScheduler
- doris_sync 仅用于本地开发测试
2. 任务分类原则
用 SQL 实现的任务：
- ✅ 表同步（INSERT ... SELECT）
- ✅ 数据统计（COUNT, SUM, AVG）
- ✅ 数据清洗（UPDATE, DELETE）
- ✅ 简单的 ETL（1-2个表 JOIN）
用 Python 实现的任务：
- ✅ JSON 解析和复杂计算
- ✅ 复杂的业务逻辑（周期数据计算）
- ✅ 多步骤的数据处理流程
- ✅ 需要灵活调试的逻辑
3. 监控和告警
DolphinScheduler 内置功能：
- ✅ 任务执行状态监控
- ✅ 成功/失败统计
- ✅ 执行时长统计
- ✅ 邮件/钉钉/企业微信告警
doris_sync 项目保留的监控：
- ✅ 日志文件记录
- ✅ 控制台输出
- ✅ 可添加自定义告警逻辑
🔧 实施清单
必须完成的步骤
- 创建 Doris 表结构（3层：ODS/DWD/ADS）
- 实现 ODS/DWD/ADS 任务代码
- 在 doris_sync 项目中测试通过
- 安装 DolphinScheduler
- 在 DolphinScheduler 配置数据源
- 在 DolphinScheduler 创建工作流
- 测试工作流执行
- 配置定时调度和告警
可选的优化步骤
- 配置数据质量检查任务
- 配置自动通知（钉钉/企业微信/邮件）
- 创建 Rollup 优化查询性能
- 配置分区表管理历史数据
- 编写运维文档和操作手册
📞 技术支持资源
官方文档
- DolphinScheduler: https://dolphinscheduler.apache.org/
- 中文文档: https://dolphinscheduler.apache.org/zh-cn/docs/
社区资源
- GitHub: https://github.com/apache/dolphinscheduler
- 国内镜像和部署教程
相关工具
- DataX（阿里开源的数据同步工具）：https://github.com/alibaba/DataX
- Airflow（另一个工作流调度系统）
📝 总结
核心观点
- 不要全部用 SQL：复杂逻辑用 Python，简单同步用 SQL
- 不需要 scripts/ 目录：直接用现有的 main.py 就够了
- 渐进式迁移：不要一次性替换，逐步验证
- 保留灵活性：doris_sync 和 DolphinScheduler 可以共存
最终方案
doris_sync/（保持现有结构）
├── jobs/ # Python 任务代码
├── config/ # 配置管理
├── db/ # 数据库连接池
├── logger/ # 日志模块
└── main.py # 统一调用入口
├── 开发测试：python main.py run clinic_log
└── DS调度：python main.py run clinic_log 2024-03-04
DolphinScheduler（调度系统）
├── Web UI 配置工作流
├── 简单任务用 SQL
└── 复杂任务用 Shell 调用 Python
命令示例：
# DolphinScheduler Shell 任务配置
# 简单任务（SQL）：
INSERT INTO ods_patient_custome SELECT ...
# 复杂任务（Shell + Python）：
cd /app/doris_sync && python main.py run clinic_log ${bizdate}
🎯 最终确认：业务逻辑应保持Python实现
用户的最终质疑
"但是你用DolphinScheduler完全替代doris_sync是不是不太可行，就那些逻辑，你用sql写死，也弄不对，我的考虑正常不？"
确认用户考虑是正常的
你的考虑非常正常！ 你说得完全对，我之前的想法有问题。
为什么不能用SQL完全替代Python？
原因1：业务逻辑的复杂度
现在的 export_goods.py 中的逻辑：
# 1. 解析JSON格式的检查结果
result_data = json.loads(exam_result['result'])
right_se = result_data.get('r_se')
# 2. 计算基线检查（15天内最近的）
if 0 < diff_days <= 15 and diff_days < min_diff:
closest_exam_date = exam_date
# 3. 根据基线时间计算各个周期数据
for period_name, period_days in [('m1', 30), ('m3', 90), ...]:
period_data = find_period_data(exam_list, baseline_date, period_days)
# 4. 补充缺失的SE数据（从AL值推算）
if has_al and missing_se:
right_se = calculate_se_from_al(right_al)
如果用SQL实现，会变成：
-- 需要写几百行的SQL，包含：
-- 1. JSON_EXTRACT 解析
-- 2. 多层嵌套的子查询
-- 3. 窗口函数（FIRST_VALUE, LAG, LEAD）
-- 4. 复杂的CASE WHEN判断
-- 5. 多次自关联
-- 6. 难以调试和验证
SELECT
patient_id,
(SELECT ... FROM ... WHERE ... AND ...) as baseline_se,
(SELECT ... FROM ... WHERE ... AND ...) as m1_se,
(SELECT ... FROM ... WHERE ... AND ...) as m3_se,
...
(几十个子查询)
FROM ...
结果：
- ❌ SQL 代码会更长、更复杂
- ❌ 调试困难（看不懂问题在哪）
- ❌ 维护困难（改一个小逻辑要改很多地方）
- ❌ 性能可能更差（几十个子查询）
原因2：灵活性和可维护性
| 维度 | Python实现 | SQL实现 |
|---|---|---|
| 代码可读性 | ✅ 清晰，逐步处理 | ❌ 几百行嵌套SQL |
| 调试便利性 | ✅ 可以打断点、打印日志 | ❌ 只能看结果，难以定位 |
| 逻辑修改 | ✅ 改函数逻辑 | ❌ 改多个子查询 |
| 数据验证 | ✅ 可以打印中间结果 | ❌ 难以验证中间步骤 |
| 复杂计算 | ✅ Python标准库 | ❌ SQL窗口函数、自关联 |
| 错误处理 | ✅ try-except | ❌ 难以处理异常 |
原因3：项目架构的清晰度
正确的架构：
doris_sync/jobs/
├── clinic_log.py # 简单逻辑 → SQL
├── consumption.py # 简单逻辑 → SQL
└── patient_exam/
├── ods_sync.py # 简单逻辑 → SQL（增量同步）
├── dwd_build.py # 简单逻辑 → SQL（表关联）
└── ads_build.py # 复杂逻辑 → Python（周期计算）
├── parse_exam_json() # JSON解析
├── find_baseline() # 找基线
├── calculate_periods() # 计算周期
└── fill_missing_data() # 补充数据
DolphinScheduler工作流：
├─ SQL任务：ODS层增量同步
├─ SQL任务：DWD层表关联
└─ Shell任务：ADS层宽表计算
└─ python main.py run ads_patient_exam_build ${bizdate}
核心原则：
- ✅ 简单逻辑 → SQL（INSERT ... SELECT，简单JOIN）
- ✅ 复杂逻辑 → Python（周期计算、JSON解析、数据补充）
- ✅ DolphinScheduler 只负责调度，不改变实现方式
✅ 最终结论
推荐方案：DolphinScheduler + doris_sync 配合使用
┌─────────────────────────────────────────────────┐
│ 开发/测试阶段 │
│ 使用 doris_sync │
│ python main.py run clinic_log 2024-03-04 │
└─────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────┐
│ 生产环境调度 │
│ 使用 DolphinScheduler │
│ 自动调用：python main.py run clinic_log ${bizdate} │
└─────────────────────────────────────────────────┘
核心原则：
- ✅ 业务逻辑继续用 Python（doris_sync/jobs/）
- ✅ DolphinScheduler 只负责调度（什么时候跑）
- ✅ 不要为了用调度工具而改变业务逻辑的实现方式
- ✅ 两者配合使用，各司其职
具体实施步骤
步骤1：继续用 doris_sync 开发所有任务
# jobs/patient_exam/ads_build.py
# 复杂的业务逻辑继续用 Python 实现
def build_wide_row(cursor, patient_id, patient_id_str):
# 1. JSON 解析
exam_data = parse_exam_json(result)
# 2. 找基线
baseline = find_baseline(exam_data, pickup_time)
# 3. 计算周期
periods = calculate_periods(exam_data, baseline)
# 4. 补充数据
fill_missing_data(periods)
return row
步骤2：测试所有任务
# 确保所有任务都能正常运行
python main.py run ods_patient_exam_sync 2026-03-04
python main.py run dwd_patient_exam_build 2026-03-04
python main.py run ads_patient_exam_build 2026-03-04
步骤3：在 DolphinScheduler 配置调度
# DolphinScheduler 工作流配置
工作流名称: 患者检查数据每日同步
任务1: ODS层同步
类型: Shell
脚本: cd /app/doris_sync && python main.py run ods_patient_exam_sync ${bizdate}
定时: 每天 01:00
任务2: DWD层计算
类型: Shell
脚本: cd /app/doris_sync && python main.py run dwd_patient_exam_build ${bizdate}
定时: 每天 02:00
前置任务: 任务1
任务3: ADS层计算
类型: Shell
脚本: cd /app/doris_sync && python main.py run ads_patient_exam_build ${bizdate}
定时: 每天 04:00
前置任务: 任务2
总结
| 问题 | 答案 |
|---|---|
| 能用DolphinScheduler完全替代doris_sync吗？ | ❌ 不能 |
| 能把所有Python逻辑改成SQL吗？ | ❌ 不能（会更复杂） |
| DolphinScheduler的作用是什么？ | ✅ 调度系统（什么时候跑任务） |
| doris_sync的作用是什么？ | ✅ 业务逻辑实现（怎么处理数据） |
| 两者如何配合？ | ✅ DolphinScheduler调用 doris_sync 的任务 |
关键点：
- 不要为了用DolphinScheduler而改变现有架构
- 复杂的业务逻辑必须继续用Python实现
- DolphinScheduler只是调度工具，不是替代品
文档保存时间：2026-03-05
版本：v2.0
说明：本文档记录了关于使用 DolphinScheduler 的完整讨论过程，包括初步方案、用户的质疑、方案的修正，以及最终的建议。保留了所有的思考过程和决策依据，以便后续实施时参考。
重要更新：
- v1.1 增加了"最终确认：业务逻辑应保持Python实现"章节，明确了DolphinScheduler和doris_sync的定位和分工
- v2.0 新增"SeaTunnel + DolphinScheduler + Doris 准实时架构最佳实践"章节
SeaTunnel + DolphinScheduler + Doris 准实时架构最佳实践
📋 文档说明
更新时间：2026-03-05
架构定位：SeaTunnel CDC + Python 脚本 + Doris + DolphinScheduler
适用场景：医疗连锁数据仓库，准实时数据同步与计算（5-15 分钟延迟）
架构水平：达到级别 2-3（准实时-近实时），对标中型互联网公司水平
🎯 核心设计原则
"选择最合适的工具，而不是最流行的工具"
为什么不用 Flink？
用户曾问：SeaTunnel + DolphinScheduler + Doris 支持分钟级别吗？
用户又问：Flink + Kafka 架构啥意思？用 flink 同步吗？我怎么感觉不如 seatunnel 呢？
用户还问：Flink 是不是也可以用 python 脚本实现？
核心问题：
- Flink 是实时计算引擎，不是用来做数据同步的
- SeaTunnel 才是数据同步工具，比 Flink 更适合 MySQL → Doris 场景
- PyFlink 存在，但对于复杂业务逻辑，仍然不如普通 Python 脚本
结论：
✅ 数据同步用 SeaTunnel CDC
✅ 复杂业务逻辑用 Python 脚本
✅ 调度管理用 DolphinScheduler
❌ 不需要 Flink（除非需要秒级延迟）
架构优势
| 优势 | 说明 |
|---|---|
| 开发成本低 | Python 脚本直观易懂，任何开发人员都能维护 |
| 维护成本低 | 不需要专门的 Flink 工程师，团队现有技能即可 |
| 部署简单 | 组件少，依赖少，运维复杂度低 |
| 性能够用 | 满足 BI 看板、运营监控、实时报表需求 |
| 扩展性好 | 后续可无缝升级到更高级架构 |
📊 整体架构图
┌─────────────────────────────────────────────────────────────────┐
│ 数据源层 │
│ MySQL 业务数据库 │
│ (Binlog 实时产生) │
└────────────────────────┬────────────────────────────────────────┘
│
↓ CDC (秒级延迟)
┌─────────────────────────────────────────────────────────────────┐
│ 数据同步层 │
│ SeaTunnel CDC │
│ (持续运行，无需调度器触发) │
│ 功能：MySQL Binlog → Doris ODS 层 │
└────────────────────────┬────────────────────────────────────────┘
│
↓ 实时写入
┌─────────────────────────────────────────────────────────────────┐
│ 数据存储层 │
│ Apache Doris │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ ODS 层 │ │ DWD 层 │ │ ADS 层 │ │
│ │ (原始数据) │ │ (明细宽表) │ │ (聚合指标) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└────────────────────────┬────────────────────────────────────────┘
│
↓ 定时调度
┌─────────────────────────────────────────────────────────────────┐
│ 调度与计算层 │
│ DolphinScheduler │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 数据转换任务 │ │ Doris SQL │ │ 维护任务 │ │
│ │ (Python脚本) │ │ (聚合计算) │ │ (清理/监控) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ 调度频率： │
│ - Python 脚本：每 5-15 分钟 │
│ - Doris SQL：每 1-15 分钟 │
│ - 维护任务：每天凌晨 │
└────────────────────────┬────────────────────────────────────────┘
│
↓ 查询展示
┌─────────────────────────────────────────────────────────────────┐
│ 应用层 │
│ BI 工具 / 业务系统 │
│ (DataV、Superset、自定义报表) │
└─────────────────────────────────────────────────────────────────┘
端到端延迟 = SeaTunnel 延迟（秒级）+ DolphinScheduler 调度间隔（5-15分钟）
🔄 各层详细设计
1. 数据同步层：SeaTunnel CDC
SeaTunnel vs 其他工具对比
| 对比项 | SeaTunnel | Flink | DataX | Maxwell |
|---|---|---|---|---|
| 实时性 | ✅ 秒级 | ✅ 秒级 | ❌ 批处理 | ✅ 秒级 |
| CDC 支持 | ✅ 原生支持 | ⚠️ 需要 Kafka | ❌ 不支持 | ✅ 支持 |
| 配置复杂度 | ⭐ 简单 | ⭐⭐⭐⭐⭐ 复杂 | ⭐⭐ 中等 | ⭐⭐⭐ 中等 |
| 学习成本 | ⭐ 低 | ⭐⭐⭐⭐⭐ 很高 | ⭐⭐ 中 | ⭐⭐⭐ 中 |
| 维护成本 | ⭐ 低 | ⭐⭐⭐⭐⭐ 很高 | ⭐⭐ 中 | ⭐⭐⭐ 中 |
| 多表同步 | ✅ 简单 | ⚠️ 需要编码 | ⚠️ 需要配置 | ⚠️ 有限支持 |
⚠️ 重要：SeaTunnel 不需要 DolphinScheduler 调度
错误做法 ❌：
DolphinScheduler（每 5 分钟）
→ 启动 SeaTunnel
→ 同步数据
→ 停止 SeaTunnel
问题：
- 频繁启停破坏 CDC offset 管理
- 可能丢失数据或重复消费
- 无法保证数据一致性
正确做法 ✅：
SeaTunnel 作为守护进程持续运行（通过 systemd 或 supervisor 管理）
DolphinScheduler 只调度计算任务（Doris SQL、Python 脚本）
2. 数据转换层：Python 脚本
为什么用 Python 而不是 Flink？
| 维度 | Python 脚本 | Flink Java | PyFlink |
|---|---|---|---|
| 开发难度 | ⭐ 简单 | ⭐⭐⭐⭐⭐ 极难 | ⭐⭐⭐ 中等 |
| 代码量 | 300-400 行 | 1500-2000+ 行 | 800-1000+ 行 |
| 维护成本 | ⭐ 低 | ⭐⭐⭐⭐⭐ 极高 | ⭐⭐⭐ 中高 |
| 调试效率 | ⭐ 高（直接运行） | ⭐ 低（需要集群） | ⭐⭐ 中（需要集群） |
| 灵活度 | ⭐⭐⭐⭐⭐ 极高 | ⭐⭐ 低 | ⭐⭐⭐ 中等 |
| 适用场景 | ✅ 复杂业务逻辑 | ❌ 过度设计 | ⚠️ 不适合复杂逻辑 |
3. 数据聚合层：Doris SQL
为什么 Doris SQL 够用？
-- 场景 1：实时销售额（每 5 分钟）
SELECT
org_id,
org_name,
COUNT(*) as order_count,
SUM(pay_amt) as total_amount,
NOW() as update_time
FROM ods_fee_order
WHERE pay_time >= DATE_SUB(NOW(), INTERVAL 5 MINUTE)
GROUP BY org_id, org_name;
4. 调度层：DolphinScheduler
DolphinScheduler 的核心价值
✅ 可视化管理：Web UI 界面，任务 DAG 依赖关系一目了然
✅ 企业级特性：任务失败重试、邮件/钉钉/飞书告警、任务优先级
✅ 团队协作：多用户权限管理、任务版本管理、审批流程
❌ 不会做的：
- 不会让批处理变成流处理
- 不会减少数据处理延迟
- SeaTunnel CDC 不需要它调度
⚠️ 常见误区与最佳实践
误区 1：用 DolphinScheduler 调度 SeaTunnel
错误做法 ❌：
DolphinScheduler（每 5 分钟）→ 启动 SeaTunnel → 同步数据 → 停止 SeaTunnel
正确做法 ✅：
SeaTunnel 作为守护进程持续运行（Systemd/Supervisor/Docker）
DolphinScheduler 只调度计算任务
误区 2：为了"实时"引入 Flink
判断标准：
✅ 5-15 分钟延迟够用 → 不需要 Flink
✅ Python 脚本逻辑复杂 → 不适合 Flink
✅ 团队不熟悉 Flink → 维护成本太高
什么时候才需要 Flink：
- 需要秒级延迟（实时大屏）
- 需要复杂窗口计算（滑动窗口、会话窗口）
- 数据量达到百亿级/天
最佳实践总结
✅ DO（推荐做法）：
1. SeaTunnel CDC 持续运行（无需调度）
2. Python 脚本处理复杂业务逻辑
3. Doris SQL 做聚合计算
4. DolphinScheduler 调度计算任务（不是数据同步）
5. 数据分层：ODS → DWD → ADS
6. 监控告警：数据延迟、数据质量、任务失败
❌ DON'T（避免做法）：
1. 不要用 DolphinScheduler 调度 SeaTunnel
2. 不要为了"实时"引入 Flink（除非真的需要秒级延迟）
3. 不要用 Flink SQL 替代 Python 脚本
4. 不要在业务高峰期运行大批量任务
5. 不要忽略数据质量检查
🎓 总结
核心观点
1. 你的架构已经很优秀：
✅ SeaTunnel CDC 做数据同步（秒级）
✅ Python 脚本处理复杂逻辑（灵活）
✅ Doris 做存储和计算（高性能）
2. DolphinScheduler 的作用：
✅ 可视化调度管理
✅ 任务依赖编排
✅ 监控告警
❌ 不是用来调度 SeaTunnel CDC
3. 不需要 Flink：
❌ 你的业务不需要秒级延迟
❌ Flink 开发成本太高（10 倍）
❌ PyFlink 仍然不适合复杂业务逻辑
4. 架构定位：
✅ 达到级别 2-3（准实时-近实时）
✅ 对标中型互联网公司水平
✅ 成本可控、维护简单
✅ 满足 90% 的业务场景
最终建议
保持现有架构，优化调度层：
SeaTunnel CDC（持续运行）✅ 已完成
↓
Doris ODS 层（实时写入）✅ 已完成
↓
Python 脚本（DWD 层）✅ 已完成
↓
DolphinScheduler（调度管理）⚠️ 建议新增
↓
Doris SQL（ADS 层聚合）⚠️ 建议新增
↓
BI 展示（准实时）✅ 目标
预期效果：
- 端到端延迟：5-15 分钟
- 开发成本：低（无需重写脚本）
- 维护成本：低（可视化界面）
- 满足业务需求：✅