编辑“︁Airflow与Spark集成”︁（章节）

== 实际案例 ==
以下是一个真实的应用场景，展示如何用 Airflow 调度 Spark 作业进行数据清洗和分析。

=== 场景描述 ===
一个电商公司需要每天处理用户行为日志，计算以下指标：
1. 每日活跃用户数（DAU）
2. 用户购买转化率
3. 热门商品排行榜

=== DAG 设计 ===
<mermaid>
graph TD
    A[开始] --> B[下载日志数据]
    B --> C[运行Spark清洗作业]
    C --> D[运行Spark分析作业]
    D --> E[存储结果到数据库]
    E --> F[发送邮件通知]
</mermaid>

=== 代码实现 ===
<syntaxhighlight lang="python">
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.operators.email import EmailOperator
from datetime import datetime

dag = DAG(
    'ecommerce_analytics',
    schedule_interval='@daily',
    start_date=datetime(2023, 1, 1),
)

clean_data = SparkSubmitOperator(
    task_id='clean_logs',
    application='jobs/clean_logs.py',
    conn_id='spark_cluster',
    dag=dag,
)

analyze_metrics = SparkSubmitOperator(
    task_id='compute_metrics',
    application='jobs/analyze_metrics.py',
    conn_id='spark_cluster',
    dag=dag,
)

send_report = EmailOperator(
    task_id='send_daily_report',
    to='analytics-team@example.com',
    subject='Daily E-commerce Report',
    html_content='The daily metrics have been processed.',
    dag=dag,
)

clean_data >> analyze_metrics >> send_report
</syntaxhighlight>