假设场景如下：

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

脚本包括：

同步数据到 Couchbase 集群。

执行数据聚合查询。

清理过期数据。

要求：

支持任务依赖管理。

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

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

使用 Airflow 实现

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

1. 创建 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();

2. 安装 Couchbase 的 Python 客户端

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

pip install couchbase

3. 定义 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

4. 部署 DAG 到 Airflow

将脚本保存为 couchbase_workflow.py 并放置到 Airflow 的 DAG 文件夹中（通常是 /airflow/dags）。

确保 Airflow 服务正常运行：

airflow webserver
airflow scheduler

登录到 Airflow Web 界面，启用并监控 couchbase_sql_workflow。

5. 优势分析

任务调度：通过 schedule_interval 定时调度任务，支持灵活的 Cron 表达式。

任务依赖管理：通过 >> 定义任务依赖，确保顺序执行。

重试机制：默认支持失败后的自动重试。

可观察性：Airflow 提供任务状态跟踪和日志记录，方便调试和监控。

6. 扩展优化

参数化 SQL：可在 SQL 中加入参数，通过 PythonOperator 动态替换。

自定义连接器：使用 Airflow 的 Hook 构建更灵活的 Couchbase 连接器。

错误处理：在 Python 函数中捕获异常并记录到外部系统（如日志系统或监控平台）。