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

Introdução ao Apache Airflow

Teste de uma tarefa no Airflow

Examinando os comandos de fluxo de ar

Fluxo de ar DAGs

Definição de um simples DAG

Trabalhando com DAGs e o shell Airflow

Solução de problemas DAG creation

Interface web do Airflow

Iniciando o servidor da Web do Airflow

Navegando pelo fluxo de ar UI

Examinando DAGs com o fluxo de ar UI

Introdução ao fluxo de ar

O que você está fazendo em DAG? Agora é hora de você aprender os conceitos básicos da implementação do Airflow DAGs. Por meio de atividades práticas, você aprenderá a configurar e implantar operadores, tarefas e agendamento.

Operadores de fluxo de ar

Definição de uma tarefa BashOperator

Vários BashOperators

Tarefas de fluxo de ar

Definir a ordem dos BashOperators

Determinação da ordem das tarefas

Solução de problemas das dependências do DAG

Operadores adicionais

Usando o PythonOperator

Mais PythonOperators

EmailOperator e dependências

Programação do fluxo de ar

Agende um DAG via Python

Decifrando as programações de fluxo de ar

Solução de problemas DAG runs

Implementação do fluxo de ar DAGs

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

Sensores de fluxo de ar

Opções

Sensores

Ambos

Operadores

Sensores versus operadores

Privação sensorial

Executores de fluxo de ar

Determinação do 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 solução de problemas no Airflow

DAGs na bolsa

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


Faltando DAG

SLAs e relatórios no Airflow

Definição de um SLA

Definição de uma tarefa SLA

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


Adição de e-mails de status

Manutenção e monitoramento dos fluxos de trabalho do Airflow

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

Trabalhando com modelos

Criando um BashOperator com modelo

Modelos com vários argumentos

Mais modelos

Uso de listas com modelos

Compreensão das opções de parâmetros

Envio de e-mails com modelos

Ramificação

Definir um BranchPythonOperator

Solução de problemas de filiais

Criação de um pipeline de produção

Criando um pipeline de produção nº 1

Criação de um pipeline de produção #2

Adicionando as alterações finais ao seu pipeline

Parabéns!

Criando pipelines de produção no Airflow

Agora atualizado para o Apache Airflow 2.7 - A entrega de dados em um cronograma pode ser um processo manual. Você escreve scripts, adiciona tarefas cron complexas e tenta várias maneiras de atender a um conjunto de requisitos em constante mudança - e é ainda mais complicado gerenciar tudo quando se trabalha com colegas de equipe. O Apache Airflow pode eliminar essa dor de cabeça, adicionando agendamento, tratamento de erros e relatórios aos seus fluxos de trabalho. Neste curso, você dominará os conceitos básicos do Apache Airflow e aprenderá a implementar pipelines complexos de engenharia de dados na produção. Você também aprenderá a usar gráficos acíclicos direcionados (DAGs), automatizar fluxos de trabalho de engenharia de dados e implementar tarefas de engenharia de dados de forma fácil e repetível, ajudando-o a manter a sanidade mental.

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

"Aprenda a implementar e agendar fluxos de trabalho de engenharia de dados."

Engenheiro de dados Em Python

DAGs na bolsa

Introdução ao Apache Airflow em Python

Exercício interativo prático