Conceitos centrais do Airflow

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Conheça seu instrutor

Foto de perfil de Volker Janz

Volker Janz

Senior Developer Advocate, Astronomer
14+ anos como engenheiro de dados em games
Trabalha com Airflow desde a versão 1.x
Palestrante, mentor e autor da newsletter Data Engineer Things

O que você vai construir

Criar Dags com a TaskFlow API
Montar workflows dinâmicos com task mapping e agendamento por assets
Tratar falhas com retries e callbacks
Executar cargas de trabalho em SQL no Airflow

Visualização do conteúdo de cada capítulo

Antes de começar

À vontade com Dags, tasks e operators
Familiar com noções de agendamento

Introdução ao Airflow - página do curso

Revisão rápida

from airflow.sdk import dag, task

@dag
def star_wars_dag():


  @task
  def get_star_wars_person():
    import requests
    return requests.get("https://swapi.dev/api/people/1/").json()


  @task.bash
  def print_name(person):
    return f"echo '{person['name']}'"


  person = get_star_wars_person()
  print_name(person)

star_wars_dag()

Um Dag é um conjunto de tasks com dependências
Tasks são unidades de trabalho
Operators/decorators definem o que cada task faz
Dependências definem a ordem de execução

Dag simples

Arquitetura do Airflow

Arquitetura do Airflow 3

Orquestração: Scheduler, Dag Processor
Execução: Worker, Triggerer
Interface e armazenamento: API Server, Metadata DB

Abordagens de agendamento

# Sem execuções automáticas: disparo manual (padrão)
@dag(schedule=None)
def my_pipeline(): ...


# Baseado em tempo: executa todo dia às 6h
@dag(schedule="0 6 * * *")
def daily_pipeline(): ...


# Ciente de dados: executa quando um Asset é atualizado
@dag(schedule=[Asset("my_asset")])
def downstream_pipeline(): ...

Duas formas de escrever Dags

Operadores clássicos

extract = PythonOperator(
    task_id="extract",
    python_callable=extract_fn)
extract >> transform

Escolha quando não houver decorators disponíveis

TaskFlow API

@task
def extract():
    return {"users": 150}

data = extract()
transform(data)

Decorators Python simples
Menos código boilerplate
Pode ser combinada com operadores clássicos

Exercícios neste curso

Screenshot do exercício no IDE

Exercícios no IDE: edite arquivos .py reais
Clique em "Run this file" ou use python3 filename.py
dag.test() executa o Dag inteiro em um processo

Vamos praticar!

Criando Pipelines de Dados com Airflow