airflow 文档学习(二) - 概念

2023-05-16

1. 核心功能

1.1 DAGs

有向无环图

反映所涉及的task的依赖关系

注:搜索dag的时候,airflow只会关注同事包含"DAG"和"airflow"字样的py文件

1.2 scope

airflow将加载任何可以从DAG file中import的DAG对象,但是它们必须出现在globals()中,例如下面的文件,只有tag_1会被加载,tag_2只会出现在本地scope中

dag_1 = DAG('this_dag_will_be_discovered')
def my_function():
dag_2 = DAG('but_this_dag_will_not')
my_function()

1.3 Default Arguments

如果一个字典default_args被传给一个DAGs,它将会将其运用到所有的它的operator中。这使得复用default_args变得非常的方便

1.4 Context Manager

dags可以被当做一个管理器,去自动的分配新的operators给dag

1.5 Operators

dags描述的是怎么去跑一个工作流,operators决定实际做什么。

一个operator描述了在一个工作流中的单个task。operators经常但不总是原子的,这意味着他们可以独立存在而不需要去和别的operator分享资源。DAG将确保operators以正确的顺序运行,在这些依赖之外,operator通常是独立运行的,甚至他们肯能运行在不同的机器上。

这是个非常微妙的关键点:事实上,如果两个operator需要去共享一些信息,就像文件名或者一些小的数据,你应该去考虑将他们合并到一个operator中,如果上述情况确实是无法避免的,airflow有operator的交叉通信(xcom)在文档中有描述。

并且airflow提供了非常多的通用operator:

BashOperator 
PythonOperator 
EmailOperator
SimpleHttpOperator 等等

1.6 DAG Assignment

operator不用立马被分配给一个dag,但是,一旦operator被分配给了一个dag,它将无法被转移或者是取消分配。dag的分配在operator被创建之后可以被非常明确的完成,通过延期分配或者从其他operator推断的方式

例如下面的方式:

dag = DAG('my_dag', start_date=datetime(2016, 1, 1))
# sets the DAG explicitly
explicit_op = DummyOperator(task_id='op1', dag=dag)
# deferred DAG assignment
deferred_op = DummyOperator(task_id='op2')
deferred_op.dag = dag
# inferred DAG assignment (linked operators must be in the same DAG)
inferred_op = DummyOperator(task_id='op3')
inferred_op.set_upstream(deferred_op)

1.7 Bitshift Composition

在以前,operator的关系描述是通过set_upstream()和set_downstream()方法,在1.8 之后可以通过<>代替依赖方法。

1.8 Tasks

当一个operator被实例化之后,它就被称为是一个task。实例化在调用抽象operator时定义了具体的值,同时,参数化之后的task会称为dag的一个节点。

1.9 Task Instances

一个task实例代表一个task的特定运行,其特征在于dag、任务、和时间点的组合。

它拥有运行状态:running、success、failed、skipped、up for retry等

1.10 Workflows

通过组合dags和operators,你会创建TaskInstances,你可以创建复杂的工作流。

2. Additional Functionality

2.1 Hooks

Hooks是连接一些平台和数据库的接口,类似于 Hive, S3, MySQL, Postgres, HDFS, Pig。hooks尽可能的实现了通用接口,并且充当operator。还需要使用airflow.models.Connection 模型来检索主机名和身份认证信息,hooks将身份认证信息和代码放在管道之外,集中在元数据库中。

2.2 pools

一些系统会因为太多的进程不堪重负,airflow的pool可以被用作限制任意的task的运行。task可以在创建时通过参数指定存在的pool名称。

例如:

aggregate_db_message_job = BashOperator(
task_id='aggregate_db_message_job',
execution_timeout=timedelta(hours=3),
pool='ep_data_pipeline_db_msg_agg',
bash_command=aggregate_db_message_job_cmd,
dag=dag)
aggregate_db_message_job.set_upstream(wait_for_empty_queue)

