Doris 同步服务

MySQL 数据同步到 Apache Doris

目录结构

doris_sync/
├── config.yaml       # 默认配置文件（YAML 格式）
├── .env              # 环境变量配置文件（优先级更高）
├── requirements.txt
├── main.py
├── config/           # 统一配置模块
│   ├── __init__.py   # 配置接口
│   ├── loader.py     # 配置加载器
│   └── checker.py    # 初始化检测
├── cron/
│   ├── scheduler.py  # 调度器
│   └── tasks.py      # 任务列表(追加任务在这里)
├── db/pool.py
├── logger/handler.py
├── logs/              # 日志目录（按天生成）
│   ├── sync-2026-03-05.log
│   └── sync-2026-03-06.log
└── jobs/
    ├── clinic_log.py
    ├── consumption.py
    └── clean_logs.py  # 日志清理任务

安装依赖

pip install -r requirements.txt

配置

项目支持两种配置文件，.env 文件优先级更高：

配置优先级

.env 环境变量 > config.yaml > 代码默认值

当同一配置在两个文件中都存在时，优先使用 .env 中的值。

方式一：config.yaml（推荐作为默认配置）

编辑 config.yaml:

database:
  mysql:
    host: "your_mysql_host"
    port: 3306
    user: "your_user"
    password: "your_password"
    database: "your_database"
    pool_size: 10
    charset: "utf8mb4"

  doris:
    host: "127.0.0.1"
    port: 9030
    user: "root"
    password: "your_password"
    database: "bolin_prd"
    pool_size: 5
    charset: "utf8mb4"

logging:
  level: "INFO"
  file: "logs/sync.log"

方式二：.env（推荐用于敏感信息和环境覆盖）

编辑 .env 文件（优先级更高，会覆盖 config.yaml 中的配置）：

# MySQL 配置
MYSQL_HOST=your_mysql_host
MYSQL_PORT=3306
MYSQL_USER=your_user
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=your_database
MYSQL_POOL_SIZE=10
MYSQL_CHARSET=utf8mb4

# Doris 配置
DORIS_HOST=127.0.0.1
DORIS_PORT=9030
DORIS_USER=root
DORIS_PASSWORD=your_password
DORIS_DATABASE=bolin_prd
DORIS_POOL_SIZE=5
DORIS_CHARSET=utf8mb4

配置建议

config.yaml：放置通用配置和默认值
.env：放置敏感信息（密码）和环境特定配置，建议加入 .gitignore

使用示例

# 获取全部配置
from config import get_config
config = get_config()

# 获取特定配置
mysql_host = get_config("database.mysql.host")
pool_size = get_config("database.mysql.pool_size", default=10)

执行命令

启动调度器

python main.py

查看任务列表

python main.py jobs

单独运行任务

# 运行指定任务(昨天)
python main.py run clinic_log

# 指定日期范围
python main.py run clinic_log 2025-01-01 2025-01-31

追加任务

在 jobs/ 目录下创建新任务文件，然后在 cron/tasks.py 中注册：

# 1. 在文件顶部导入
from jobs.new_task import run as new_task_run

# 2. 注册任务
register(
    "new_task",             # 任务名称
    new_task_run,           # 执行函数
    "0 4 * * *",           # cron 表达式
    enabled=True           # 是否启用
)

任务函数的标准格式：

# jobs/new_task.py
def run(start_date=None, end_date=None):
    """执行任务"""
    if start_date is None or end_date is None:
        # 默认同步昨天的数据
        yesterday = datetime.date.today() - datetime.timedelta(days=1)
        start_date = end_date = yesterday.strftime("%Y-%m-%d")

    # 你的同步逻辑...
    return success_count

启用/禁用任务

在 cron/tasks.py 中修改 enabled 参数：

# 启用任务
register("clinic_log", clinic_log_run, "0 2 * * *", enabled=True)

# 禁用任务
register("clinic_log", clinic_log_run, "0 2 * * *", enabled=False)

# 或者注释掉整行
# register("clinic_log", clinic_log_run, "0 2 * * *", enabled=True)

Cron 表达式

格式：分时日月周

表达式	含义
`0 2 * * *`	每天凌晨 2 点
`0 3 * * *`	每天凌晨 3 点
`0 /2 * *`	每隔 2 小时
`0 0 * * 0`	每周日凌晨 0 点
`/30 * * *`	每 30 分钟

在线生成工具：https://crontab.guru/

日志

日志文件命名

日志文件按天生成独立文件，格式为 任务名-日期.log：

logs/
├── sync-2026-03-05.log    # 今天的日志
├── sync-2026-03-04.log    # 昨天的日志
├── sync-2026-03-03.log    # 前天的日志
└── ...

查看日志

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

# 查看昨天的日志
tail -f logs/sync-$(date -d yesterday +%Y-%m-%d).log

# 查看所有日志文件
ls -lh logs/

日志清理

系统每天凌晨 1 点自动清理超过 30 天的旧日志文件。

如需手动清理或修改保留天数：

# 在 jobs/clean_logs.py 中修改
clean_old_logs(days=30)  # 修改为其他天数

README.md 5.0 KB История Директен файл