编辑“︁Airflow BashOperator”︁

= Airflow BashOperator =

== 介绍 ==
'''BashOperator''' 是 Apache Airflow 的核心操作符之一，用于在 DAG（有向无环图）中执行 Bash 命令或脚本。它继承自 <code>BaseOperator</code>，允许用户直接在 Airflow 任务中运行 Shell 命令，适用于文件操作、环境配置、脚本调用等场景。BashOperator 是初学者快速上手 Airflow 的重要工具，同时也被高级用户广泛用于复杂任务编排。

== 基本语法 ==
BashOperator 的主要参数包括：
* <code>bash_command</code>: 要执行的 Bash 命令或脚本。
* <code>env</code> (可选): 设置环境变量。
* <code>append_env</code> (可选): 是否追加系统环境变量。
* <code>output_encoding</code> (可选): 输出编码格式。

基本使用示例：
<syntaxhighlight lang="python">
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="bash_operator_example",
    start_date=datetime(2023, 1, 1),
    schedule_interval="@daily",
) as dag:
    task = BashOperator(
        task_id="print_date",
        bash_command="date",
    )
</syntaxhighlight>

== 详细示例 ==

=== 简单命令执行 ===
以下示例展示如何输出当前日期并保存到文件：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="save_date",
    bash_command="date > /tmp/current_date.txt",
)
</syntaxhighlight>

=== 多命令执行 ===
使用 <code>&&</code> 或分号连接多个命令：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="multi_command",
    bash_command="echo 'Start' && date && echo 'End'",
)
</syntaxhighlight>

=== 使用环境变量 ===
通过 <code>env</code> 参数传递环境变量：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="env_vars",
    bash_command="echo $MY_VAR",
    env={"MY_VAR": "Hello Airflow!"},
)
</syntaxhighlight>

== 实际应用案例 ==

=== 数据管道预处理 ===
假设需要每天下载并处理 CSV 文件：
<syntaxhighlight lang="python">
process_data = BashOperator(
    task_id="process_data",
    bash_command="""
        wget https://example.com/data.csv -O /tmp/data.csv &&
        awk -F',' '{print $1}' /tmp/data.csv > /tmp/processed.csv
    """,
)
</syntaxhighlight>

=== 服务健康检查 ===
检查 Web 服务是否响应：
<syntaxhighlight lang="python">
health_check = BashOperator(
    task_id="health_check",
    bash_command="curl -sSf http://service:8080/health > /dev/null",
)
</syntaxhighlight>

== 错误处理 ==
BashOperator 默认会检查命令的退出码（非零表示失败）。可以通过 <code>skip_exit_code</code> 参数自定义忽略的退出码：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="ignore_error",
    bash_command="exit 1",  # 故意失败
    skip_exit_code=1,      # 忽略退出码1
)
</syntaxhighlight>

== 性能优化 ==

=== 模板化命令 ===
使用 Jinja 模板动态生成命令：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="templated_command",
    bash_command="echo 'Execution date is {{ ds }}'",
)
</syntaxhighlight>

=== 输出重定向 ===
建议将输出重定向到日志文件而非标准输出：
<syntaxhighlight lang="python">
task = BashOperator(
    task_id="log_output",
    bash_command="my_script.sh >> /var/log/airflow/my_task.log 2>&1",
)
</syntaxhighlight>

== 最佳实践 ==
1. 复杂逻辑应封装到单独的脚本文件中，而非直接写在 DAG 里
2. 使用绝对路径避免依赖环境变量
3. 敏感信息（如密码）应通过 Airflow Connections 管理
4. 为长时间运行的任务设置 <code>execution_timeout</code>

== 与其他操作符对比 ==
{| class="wikitable"
|-
! 操作符 !! 适用场景 !! 特点
|-
| BashOperator || 执行 Shell 命令 || 轻量级，无需额外依赖
|-
| PythonOperator || 执行 Python 函数 || 适合复杂逻辑
|-
| DockerOperator || 容器化任务 || 环境隔离
|}

== 高级用法 ==
=== 动态命令生成 ===
结合 Python 变量生成命令：
<syntaxhighlight lang="python">
def generate_command(**context):
    return f"process_data --date {context['ds']}"

task = BashOperator(
    task_id="dynamic_command",
    bash_command=generate_command,
)
</syntaxhighlight>

=== 任务流程图 ===
<mermaid>
graph LR
    A[BashOperator: 下载数据] --> B[BashOperator: 处理数据]
    B --> C[BashOperator: 上传结果]
</mermaid>

== 数学公式示例 ==
当需要计算任务执行时间时，可能用到公式：
<math>
T_{total} = \sum_{i=1}^{n} (T_{task_i} + T_{overhead})
</math>

== 总结 ==
BashOperator 是 Airflow 中最灵活的操作符之一，适合快速实现各种系统级操作。通过合理设计命令和错误处理机制，可以构建健壮的数据管道。对于更复杂的逻辑，建议结合 PythonOperator 或自定义 Operator 使用。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow DAG开发]]