pool中可以使用priority_weight参数去定义他的在队列中的权重,并且决定哪个task先行执行。

当容量被撑满时,task将会放入计划执行,一旦有容量,可运行的task和他们的状态将会被在前端展示,当插槽空闲,队列中的task将会基于权重进行排序执行。

2.3 Connections

外部系统的connection信息被存储在airflow的元数据库中,airflow的管道可以很简单地引用被集中管理的conn_id,无需另外进行操作。

当许多的connections被定义在同一个conn_id下,在这种情况下,当hooks使用get_connection方法时,airflow将随机选择一个connection,当重试时允许一些基本的负载均衡和容错。

一些hooks有默认的conn_id,当operators使用这个hook时不需要一个明确的conn_id。例如:PostgresHook的默认conn_id是postgres_default

2.4 Queues

当我们使用CeleryExecutor时,被塞入celery队列的task是可以被规定的,队列是BaseOperator的属性,所以任何task可以被分配到任何的队列中,而默认的队列环境是在配置文件的celery下的default_queue中配置的。

workers可以监听一个或多个队列中的task,当一个worker启动的时候(使用airflow worker命令),一个以逗号分隔的对列名可以被指定(airflow worker -q spark),这个worker将只会选择那些被连接到指定对列的task。

2.5 XComs

XComs使得tasks可以交换信息,允许更加细微的控制形式和分享状态。XComs原则上定义成key、value和timestamp,但是也可以跟踪一些属性,例如创建XCom的task/DAG。

XComs可以被pushed或者pulled,当一个task发送一个xcom,这个xcom是普遍可获得的。task可以被发送通过使用方法xcom_push(),此外,当一个task返回一个值时(不管是operators的execute()方法还是PythonOperators的python_callable方法),一个包含着返回值的xcom会自动发送。

tasks调用xcom_pull()去接受xcoms,可选择的根据key、task_ids、dag_id进行过滤。默认的,xcom_pull()在获得值时会根据keys自动筛选执行方法。

如果xcom_pull被传了一个task_id,则对应task最近一次的xcom值会被返回,如果一组task_ids传过去,会返回一组对应的xcom值

也可以直接在模板中获取xcom,例如:SELECT * FROM {{ task_instance.xcom_pull(task_ids='foo', key='table_name') }}

值得注意的是,xcom与variable非常相似,但它是专门用于任务之间的通信而不是全局设置

2.6 Variables

variables是一种传统的方式去存储和取回任意的内容或者是key-value形式的airflow的设置,它可以在前端界面、代码或者cli中进行增删改查的操作,当你定义管道代码,就可非常方便的使用,例如:

from airflow.models import Variable
foo = Variable.get("foo")
bar = Variable.get("bar", deserialize_json=True)

你可以使用variables在一个jinjia模板中:

echo {{ var.value.<variable_name> }}

2.7 Branching

有时候你需要你的工作流进行分支,或者是根据任意上游发生的条件走某条路。这其中一种实现方法就是使用BranchPythonOperator

BranchPythonOperator和PythonOperator十分相似,除了python会期望一个python_callable去返回一个task_id,返回的task_id跳过所有其他路径,python的返回函数的task_id必须直接引用BranchPythonOperator任务下游的任务。

注意,在BranchPythonOperator中使用depend_on_past = True下游的任务在逻辑上是不合理的,因为跳过状态总是导致依赖于过去成功的块任务。如果非要这样的话可以中间建立一个虚拟任务进行过度。

2.8 SubDAGs

2.9 SLAs

记录失败的过错的sla任务列表

2.10 Trigger Rules

虽然正常的工作流行为是在所有的直接上游任务成功之后触发的,但是airflow允许更为复杂的依赖项。

所有的operators有一个trigger_rule,用来定义生成的任务被触发的规则,trigger_rule的默认参数是all_success,以下为别的参数解释:

