Оператори Airflow

Вступ до Apache Airflow на Python

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 of connected tasks showing dependency order between them

Синтаксис зсуву бітів

>> і <<

task1() >> task2()

# task1 завершується перед стартом task2


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

# task1 завершується перед task2, а task2 — перед task3


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

# task1 і task2 можуть виконуватись паралельно, але обидва мають завершитись до старту task3

Приклад синтаксису зсуву бітів

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

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

Обидва мають завершитись до узгодження, але порядок не визначено

Давайте потренуємось!

Вступ до Apache Airflow на Python