Neste capítulo, você terá uma introdução completa aos componentes do Apache Airflow e aprenderá como e por que usá-los.

Introdução ao Apache Airflow

Testando uma tarefa no Airflow

Examinando os comandos do Airflow

DAGs do Airflow

Definindo um DAG simples

Trabalhando com DAGs e o shell do Airflow

Solucionando problemas na criação de DAG

Interface web do Airflow

Iniciando o webserver do Airflow

Navegando pela interface do Airflow

Analisando DAGs com a UI do Airflow

Introdução ao Airflow

Tudo certo com o DAG? Agora é hora de aprender o básico de como implementar DAGs no Airflow. Com atividades práticas, você vai aprender a configurar e implantar operadores, tarefas e agendamento.

Operadores do Airflow

Definindo uma tarefa BashOperator

Vários BashOperators

Tarefas do Airflow

Definir a ordem de BashOperators

Determinando a ordem das tasks

Solucionando problemas de dependências de DAG

Operadores adicionais

Usando o PythonOperator

Mais PythonOperators

EmailOperator e dependências

Agendamento no Airflow

Agendar um DAG via Python

Decifrando agendamentos no Airflow

Solucionando problemas de execuções de DAG

Implementando DAGs no Airflow

Neste capítulo, você vai aprender a ganhar tempo usando componentes do Airflow, como sensores e executores, enquanto monitora e soluciona problemas em fluxos de trabalho do Airflow.

Sensores do Airflow

Options

Sensors

Both

Operators

Sensores vs operadores

Privação sensorial

Executores do Airflow

Determinando o executor

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='poke',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='reschedule',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


Implicações do executor

Depuração e troubleshooting no Airflow

DAGs na mão

from airflow import DAG
#from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='poke',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

sample_dag = DAG(
    dag_id = 'sample_dag',
    schedule_interval = "0 0 * * *"
)

sample_task = BashOperator(
    task_id='sample',
    bash_command='generate_sample.sh',
    start_date=datetime(2023,2,20),
    dag=sample_dag
)


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


DAG ausente

SLAs e relatórios no Airflow

Definindo um SLA

Definindo um SLA de tarefa

Gerar e enviar um relatório por e-mail

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    '____': [____],
    '____': True,
    'email_on_success': ____
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    'email': ['airflowalerts@datacamp.com', 'airflowadmin@datacamp.com'],
    'email_on_failure': True,
    'email_on_success': True
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


Adicionando e-mails de status

Manutenção e monitoramento de fluxos de trabalho no Airflow

Junte tudo. Neste capítulo final, você vai aplicar tudo o que aprendeu para construir um fluxo de trabalho com qualidade de produção no Airflow.

Trabalhando com templates

Criando um BashOperator com template

Templates com múltiplos argumentos

Mais templates

Usando listas com templates

Entendendo opções de parâmetros

Enviando e-mails com templates

Ramificação

Defina um BranchPythonOperator

Solucionando problemas de ramificação

Criando um pipeline de produção

Criando um pipeline de produção #1

Criando um pipeline de produção #2

Adicionando as últimas alterações ao seu pipeline

Parabéns!

Criando pipelines de produção no Airflow

Agora atualizado para o Apache Airflow 2.7 — Entregar dados em um cronograma pode ser um processo manual. Você escreve scripts, adiciona tarefas cron complexas e tenta várias formas de atender a requisitos que mudam o tempo todo — e gerenciar tudo isso fica ainda mais difícil trabalhando em equipe. O Apache Airflow resolve esse problema ao adicionar agendamento, tratamento de erros e relatórios aos seus fluxos de trabalho. Neste curso, você vai dominar o básico do Apache Airflow e aprender a implementar pipelines complexos de engenharia de dados em produção. Você também vai aprender a usar Grafos Acíclicos Dirigidos (DAGs), automatizar fluxos de trabalho de engenharia de dados e implementar tarefas de engenharia de dados de forma simples e reproduzível — ajudando você a manter a sanidade.

Intermediate Python

Introduction to Shell

Aprenda a implementar e agendar fluxos de engenharia de dados com Airflow e DAGs em Python.

Introdução ao Apache Airflow em Python

Saiba como implementar e programar fluxos de trabalho de engenharia de dados.

Engenheiro de dados Em Python

Só executa uma vez por execução da DAG.

Sensores vs operadores

Introdução ao Apache Airflow em Python

Exercício interativo prático