all_success: (default) 所有的父级任务成功
all_failed: 所有的父级任务失败,或者上游状态为失败
all_done: 所有的父级任务执行完成
one_failed: 至少一个失败,并且不会等待所有任务执行完成
one_success: 至少一个成功,并且不会等待所有任务执行完成
dummy: 依赖只是为了展示,随意触发

注意,这些可以与depends_on_past结合使用,当设置为true时,如果任务的先前计划未成功,则不会触发

2.11 Latest Run Only

标准工作流行为涉及为特定日期/时间范围内运行的一系列任务,但是,某些工作流执行的任务与运行时间无关,但是需要按计划运行,就像标准的cron作业一样,在这些情况下,暂停期间错过的回填运行作业会浪费cpu周期。

2.12 Zombies & Undeads

僵尸任务的特点是没有心跳(由工作定期发出)和数据库中的运行状态,当工作节点无法访问数据库的时候,airflow进程在外部被终止或者节点重启的时候,他们可能会发生。僵尸查杀由调度程序的进程定期执行。

undead进程的特点是存在进程和匹配的心跳,但是airflow不知道此任务在数据库中运行。这种不匹配通常在数据库状态发生改变的时候发生,最有可能是通过删除UI中的任务实例视图中的行,指示任务验证其作为心跳例程的一部分的状态,并在确定他们处于这种不死的状态时终止自身。

2.13 Cluster Policy

你本地airflow设置文件可以定义一个策略功能,该功能可以根据其他任务或DAG属性改变其任务属性。

2.14 Documentation & Notes

2.15 Jinja Templating

转载于:https://www.cnblogs.com/ZhangShY/p/10102210.html

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

airflow 文档学习(二) - 概念 的相关文章

