Airflow 运算符

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

运算符

表示工作流中的单个任务
通常独立运行
一般不共享信息
使用不同运算符执行不同任务

@dag(
  dag_id="Example_Dag"
)
def example_dag():
  @task
  def task1():
    return "The result from task1"

  task1()

example_dag()

@task（PythonOperator）

执行一个 Python 函数
任何 Python 函数都可用 @task 装饰
可与其他函数/任务传递数据

from airflow.sdk import task

@task
def printme():
    print("This goes in the logs!")

printme()

@task 参数

可像普通 Python 函数一样向任务/函数传参

@task
def printme(name: str):
    print(f"Hi {name} - This goes in the logs!")


printme(name='DataCamp')


# 日志将增加：
#   Hi DataCamp - This goes in the logs!
# 位于 Airflow 日志中

@task.bash（BashOperator）

@task.bash
def bash_example():
  return "echo 'Example!'"

bash_example()

@task.bash
def run_cleanup():
  return "runcleanup.sh"

run_cleanup()

执行指定的 Bash 命令或脚本
在临时目录中运行该命令
可为命令指定环境变量

任务依赖

每个 Dag 有一组待完成的任务
任务依赖决定运行顺序
有多种方式声明依赖

显示任务依赖顺序的 Airflow Dag 图

位移语法

>> 和 <<

task1() >> task2()

# task1 完成后才开始 task2


task1() >> task2() >> task3()

# task1 完成于 task2 之前，task2 完成于 task3 之前


task1() >> task3()
task2() >> task3()

# task1 与 task2 可并行，但均需完成后才运行 task3

位移语法示例

# 下载销售数据后再对账
download_sales_data() >> reconcile()

# 下载库存数据后再对账
download_inventory_data() >> reconcile()

二者都需完成后再对账，先后次序未指定

¡Vamos a practicar!

Python 中的 Apache Airflow 入门