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

두 태스크 모두 완료되어야 reconcile이 실행되며, 순서는 지정되지 않습니다

연습해 보겠습니다!

Python으로 배우는 Apache Airflow 입문