# 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点：全量重建宽表                           │
│    - 完全重新计算所有数据                                │
│    - 修正历史数据，确保准确性                             │
└─────────────────────────────────────────────────────────┘
```

**任务配置**：
```python
# 每日增量同步新患者
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）：
```python
# 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 会变成**：
```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（复杂）❌**
```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 混合（简单）✅**
```sql
-- DolphinScheduler: SQL 任务（简单）
INSERT INTO ods_patient_custome
SELECT * FROM mysql_source.patient_custome
WHERE update_time >= '${bizdate}'
```

```python
# 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 业务代码
- ✅ 调度更专业，监控更完善

**实施方式**：

```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
  ```bash
  python main.py run clinic_log 2024-01-01
  ```

- **生产环境**：用 DolphinScheduler
  ```bash
  # DS 自动调用
  python scripts/run_job.py --job clinic_log --start-date ${bizdate}
  ```

#### 方案C：混合使用

**特点**：
- ✅ 简单的定时任务用 doris_sync
- ✅ 复杂的工作流用 DolphinScheduler
- ✅ 各取所长

---

## 📁 scripts/ 目录讨论

### 用户的问题
> "scripts/ # 新增：供 DolphinScheduler 调用的脚本 啊意思？"

### 我的解释

#### 为什么要 scripts/ 目录？

我之前的考虑是：
1. **关注点分离**：调度入口和业务代码分开
2. **灵活性**：可以有不同的调用方式
3. **可维护性**：调度相关逻辑集中管理

**但实际上**：
- ❌ 增加了复杂度
- ❌ 多一层目录结构
- ❌ 对实际帮助不大

### 对比两种方案

#### 方案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` 已经支持命令行调用：

```python
# 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 调用**：
```bash
# 直接用现有的 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 配置示例

```yaml
# 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 表结构

```bash
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步：测试任务执行

```bash
# 测试 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

```bash
# 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/ 目录下也有日志

**数据验证**：
```sql
-- 查看同步的数据量
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（另一个工作流调度系统）

---

## 📝 总结

### 核心观点

1. **不要全部用 SQL**：复杂逻辑用 Python，简单同步用 SQL
2. **不需要 scripts/ 目录**：直接用现有的 main.py 就够了
3. **渐进式迁移**：不要一次性替换，逐步验证
4. **保留灵活性**：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
```

**命令示例**：
```bash
# 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` 中的逻辑：

```python
# 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
-- 需要写几百行的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} │
└─────────────────────────────────────────────────┘
```

**核心原则**：
1. ✅ **业务逻辑继续用 Python**（doris_sync/jobs/）
2. ✅ **DolphinScheduler 只负责调度**（什么时候跑）
3. ✅ **不要为了用调度工具而改变业务逻辑的实现方式**
4. ✅ **两者配合使用**，各司其职

### 具体实施步骤

**步骤1**：继续用 doris_sync 开发所有任务
```python
# 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**：测试所有任务
```bash
# 确保所有任务都能正常运行
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 配置调度
```yaml
# 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？

```yaml
用户曾问：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 调度

```yaml
错误做法 ❌：
  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 够用？

```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 的核心价值

```yaml
✅ 可视化管理：Web UI 界面，任务 DAG 依赖关系一目了然
✅ 企业级特性：任务失败重试、邮件/钉钉/飞书告警、任务优先级
✅ 团队协作：多用户权限管理、任务版本管理、审批流程

❌ 不会做的：
  - 不会让批处理变成流处理
  - 不会减少数据处理延迟
  - SeaTunnel CDC 不需要它调度
```

---

## ⚠️ 常见误区与最佳实践

### 误区 1：用 DolphinScheduler 调度 SeaTunnel

```yaml
错误做法 ❌：
  DolphinScheduler（每 5 分钟）→ 启动 SeaTunnel → 同步数据 → 停止 SeaTunnel

正确做法 ✅：
  SeaTunnel 作为守护进程持续运行（Systemd/Supervisor/Docker）
  DolphinScheduler 只调度计算任务
```

### 误区 2：为了"实时"引入 Flink

```yaml
判断标准：
  ✅ 5-15 分钟延迟够用 → 不需要 Flink
  ✅ Python 脚本逻辑复杂 → 不适合 Flink
  ✅ 团队不熟悉 Flink → 维护成本太高

什么时候才需要 Flink：
  - 需要秒级延迟（实时大屏）
  - 需要复杂窗口计算（滑动窗口、会话窗口）
  - 数据量达到百亿级/天
```

### 最佳实践总结

```yaml
✅ 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. 不要忽略数据质量检查
```

---

## 🎓 总结

### 核心观点

```yaml
1. 你的架构已经很优秀：
   ✅ SeaTunnel CDC 做数据同步（秒级）
   ✅ Python 脚本处理复杂逻辑（灵活）
   ✅ Doris 做存储和计算（高性能）

2. DolphinScheduler 的作用：
   ✅ 可视化调度管理
   ✅ 任务依赖编排
   ✅ 监控告警
   ❌ 不是用来调度 SeaTunnel CDC

3. 不需要 Flink：
   ❌ 你的业务不需要秒级延迟
   ❌ Flink 开发成本太高（10 倍）
   ❌ PyFlink 仍然不适合复杂业务逻辑

4. 架构定位：
   ✅ 达到级别 2-3（准实时-近实时）
   ✅ 对标中型互联网公司水平
   ✅ 成本可控、维护简单
   ✅ 满足 90% 的业务场景
```

### 最终建议

```yaml
保持现有架构，优化调度层：

  SeaTunnel CDC（持续运行）✅ 已完成
      ↓
  Doris ODS 层（实时写入）✅ 已完成
      ↓
  Python 脚本（DWD 层）✅ 已完成
      ↓
  DolphinScheduler（调度管理）⚠️ 建议新增
      ↓
  Doris SQL（ADS 层聚合）⚠️ 建议新增
      ↓
  BI 展示（准实时）✅ 目标

预期效果：
  - 端到端延迟：5-15 分钟
  - 开发成本：低（无需重写脚本）
  - 维护成本：低（可视化界面）
  - 满足业务需求：✅
```