编辑“︁Airflow高级配置选项”︁

= Airflow高级配置选项 =

== 介绍 ==
'''Airflow高级配置选项'''是指Apache Airflow工作流管理系统中用于精细控制调度行为、资源分配和系统性能的可配置参数。这些选项通常通过`airflow.cfg`配置文件或环境变量进行设置，适用于需要优化生产环境、实现复杂调度逻辑或集成特殊后端服务的场景。本指南将系统讲解关键配置项及其应用场景。

== 核心配置类别 ==
=== 1. 调度器优化 ===
{| class="wikitable"
|+ 主要调度器配置
! 参数 !! 默认值 !! 说明
|-
| `scheduler.min_file_process_interval` || 300 || 同一DAG文件两次解析的最小间隔（秒）
|-
| `scheduler.dag_dir_list_interval` || 300 || DAG目录刷新间隔（秒）
|-
| `scheduler.max_threads` || 2 || 调度器最大线程数
|}

'''示例配置'''（提升高频DAG处理能力）：
<syntaxhighlight lang="ini">
[scheduler]
min_file_process_interval = 60
max_threads = 4
</syntaxhighlight>

=== 2. 执行器配置 ===
支持多种执行器类型：
* '''LocalExecutor'''：单机执行
* '''CeleryExecutor'''：分布式执行
* '''KubernetesExecutor'''：容器化执行

<mermaid>
graph LR
    A[调度器] -->|任务队列| B[LocalExecutor]
    A -->|消息队列| C[CeleryExecutor]
    A -->|API调用| D[KubernetesExecutor]
</mermaid>

=== 3. 并行度控制 ===
关键参数：
* `parallelism`：全局并行任务数上限
* `dag_concurrency`：单个DAG的最大并发任务数
* `max_active_runs_per_dag`：单个DAG的最大活跃运行实例数

数学关系：
<math>
\text{实际并发} = \min(\text{parallelism}, \text{dag\_concurrency} \times \text{max\_active\_runs\_per\_dag})
</math>

== 实战案例 ==
=== 场景：电商数据处理流水线 ===
需求：每日处理千万级订单数据，需保证：
1. 资源隔离（ETL与报表任务分开）
2. 失败任务自动重试
3. 高峰时段动态扩容

'''解决方案配置'''：
<syntaxhighlight lang="python">
# 在DAG定义中设置并发参数
default_args = {
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'pool': 'etl_pool'  # 资源隔离池
}

dag = DAG(
    'ecommerce_pipeline',
    max_active_runs=2,
    concurrency=6,
    pool='etl_pool'
)
</syntaxhighlight>

'''airflow.cfg对应配置'''：
<syntaxhighlight lang="ini">
[core]
parallelism = 32
dag_concurrency = 16

[celery]
worker_concurrency = 8  # 每个worker的并发数
</syntaxhighlight>

== 高级特性详解 ==
=== 动态资源配置 ===
使用`executor_config`实现任务级资源分配：
<syntaxhighlight lang="python">
task = PythonOperator(
    task_id='data_processing',
    executor_config={
        "KubernetesExecutor": {
            "request_memory": "512Mi",
            "limit_memory": "1Gi"
        }
    }
)
</syntaxhighlight>

=== 敏感信息加密 ===
通过`Fernet`密钥加密连接密码：
1. 生成密钥：
<syntaxhighlight lang="bash">
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key())"
</syntaxhighlight>

2. 配置`airflow.cfg`：
<syntaxhighlight lang="ini">
[core]
fernet_key = your_generated_key_here
</syntaxhighlight>

== 性能调优建议 ==
* '''监控指标'''：
  * DagFileProcessorManager 的CPU使用率
  * Scheduler 的任务排队时间
  * Executor 的任务执行延迟

* '''推荐调整步骤'''：
  1. 基准测试确定当前瓶颈
  2. 逐步调整线程/进程参数
  3. 使用`airflow config list`验证配置

== 常见问题 ==
'''Q：如何验证配置生效？'''
A：使用命令：
<syntaxhighlight lang="bash">
airflow config get-value core parallelism
</syntaxhighlight>

'''Q：生产环境推荐配置？'''
A：建议基础配置：
* `parallelism = 2 * CPU核心数`
* `worker_concurrency = CPU核心数`
* 启用`worker_precheck`防止资源过载

== 总结 ==
Airflow的高级配置选项提供了对工作流执行的细粒度控制，通过合理调整这些参数可以显著提升系统稳定性和资源利用率。建议在生产环境部署前进行充分测试，并持续监控关键性能指标。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow高级特性]]