AI 助理
文档备案控制台
开发者社区 云计算 文章 正文

[AirFlow]AirFlow使用指南三 第一个DAG示例

2017-08-03 3317
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/SunnyYoona/article/details/76615699 经过前两...
版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/SunnyYoona/article/details/76615699

经过前两篇文章的简单介绍之后，我们安装了自己的AirFlow以及简单了解了DAG的定义文件．现在我们要实现自己的一个DAG．

1. 启动Web服务器

使用如下命令启用:

airflow webserver

现在可以通过将浏览器导航到启动Airflow的主机上的8080端口来访问Airflow UI，例如：http://localhost:8080/admin/

img

备注

Airflow附带了许多示例DAG。 请注意，在你自己的`dags_folder`中至少有一个DAG定义文件之前，这些示例可能无法正常工作。你可以通过更改`airflow.cfg`中的`load_examples`设置来隐藏示例DAG。

2. 第一个AirFlow DAG

现在一切都准备好了，我们开始写一些代码，来实现我们的第一个DAG。 我们将首先创建一个Hello World工作流程，其中除了向日志发送"Hello world！"之外什么都不做。

创建你的dags_folder，那就是你的DAG定义文件存储目录---$AIRFLOW_HOME/dags。在该目录中创建一个名为hello_world.py的文件。

AIRFLOW_HOME
├── airflow.cfg
├── airflow.db
├── airflow-webserver.pid
├── dags
│   ├── hello_world.py
│   └── hello_world.pyc
└── unittests.cfg

将以下代码添加到dags/hello_world.py中:

# -*- coding: utf-8 -*-

import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta

#-------------------------------------------------------------------------------
# these args will get passed on to each operator
# you can override them on a per-task basis during operator initialization

