Data Pipelines Inteligentes com Apache Airflow | Live#95

Introduction

This tutorial provides a comprehensive guide on creating intelligent data pipelines using Apache Airflow. Based on insights from the video by Luan Moreno, you'll learn best practices and new features introduced in Airflow 2.0 that can optimize your ETL (Extract, Transform, Load) processes significantly. The focus will be on utilizing the Astronomer Registry, the TaskFlow API, and other innovative tools to enhance your workflow.

Step 1: Setting Up Apache Airflow

• Install Apache Airflow using pip:

  pip install apache-airflow
  

• Initialize the Airflow database:

  airflow db init
  

• Start the web server:

  airflow webserver --port 8080
  

• In a new terminal, start the scheduler:

  airflow scheduler
  

• Access the Airflow UI at http://localhost:8080.

Step 2: Creating Your First DAG

• Create a new Python file for your DAG (e.g., my_first_dag.py).
• Use the following template to define your DAG:

  from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from datetime import datetime
  
  default_args = {
      'owner': 'airflow',
      'start_date': datetime(2023, 1, 1),
      'retries': 1,
  }
  
  dag = DAG('my_first_dag', default_args=default_args, schedule_interval='@daily')
  
  start = DummyOperator(task_id='start', dag=dag)
  end = DummyOperator(task_id='end', dag=dag)
  
  start >> end
  

• Save the file in the dags folder of your Airflow installation.

Step 3: Utilizing TaskFlow API

• Replace the DummyOperator with the TaskFlow API for better readability and functionality:

  from airflow.decorators import dag, task
  
  @dag(default_args=default_args, schedule_interval='@daily', catchup=False)
  def my_taskflow_dag():
      @task
      def start():
          print("Starting the workflow")
  
      @task
      def end():
          print("Ending the workflow")
  
      start() >> end()
  
  dag_instance = my_taskflow_dag()
  

Step 4: Implementing Task Groups

• Organize your tasks into groups for better structure:

  from airflow.utils.task_group import TaskGroup
  
  @dag(default_args=default_args, schedule_interval='@daily')
  def my_grouped_dag():
      with TaskGroup('group_1') as tg1:
          start = task()(lambda: print("Start Task"))()
          middle = task()(lambda: print("Middle Task"))()
          end = task()(lambda: print("End Task"))()
  
      start >> middle >> end
  
  dag_instance = my_grouped_dag()
  

Step 5: Leveraging Deferrable Operators

• Use deferrable operators to optimize resource usage:

  from airflow.operators.dummy import DummyOperator
  from airflow.operators.python import PythonOperator
  
  @dag(default_args=default_args, schedule_interval='@daily')
  def my_deferrable_dag():
      def run_task():
          print("Task executed")
  
      task1 = PythonOperator(
          task_id='run_task',
          python_callable=run_task,
          dag=dag
      )
  
  dag_instance = my_deferrable_dag()
  

Step 6: Exploring Dynamic Task Mapping

• Create dynamic tasks based on input:

  @dag(default_args=default_args, schedule_interval='@daily')
  def my_dynamic_dag():
      @task
      def generate_tasks():
          return [1, 2, 3]
  
      @task
      def process_task(task_id):
          print(f"Processing task {task_id}")
  
      task_list = generate_tasks()
      process_task.expand(task_id=task_list)
  
  dag_instance = my_dynamic_dag()
  

Conclusion

In this tutorial, you've learned how to set up Apache Airflow, create your first DAG, and utilize advanced features such as the TaskFlow API, Task Groups, Deferrable Operators, and Dynamic Task Mapping. These tools will enable you to build efficient and maintainable data pipelines. To further enhance your skills, consider exploring the Astronomer Python SDK and participating in hands-on workshops.