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 入門