DonpinScheduler 是一款轻量级的分布式任务调度工具，适合用来管理和调度 SQL 脚本任务，包括 Couchbase SQL 脚本。以下是一个使用 DonpinScheduler 执行 Couchbase SQL 脚本的实际例子。
使用场景
目标：使用 DonpinScheduler 定时执行一组 Couchbase SQL 脚本，完成以下任务：
每小时同步数据到 Couchbase。
每天聚合统计数据。
每周清理过期数据。
需求：
定时调度 SQL 脚本。
支持分布式任务调度。
提供任务日志记录和失败后的重试机制。
实现步骤
- 准备 Couchbase SQL 脚本
示例脚本文件：
sync_data.sql：
INSERT INTO bucket-name (KEY, VALUE)
SELECT META().id, new_data.*
FROM source-bucket new_data
WHERE META().id NOT IN (SELECT RAW META().id FROM bucket-name);
aggregate_data.sql：
SELECT category, COUNT(*) AS count
FROM bucket-name
WHERE type = “product”
GROUP BY category;
cleanup_expired_data.sql：
DELETE FROM bucket-name
WHERE expiration_date < NOW_STR();
- 安装 DonpinScheduler 和 Couchbase 驱动
安装所需依赖：
pip install donpinscheduler couchbase
- 配置 DonpinScheduler 和任务
DonpinScheduler 允许通过装饰器定义任务，并支持任务依赖和定时。
SQL 执行函数
from couchbase.cluster import Cluster, ClusterOptions
from couchbase.auth import PasswordAuthenticator
from donpinscheduler import DonpinScheduler, task
初始化 Couchbase 连接
cluster = Cluster(
‘couchbase://localhost’,
ClusterOptions(PasswordAuthenticator(‘username’, ‘password’))
)
query_service = cluster.query_indexes()
执行 SQL 脚本的通用函数
def execute_sql(file_path):
with open(file_path, ‘r’) as file:
sql = file.read()
result = query_service.query(sql)
print(f"Executed SQL from {file_path}: {result}")
任务定义
定义 SQL 脚本任务：
scheduler = DonpinScheduler()
@task(schedule=“0 * * * *”) # 每小时执行
def sync_data():
execute_sql(“/path/to/sync_data.sql”)
@task(schedule=“0 0 * * *”) # 每天午夜执行
def aggregate_data():
execute_sql(“/path/to/aggregate_data.sql”)
@task(schedule=“0 0 * * 0”) # 每周日午夜执行
def cleanup_expired_data():
execute_sql(“/path/to/cleanup_expired_data.sql”)
注册任务到调度器
scheduler.register(sync_data)
scheduler.register(aggregate_data)
scheduler.register(cleanup_expired_data)
- 启动调度器
运行调度器以管理任务：
if name == “main”:
scheduler.run()
功能亮点
- 定时调度
DonpinScheduler 使用类似 Cron 的调度表达式：
0 * * * *：每小时运行一次。
0 0 * * *：每天运行一次。
0 0 * * 0：每周运行一次。
- 任务状态和日志
DonpinScheduler 会记录每个任务的运行状态和日志，包括任务成功、失败以及重试次数。
- 失败重试机制
任务装饰器支持设置自动重试策略：
@task(schedule=“0 * * * *”, retries=3, retry_delay=60) # 失败后最多重试3次，间隔60秒
def sync_data():
execute_sql(“/path/to/sync_data.sql”)
- 分布式任务调度
DonpinScheduler 支持多实例运行，便于扩展为分布式任务调度系统。
总结
通过 DonpinScheduler，可以轻松实现 Couchbase SQL 脚本的管理和调度。它的轻量级设计非常适合中小型项目，并且通过定时、失败重试和日志记录功能，使 SQL 执行任务更加高效和可靠。