Operadores do Airflow

Introdução ao Apache Airflow em Python

Mike Metzger

Data Engineer

Operadores

Representam uma única tarefa no fluxo
Roda de forma independente (geralmente)
Normalmente não compartilham dados
Vários operadores para diferentes tarefas

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

  task1()

example_dag()

@task (PythonOperator)

Executa uma função Python
Qualquer função Python pode usar @task
Pode trocar dados com outras funções/tarefas

from airflow.sdk import task

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

printme()

Argumentos de @task

Dá para passar argumentos como numa função Python comum

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


printme(name='DataCamp')


# Adiciona:
#   Hi DataCamp - This goes in the logs!
# aos logs do Airflow

@task.bash (BashOperator)

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

bash_example()

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

run_cleanup()

Executa um comando ou script Bash
Roda o comando num diretório temporário
Permite definir variáveis de ambiente

Dependências entre tarefas

Cada DAG tem um conjunto de tarefas
Dependências definem a ordem de execução
Há vários jeitos de definir dependências

DAG do Airflow com tarefas conectadas mostrando a ordem de dependência entre elas

Sintaxe de bitshift

>> e <<

task1() >> task2()

# task1 termina antes de task2 começar


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

# task1 termina antes de task2 e task2 termina antes de task3


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

# task1 e task2 podem rodar juntas, mas ambas devem terminar antes de task3

Exemplo de sintaxe de bitshift

# Baixe os dados de vendas antes de reconciliar
download_sales_data() >> reconcile()

# Baixe os dados de inventário antes de reconciliar
download_inventory_data() >> reconcile()

Ambas devem terminar antes de reconciliar; a ordem entre elas não é definida

Vamos praticar!

Introdução ao Apache Airflow em Python