随机推荐

  • cmake之install总结(指定安装文件和安装位置)

    背景 cmake不仅可以构建可执行文件 xff0c 还可以安装 使用CMake xff0c 我们可以使用install 命令指定安装规则 只需要指定安装位置和待安装的目标文件 xff0c 十分简单 以cmake官方教程为例 xff08 3
  • 工训备赛日志(二)——上位机与stm32之间的一种简易通信方式

    工训备赛日志 xff08 二 xff09 上位机与stm32之间的一种简易通信方式 一 摘要 上位机检测到物体之后 xff0c 要将信号传递给控制底层的STM32 xff0c 这里提出了一种基于读取电平和外部中断的简易通信方式 此方法占用五
  • 【2】中断方式和查询方式的区别

    1 中断方式 xff1a 接收数据时 xff0c MCU转入中断服务程序 xff0c 再处理接收到的数据 xff1b CMT2300A接收数据的中断 xff1a 检测RX FIFO TH 中断 xff0c 一旦有效表示FIFO 已经被填入预
  • Ubantu 安装到VMware详解

    想要在VMware中运行Linux系统 xff0c 那么就需要Linux系统安装到VMware虚拟机上面 在这里 xff0c 以把ubantu16 04安装到VMware虚拟机中为教程进行图文讲解 xff0c 共分为三个步骤 xff0c 分
  • 数字直流电源设计

    设计目标 xff1a 市电输入 开关频率50KHz 60V 20A 1200W输出 输出电压纹波0 2 输出电流纹波20 效率0 8 1 确定输入电压 经查阅 xff0c 我国市电电压标准 xff0c 220V单相供电时 xff0c 为额定
  • CAN报文格式

    CAN协议的报文传输过程中有 数据帧 远程帧 错误帧 过载帧和帧间隔 nbsp nbsp nbsp nbsp nbsp nbsp nbsp 1 数据帧 用于发送节点向接收节点传送数据的帧 nbsp nbsp nbsp nbsp nbsp n
  • FreeRTOS调度器启动

    nbsp 在真正启动调度器之前 先创建空闲任务 初始化一些变量 真正启动调度器的函数是xPortStartScheduler 启动调度器 void vTaskStartScheduler void BaseType t xReturn if
  • 安装novnc,并加入开机自启

    1 安装git工具 apt get install git y 2 下载novnc git clone https github com novnc noVNC 3 ls 查看 xff0c 已经下载完成 4 vim novnc sh把启动命
  • 【C++初阶11-模版进阶】解锁C++模版更多玩法

    今天 xff0c 带来模版进阶的讲解 主要讲解 xff1a 非类型模版参数模版的特化模版的分离编译 文中不足错漏之处望请斧正 xff01 非类型模版参数 模版参数分为类型模版参数和非类型模版参数 可以这么理解 xff1a 前者是为了泛化类型
  • Qt 获取当前时间

    1 获取带日期的时间 xff0c 使用QDateTime类 QDateTime current date time 61 QDateTime currentDateTime QString current date 61 current d
  • Qt 判断文件或文件夹是否存在及创建文件夹

    1 判断文件夹是不是存在 参数说明 xff1a QString fullPath 文件夹全路径 方法1 bool isDirExist QString fullPath QDir dir fullPath if dir exists ret
  • Qt阻塞延时与非阻塞延时(四种方法之个人笔记)

    功能描述 毫秒级非阻塞延时函数 参数 延时毫秒数 其他说明 已试验 可用 但CPU占用率较高 void VdpuMngUI msecSleep int msec QTime dieTime 61 QTime currentTime addM
  • 获取本机所有网卡的网卡名、网卡描述、网卡MAC地址、网卡IP、网卡类型等信息及网线是否插入状态

    使用windows sdk提供的API函数GetAdaptersInfo 可以获得本机所有网卡的网卡名 网卡描述 网卡MAC地址 网卡IP 网卡类型等信息 xff0c 并用IP ADAPTER INFO结构体存储 xff0c 使用GetIf
  • C++ GUI Programming with Qt4 Second Edition 之 附录C.1 Qt Jambi入门

    Qt Jambi入门 本节 xff0c 我们将开发一个简单的Java应用程序并显示如图C 1所示的窗口 除窗口标题之外 xff0c Jambi Find对话框与第二章中创建的Find对话框的外观和功能均相同 通过使用相同的例子 xff0c
  • WinRAR 5.5 简体中文版去广告弹窗方法

    WinRAR 5 5 简体中文版去广告弹窗方法 1 下载WinRAR 5 5WinRAR官网 xff08 http www rarlab com xff09 2 下载 Resource Hacker 3 用压缩软件打开WinRAR安装包 x
  • C++子类的构造函数后面加:冒号的作用

    在C 43 43 类的构造函数中经常会看到如下格式的写法 xff1a MyWindow MyWindow QWidget parent Qt WindowFlags flag QMainWindow parent flag 上述语句中单冒号
  • “真相”重要吗?

    这是一篇个人思索的鸡汤 xff0c 并不想表达什么观点 xff0c 只是因为有了要写的冲动才写的 言归正传 xff0c 我认为 真相很重要 周末回老家看了一岁半儿子 xff0c 从国庆送回老家已经一个半月了 周日下午我临走时 xff0c 他
  • TDD(测试驱动开发)

    今天看到的新的有意思的名词 xff0c 在此将其记录下来 一 TDD是什么 xff1f Test Drive Development xff0c 是敏捷开发中的一项核心实践和技术 也是一种设计方法论 xff0c 其基本思想就是 xff1a
  • c++ 常见的一些问题总结

    目录 目录 1 深拷贝 xff08 string xff09 深拷贝的实现 xff08 xff09 2 迭代器失效 xff08 vector insert xff0c earse xff09 3 头文件展开问题 4 vector 的深浅拷贝
  • airflow 文档学习(二) - 概念

    1 核心功能 1 1 DAGs 有向无环图 反映所涉及的task的依赖关系 注 xff1a 搜索dag的时候 xff0c airflow只会关注同事包含 34 DAG 34 和 34 airflow 34 字样的py文件 1 2 scope