当前位置：首页 > article >正文

调度系统：使用 Apache Airflow 管理和调度 Couchbase SQL 脚本的实际例子

article 2025/1/13 10:06:18

假设场景如下：

每天定时执行一组 Couchbase SQL 脚本，用于数据同步、聚合和清洗。

脚本包括：

同步数据到 Couchbase 集群。

执行数据聚合查询。

清理过期数据。

要求：

支持任务依赖管理。

提供任务失败后的重试机制。

支持日志和运行状态的监控。

使用 Airflow 实现

Airflow 提供了强大的调度和任务依赖管理能力，可以将上述流程定义为一个 DAG（有向无环图）。

创建 Couchbase SQL 脚本

创建三个 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();

安装 Couchbase 的 Python 客户端

通过 pip 安装所需的 Couchbase 依赖：

pip install couchbase

定义 Airflow DAG 和任务

couchbase_workflow.py：

from airflow import DAG

from airflow.operators.python import PythonOperator

from datetime import datetime, timedelta

from couchbase.cluster import Cluster, ClusterOptions

from couchbase_core.cluster import PasswordAuthenticator

Couchbase 连接函数

def execute_couchbase_query(sql_file_path):
# 连接 Couchbase 集群
cluster = Cluster(
‘couchbase://localhost’,
ClusterOptions(PasswordAuthenticator(‘username’, ‘password’))
)
bucket = cluster.bucket(‘bucket-name’)
query_service = cluster.query_indexes()

# 读取并执行 SQL 脚本
with open(sql_file_path, 'r') as file:
    query = file.read()
result = query_service.query(query)
print(f"Executed query from {sql_file_path}: {result}")

定义默认参数

default_args = {
‘owner’: ‘admin’,
‘depends_on_past’: False,
‘email_on_failure’: True,
‘email’: [‘admin@example.com’],
‘retries’: 2,
‘retry_delay’: timedelta(minutes=5),
}

定义 DAG

with DAG(
dag_id=‘couchbase_sql_workflow’,
default_args=default_args,
description=‘A workflow to execute Couchbase SQL scripts’,
schedule_interval=‘0 3 * * *’, # 每天凌晨 3 点运行
start_date=datetime(2024, 1, 1),
catchup=False,
tags=[‘couchbase’, ‘sql’],
) as dag:

# 任务 1: 同步数据
sync_data_task = PythonOperator(
    task_id='sync_data',
    python_callable=execute_couchbase_query,
    op_args=['/path/to/sql_scripts/sync_data.sql']
)

# 任务 2: 数据聚合
aggregate_data_task = PythonOperator(
    task_id='aggregate_data',
    python_callable=execute_couchbase_query,
    op_args=['/path/to/sql_scripts/aggregate_data.sql']
)

# 任务 3: 清理过期数据
cleanup_data_task = PythonOperator(
    task_id='cleanup_data',
    python_callable=execute_couchbase_query,
    op_args=['/path/to/sql_scripts/cleanup_expired_data.sql']
)

# 定义任务依赖
sync_data_task >> aggregate_data_task >> cleanup_data_task