default_args = {
    'owner': 'jifeng.si',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['1203745031@qq.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
    # 'wait_for_downstream': False,
    # 'dag': dag,
    # 'adhoc':False,
    # 'sla': timedelta(hours=2),
    # 'execution_timeout': timedelta(seconds=300),
    # 'on_failure_callback': some_function,
    # 'on_success_callback': some_other_function,
    # 'on_retry_callback': another_function,
    # 'trigger_rule': u'all_success'
}

#-------------------------------------------------------------------------------
# dag

dag = DAG(
    'example_hello_world_dag',
    default_args=default_args,
    description='my first DAG',
    schedule_interval=timedelta(days=1))

#-------------------------------------------------------------------------------
# first operator

date_operator = BashOperator(
    task_id='date_task',
    bash_command='date',
    dag=dag)

#-------------------------------------------------------------------------------
# second operator

sleep_operator = BashOperator(
    task_id='sleep_task',
    depends_on_past=False,
    bash_command='sleep 5',
    dag=dag)

#-------------------------------------------------------------------------------
# third operator

def print_hello():
    return 'Hello world!'

hello_operator = PythonOperator(
    task_id='hello_task',
    python_callable=print_hello,
    dag=dag)

#-------------------------------------------------------------------------------
# dependencies

sleep_operator.set_upstream(date_operator)
hello_operator.set_upstream(date_operator)

该文件创建一个简单的DAG，只有三个运算符，两个BaseOperator(一个打印日期一个休眠5秒)，另一个为PythonOperator在执行任务时调用print_hello函数。

3. 测试代码

使用如下命令测试一下我们写的代码的正确性:

python ~/opt/airflow/dags/hello_world.py

如果你的脚本没有抛出异常，这意味着你代码中没有错误，并且你的Airflow环境是健全的。

下面测试一下我们的DAG中的Task．使用如下命令查看我们example_hello_world_dagDAG下有什么Task:

xiaosi@yoona:~$ airflow list_tasks example_hello_world_dag
[2017-08-03 11:41:57,097] {__init__.py:57} INFO - Using executor SequentialExecutor
[2017-08-03 11:41:57,220] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python2.7/lib2to3/Grammar.txt
[2017-08-03 11:41:57,241] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python2.7/lib2to3/PatternGrammar.txt
[2017-08-03 11:41:57,490] {models.py:167} INFO - Filling up the DagBag from /home/xiaosi/opt/airflow/dags
date_task
hello_task
sleep_task

可以看到我们有三个Task：

date_task
hello_task
sleep_task

下面分别测试一下这几个Task:

(1) 测试date_task

xiaosi@yoona:~$ airflow test example_hello_world_dag date_task 20170803
...
--------------------------------------------------------------------------------
Starting attempt 1 of 2
--------------------------------------------------------------------------------

[2017-08-03 11:44:02,248] {models.py:1342} INFO - Executing <Task(BashOperator): date_task> on 2017-08-03 00:00:00
[2017-08-03 11:44:02,258] {bash_operator.py:71} INFO - tmp dir root location:
/tmp
[2017-08-03 11:44:02,259] {bash_operator.py:80} INFO - Temporary script location :/tmp/airflowtmpxh6da9//tmp/airflowtmpxh6da9/date_tasktQQB0V
[2017-08-03 11:44:02,259] {bash_operator.py:81} INFO - Running command: date
[2017-08-03 11:44:02,264] {bash_operator.py:90} INFO - Output:
[2017-08-03 11:44:02,265] {bash_operator.py:94} INFO - 2017年 08月 03日 星期四 11:44:02 CST
[2017-08-03 11:44:02,266] {bash_operator.py:97} INFO - Command exited with return code 0

(2) 测试hello_task

xiaosi@yoona:~$ airflow test example_hello_world_dag hello_task 20170803
...
--------------------------------------------------------------------------------
Starting attempt 1 of 2
--------------------------------------------------------------------------------

[2017-08-03 11:45:29,546] {models.py:1342} INFO - Executing <Task(PythonOperator): hello_task> on 2017-08-03 00:00:00
[2017-08-03 11:45:29,551] {python_operator.py:81} INFO - Done. Returned value was: Hello world!

(3) 测试sleep_task

xiaosi@yoona:~$ airflow test example_hello_world_dag sleep_task 20170803
...
--------------------------------------------------------------------------------
Starting attempt 1 of 2
--------------------------------------------------------------------------------

[2017-08-03 11:46:23,970] {models.py:1342} INFO - Executing <Task(BashOperator): sleep_task> on 2017-08-03 00:00:00
[2017-08-03 11:46:23,981] {bash_operator.py:71} INFO - tmp dir root location:
/tmp
[2017-08-03 11:46:23,983] {bash_operator.py:80} INFO - Temporary script location :/tmp/airflowtmpsuamQx//tmp/airflowtmpsuamQx/sleep_taskuKYlrh
[2017-08-03 11:46:23,983] {bash_operator.py:81} INFO - Running command: sleep 5
[2017-08-03 11:46:23,988] {bash_operator.py:90} INFO - Output:
[2017-08-03 11:46:28,990] {bash_operator.py:97} INFO - Command exited with return code 0

如果没有问题，我们就可以运行我们的DAG了．

4. 运行DAG

为了运行你的DAG，打开另一个终端，并通过如下命令来启动Airflow调度程序:

airflow scheduler

备注

调度程序将发送任务进行执行。默认Airflow设置依赖于一个名为`SequentialExecutor`的执行器，它由调度程序自动启动。在生产中，你可以使用更强大的执行器，如`CeleryExecutor`。

当你在浏览器中重新加载Airflow UI时，应该会在Airflow UI中看到你的hello_world DAG。

img

为了启动DAG Run，首先打开工作流(off键)，然后单击Trigger Dag按钮(Links 第一个按钮)，最后单击Graph View按钮(Links 第三个按钮)以查看运行进度:

img

你可以重新加载图形视图，直到两个任务达到状态成功。完成后，你可以单击hello_task，然后单击View Log查看日志。如果一切都按预期工作，日志应该显示一些行，其中之一是这样的：

...
[2017-08-03 09:46:43,213] {base_task_runner.py:95} INFO - Subtask: --------------------------------------------------------------------------------
[2017-08-03 09:46:43,213] {base_task_runner.py:95} INFO - Subtask: Starting attempt 1 of 2
[2017-08-03 09:46:43,214] {base_task_runner.py:95} INFO - Subtask: --------------------------------------------------------------------------------
[2017-08-03 09:46:43,214] {base_task_runner.py:95} INFO - Subtask:
[2017-08-03 09:46:43,228] {base_task_runner.py:95} INFO - Subtask: [2017-08-03 09:46:43,228] {models.py:1342} INFO - Executing <Task(PythonOperator): hello_task> on 2017-08-03 09:45:49.070859
[2017-08-03 09:46:43,236] {base_task_runner.py:95} INFO - Subtask: [2017-08-03 09:46:43,235] {python_operator.py:81} INFO - Done. Returned value was: Hello world!
[2017-08-03 09:46:47,378] {jobs.py:2083} INFO - Task exited with return code 0
文章标签：
调度
Python
文件存储
sjf0115
目录
相关文章
estar.lee
|
10月前
|
JSON API PHP
免费ICP备案查询API接口详细教程
本文介绍“接口盒子”提供的免费ICP备案查询API，支持通过域名快速获取备案信息，如主办单位、备案号、审核时间等。基于离线库查询，适合非实时场景，开发者可免费调用并集成至应用，需注册获取ID和KEY，另有付费优享版供高并发需求使用。
estar.lee
2476 2
null_is_null
|
机器学习/深度学习 人工智能 PyTorch
GPT为定制AI应用工程师转型第一周学习计划
本计划帮助开发者快速入门AI领域，首周涵盖AI基础理论、Python编程及PyTorch实战。前两天学习机器学习、深度学习与Transformer核心概念，掌握LLM工作原理。第三至四天快速掌握Python语法与Jupyter使用，完成基础编程任务。第五至七天学习PyTorch，动手训练MNIST手写识别模型，理解Tensor操作与神经网络构建。
null_is_null
731 0
三桥君
|
人工智能 开发框架 搜索推荐
AI Agent构建强大外部工具调用能力不足，MCP Server怎样应对？MCP Serve在企业级Agent系统中的关键意义
本文AI产品专家三桥君探讨了MCP Server在企业级AI Agent系统中的关键作用，通过标准化工具接口实现AI与外部服务的无缝集成。三桥君重点阐述了分布式系统中的会话管理、状态持久化等实践方案，强调MCP Server在降低AI决策风险、提升系统可靠性方面的企业价值，为AI产品经理提供了架构设计与优化策略的实践指导。
三桥君
904 0
霍格沃兹测试开发学社
|
11月前
|
人工智能 自然语言处理 安全
MCP 打造 AI Agent 的 5 大最佳实践
面对流量激增、系统告急，单一渠道AI Agent难堪重负？问题不在能力，在架构！MCP（多渠道平台）作为智能体“神经中枢”，可统一调度Web、微信、钉钉等多渠道任务，实现高并发、可扩展的自动化。本文详解企业级MCP构建AI Agent的五大最佳实践：模块拆分、工具选型、设计模式、上下文管理与安全可控，助你快速落地高效、稳定的智能系统。
霍格沃兹测试开发学社
1311 0
1613249545098970
|
人工智能 数据可视化 数据处理
2025低代码前瞻：平台赋能的无限可能
低代码平台正成为企业数字化转型的核心工具，2025年将迎来新的高峰。其核心功能包括可视化开发、智能引擎、模型驱动、数据处理增强及AI深度融合等，助力高效协作与灵活扩展。通过降低技术门槛、提升开发效率和智能化水平，低代码将赋能企业实现更快的创新和更高的竞争力，推动数字化生态的全面发展。
1613249545098970
981 31
modelscope
|
物联网 异构计算
高效部署通义万相Wan2.1：使用Gradio搭建WebUI体验实战
随着通义万相Wan2.1 在社区的热度持续上涨，魔搭创空间的体验Demo（https://modelscope.cn/studios/Wan-AI/Wan-2.1）已经排起长队。
modelscope
2920 12
estar.lee
|
API
icp备案查询免费API接口教程
该接口用于查询指定域名的ICP备案信息，支持POST或GET请求方式。请求时需提供用户ID、用户KEY及待查询的域名，可选参数为查询通道。响应中包含状态码、消息内容、备案号、备案主体、域名及审核时间等信息。示例中提供了GET和POST请求方式及返回数据样例。
estar.lee
2559 4
子午s
|
存储 Linux 调度
高效定时任务处理：深入学习Python中APScheduler库的奥秘
高效定时任务处理：深入学习Python中APScheduler库的奥秘
子午s
17851 7
高效定时任务处理：深入学习Python中APScheduler库的奥秘
ALex_zry
|
Linux 开发工具
Linux yum 使用时提示 获取 GPG 密钥失败Couldn‘t open file RPM-GPG-KEY-EPEL-7
Linux yum 使用时提示 获取 GPG 密钥失败Couldn‘t open file RPM-GPG-KEY-EPEL-7
ALex_zry
697 2
VipSoft
|
Java Spring
Spring Boot Admin 查看 Client 日志
Spring Boot Admin 查看 Client 日志
VipSoft
705 0

热门文章

最新文章

  • 1
    Z-library 稳定访问指南
  • 2
    《量化金融R语言初级教程》一1.4　波动率建模
  • 3
    拒绝版权流氓！阿里巴巴重磅发布免费商用字体
  • 4
    《熬夜整理》保姆级系列教程-玩转Wireshark抓包神器教程(2)-Wireshark在Windows系统上安装部署
  • 5
    Layui上传文件时choose事件只触发一次的问题（两种解决方案+最终解决方案源码）
  • 6
    机器学习系列8 基于Python构建Web应用以使用机器学习模型
  • 7
    如何用纯 CSS 创作一个货车 loader
  • 8
    Autodesk 招人了，开发顾问，感兴趣的或者有推荐的人扔简历过来啊
  • 9
    WPF中Timer与DispatcherTimer类的区别
  • 10
    Vim 使用设置
  • 1
    Paimon把LSM搬进湖里的得与失
    30
  • 2
    我的推理引擎只有 1.35 MB，却塞进了五类模型 + 国密安全栈，还要跟 llama.cpp 打擂台
    32
  • 3
    github一个神级skill：把技术概念做成一台真机器或者任何的科普视频
    47
  • 4
    私有知识库工程化建设方法：一套五阶段流程值得采用
    48
  • 5
    阿里云38元一年服务器适合做什么？200M带宽38元服务器可以做哪些事？
    23
  • 6
    用 AI 开发 260 个在线工具：3Q工具箱 的工程实践与方法论
    26
  • 7
    阿里云万小智轻量版和标准版有什么区别？如何选择？
    23
  • 8
    阿里云 CLB / ALB 证书更新后，为什么网站还在用旧证书？
    28
  • 9
    阿里云万小智AI建站版本如何选择？标准版和高级版有什么区别？
    29
  • 10
    260+异构同步链路实战：KFS如何实现DDL语义化自动翻译，破解CDC同步痛点
    44

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！