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')


# Adds:
#   Hi DataCamp - This goes in the logs!
# to the Airflow logs

@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 completes before task2 starts


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

# task1 completes before task2 and task2 completes before task3


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

# task1 and task2 can run together but both must complete before task3 runs

位移運算子語法範例

# Download sales data before reconciling
download_sales_data() >> reconcile()

# Download inventory data before reconciling
download_inventory_data() >> reconcile()

兩者都必須完成後才對帳，但未指定先後順序

一起來練習吧！

Python 中的 Apache Airflow 入門