Trong chương này, bạn sẽ được giới thiệu đầy đủ về các thành phần của Apache Airflow và học cách cũng như lý do bạn nên sử dụng chúng.

Giới thiệu về Apache Airflow

Bộ đếm Dag

Chạy Dag đầu tiên của bạn

Airflow DAGs

Tạo một Dag đơn giản

Làm việc với Dag và shell của Airflow

Khám phá các lệnh Airflow

Kiểm thử một task trong Airflow

Giao diện web của Airflow

Điều hướng trong Airflow UI

Khám phá các Dag với Airflow UI

Lỗi Dag

Giới thiệu về Airflow

Xin chào Dag! Đến lúc học những điều cơ bản để triển khai Dag trong Airflow. Qua các hoạt động thực hành, bạn sẽ học cách thiết lập và triển khai operator, task và lập lịch.

Các operator trong Airflow

Thêm decorator cho một task

Định nghĩa BashOperator với @task.bash

Xác định thứ tự Task

Nhiều @tasks hơn

Lập lịch cho Dags

Lên lịch một Dag bằng Python

Giải mã lịch chạy của Airflow

Khắc phục sự cố chạy Dag

Truyền dữ liệu giữa các tác vụ với XCom

Truyền dữ liệu với XCom

Biến XCom

Airflow Sensor

Xác định thứ tự các tác vụ

Tạo một FileSensor

Mất giác quan

Xây dựng Dag trong Airflow

Trong chương này, bạn sẽ học cách tiết kiệm thời gian với các thành phần của Airflow như sensor và executor, đồng thời giám sát và xử lý sự cố các quy trình Airflow.

Giám sát, Cảnh báo và Callbacks

Triển khai hàm callback

Triển khai SmtpNotifier

Có gì trong log?

Dùng mẫu với Jinja

Viết với Jinja

Thay thế tham số

Airflow Variables

Câu trả lời là Variable

Đọc một Variable trong Python

Biến chưa chuẩn lắm

Gỡ lỗi và xử lý sự cố trong Airflow

Khắc phục sự cố Dag

Thiếu Dag

Duy trì và giám sát quy trình Airflow

Gom mọi thứ lại với nhau. Trong chương cuối, bạn sẽ áp dụng mọi thứ đã học để xây dựng một quy trình làm việc chất lượng production trong Airflow.

Triggers và khả năng chịu lỗi

Triển khai Trigger Rule

Thêm retry

Kích hoạt một Dag con

Rẽ nhánh

Định nghĩa một @task.branch

Khắc phục sự cố nhánh (branch)

Toán tử HITL

Thêm bước phê duyệt của con người

Tác vụ ApprovalOperator

Tạo một pipeline chạy trên môi trường production

# Imports
from airflow.sdk import ____, ____, ____
from airflow.providers.standard.sensors.filesystem import ____
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='____',
  on_success_callback=____(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.____}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
    
  initialized = initialize()
  initialized >> filewatcher
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
      
  initialized = initialize()
  initialized >> filewatcher 
  
sales_etl()

Thiết lập pipeline ETL cho dữ liệu bán hàng

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @____
  def validate_data():
    sales_path = Variable.____('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = '____'
    if '-09-' in sales_path:
      next_task = '____'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

Thêm nhánh theo tháng

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ____

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ____(
    task_id="____",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), ____] >> push_to_warehouse()
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ApprovalOperator

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ApprovalOperator(
    task_id="yearend_approval_task",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), yearend_approval_task] >> push_to_warehouse()
  
  
sales_etl()

Thêm bước phê duyệt bởi con người

Chúc mừng bạn!

Kiểm soát logic của Dag

Đã cập nhật lên Apache Airflow 3.1.6 - Chuyển giao dữ liệu theo lịch có thể là một quy trình thủ công. Bạn viết script, thêm các tác vụ cron phức tạp, và thử nhiều cách để đáp ứng một bộ yêu cầu luôn thay đổi — và việc quản lý mọi thứ càng khó hơn khi làm việc theo nhóm. Apache Airflow có thể giải quyết vấn đề này bằng cách bổ sung tính năng lập lịch, xử lý lỗi và báo cáo cho quy trình làm việc của bạn. Trong khóa học này, bạn sẽ nắm vững các kiến thức cơ bản về Apache Airflow và học cách triển khai các pipeline kỹ thuật dữ liệu phức tạp trong môi trường production. Bạn cũng sẽ học cách sử dụng Directed Acyclic Graphs (Dag), tự động hóa quy trình kỹ thuật dữ liệu, và triển khai các tác vụ kỹ thuật dữ liệu theo cách dễ dàng và có thể lặp lại — giúp bạn giữ mọi thứ trong tầm kiểm soát.

Intermediate Python

Introduction to Shell

Học triển khai và lập lịch quy trình dữ liệu với Airflow và DAGs trong Python.

Giới thiệu về Apache Airflow bằng Python

Tìm hiểu cách triển khai và lên lịch các quy trình công việc kỹ thuật dữ liệu.

Triggers và khả năng chịu lỗi

Create Your Free Account