A Comprehensive Guide to Airflow for Mac Workflow Management Platform
爱上MAC
10月26日
26
Airflow Mac Web Mac Airflow Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI Web UI
任务实例详情面板： 点击任意任务实例，可查看日志、重记成功/失败等操作选项、标记成功/1.2 Terminal
Airflow Terminal 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023 2023</strong
再打开一个新的终端窗口或标签页，启动调度器：airflow scheduler
此时，在浏览器中访问 http://localhost:8080，使用步骤 1 创建的用户名和密码登录，即可看到 Web UI。
步骤 3：创建你的第一个 DAG（工作流）
在 AIRFLOW_HOME 目录下，创建 dags 文件夹（如果不存在）：mkdir dags。
使用任何文本编辑器或 IDE（如 VS Code）在 dags 文件夹内创建一个 Python 文件，例如 my_first_dag.py。以下是下是下是为到
from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
with DAG(
dag_id='my_first_dag',
start_date=datetime(2023, 10, 1),
schedule_interval='@daily',
catchup=False
) as dag:
task1 = BashOperator(
task_id='print_date',
bash_command='date'
)
task2 = BashOperator(
task_id='echo_hello',
bash_command='echo "Hello from Airflow!"'
)
task1 >> task2 # 定义依赖关系：task1 关用作流）以激活（unpause）调度器会根据 schedule_interval 自动调度运行。
保存文件。Airflow 调度器会自动扫描 dags 文件夹并后 DAG。
步骤 4：在 Web UI 中触发下是上角的 “开关”按钮 （位于 DAG ID 右侧）以激活（unpause）该 DAG。激活后，调度器会根据 schedule_interval 自动调度运行。
若要立即手动触发一次运行，点击 “Trigger DAG” 或击 .py 或击 DAG 文件（如使用 Git 或击 DAG 可执行用到的辅助脚本。
在“树状图”或“图表”视图中，点击一个任务实例（如 queued, running, success, failed）。
步骤 5：查看结果与日志
在树状图或图表视图中，点击一个任务实例（如 print_date）的方块。
在弹出的菜单中，选择 “Log” 。
日志页面将显示该任务实例执行标准输出和错误信息，这是排查任务失败原因的主觀执行。
步骤 6：管理任务状态（常见操作）
重试失败任务， 在任务实例菜单中，选择 “Clear” 。这将清除该实例的状态，调度器会重新将其排队执行。标记任务为成功/失败， 在任务实例菜单中，选择 “Mark Success” 或 “Mark Failed” 。用于手动干预工作流状态，跳过成强制失败某个任务。
关于“导出”： Airflow 工作流本身以 Python 代码形式存在，因此“导出”即是对 DAG 文件（.py）和可能用到的辅助脚本。Airflow 不提供将整个工作流“导出”为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将整个工作流，图为单不提供将&#
Templated Task and XCom in Jinja2
A templated task in Jinja2 allows for dynamic execution dates, which can be useful for tasks that need to run at specific times or based on a schedule. The `ds` variable represents the execution date in Airflow.
task = BashOperator(
task_id='templated_task',
bash_command='echo "Execution date is {{ ds }}"', # ds represents the execution date in Airflow
dag=dag,
)
Using XCom for Data Exchange Between Tasks
XCom is a mechanism for exchanging data between tasks in Airflow. It allows a task to push data to another task, and the receiving task can then pull this data.
# Task A
task_a = PythonOperator(
task_id='push_data',
python_callable=lambda ti: ti.xcom_push(key='my_key', value='my_value'),
dag=dag,
)
# Task B
task_b = PythonOperator(
task_id='pull_data',
python_callable=lambda ti: print(ti.xcom_pull(task_ids='push_data', key='my_key')),
dag=dag,
)
Task Dependencies
Sequential execution: task1 >> task2 >> task3
Using set_upstream and set_downstream to define dependencies: task1.set_downstream(task2)
Using a list to define dependencies: [task1, task2] >> task3
DAGs in Airflow
A DAG (Directed Acyclic Graph) represents a workflow in Airflow. It can be defined in Python and executed using the Airflow CLI.
# Define a DAG
python ~/airflow_projects/my_workflow/dags/my_first_dag.py
# Execute a task within a DAG
airflow tasks test my_first_dag print_date 2023-10-01
Notes on DAGs in the Web UI
Note 1: Viewing DAGs in the Web UI
Accessing DAGs via the Web UI
Locate DAGs in the Web UI 确保文件在 AIRFLOW_HOME/dags
DAG 文件有 Python 语法错误： 在终端运行 python your_dag.py 检查错误。调度器未运行或未加载： 确保 airflow scheduler 进程正在运行）状态不执行中的DAG 被过滤： 检查 Web UI 左上角的“Hide paused DAGs”等过滤选项。
问题 2：修改与解决：
可能原因与解决：
执行器（Executor）状态不执行中的定义未更新 默认的 SequentialExecutor 一次只能运行一个任务。对于本在开刷标）。或者（增加调度器的解析退出资源不足： 检查调度器日志，看是否有关于资源（如 CPU/内存）的警告。
问题 3：修改 DAG 后，Web UI 中的定义未更新
解决： Airflow 默认每 30 秒扫描一次 DAG 文件。可以手动刷新 DAG：在 Web UI 的 DAG 详情页）皠为出的错误日志以获取具体原因。查看终端输出的错误日志以获取具体原因。可以使用
问题 4：Web 服务器或调度器进程意外退出
解决： 通常是由于端口冲突或数据库连接问题。查看终端输出的错误日志以获取具体原因。检查端口 8080 是否被占用，或尝试重启服务。查看终端输出的错误日志以获取具体原因。可以使用
