Hướng dẫn sử dụng Airflow: Hướng dẫn toàn diện và thực tế để xây dựng các Pipeline dữ liệu đáng tin cậy
Nếu bạn di chuyển dữ liệu hoặc điều phối các tác vụ ML, bạn có thể đã nghe câu nói quen thuộc: “Cứ đưa nó vào Airflow”. Sự thật là, Apache Airflow tỏa sáng khi bạn cần khả năng hiển thị, độ tin cậy và kiểm soát đối với các quy trình làm việc phức tạp. Trong hướng dẫn thực tế này, chúng ta sẽ từng bước tìm hiểu cách sử dụng Airflow—từ các khái niệm cốt lõi đến các mẫu sẵn sàng cho sản xuất—để bạn có thể triển khai các pipeline mà bạn tin tưởng.
Chúng ta sẽ giữ cho nó mang tính hành động: bạn sẽ có được một mô hình tư duy cho DAG và các tác vụ, các ví dụ thực hành với TaskFlow API, các tùy chọn triển khai, chiến lược kiểm thử và các phương pháp hay nhất. Đến cuối cùng, bạn sẽ chuyển từ “Tôi có thể chạy hướng dẫn” thành “Tôi có thể chạy cái này trong prod”.
Lưu ý: Để tìm hiểu sâu hơn và tham khảo, tài liệu chính thức rất tuyệt vời và được cập nhật thường xuyên.
Apache Airflow Thực Sự Là Gì?
Airflow là một trình điều phối—không phải là một bộ xử lý dữ liệu. Nó lên lịch, sắp xếp và giám sát công việc bạn chạy ở nơi khác (cơ sở dữ liệu, kho dữ liệu, công việc Spark, API, container). Bạn định nghĩa quy trình làm việc dưới dạng DAG (Đồ thị có hướng không chu trình), đây chỉ là các tệp Python mã hóa các tác vụ và sự phụ thuộc của chúng. Sau đó, Airflow thực thi các tác vụ đó theo lịch trình, tham số và môi trường của bạn.
- DAG: Định nghĩa quy trình làm việc (đồ thị các tác vụ với các phụ thuộc).
- Task (Tác vụ): Một đơn vị công việc (hàm Python, thực thi SQL, lệnh Bash, trình kích hoạt công việc bên ngoài, v.v.).
- Operator (Toán tử): Một template cho một loại tác vụ (ví dụ:
PythonOperator, BashOperator, KubernetesPodOperator).
- Scheduler (Bộ lập lịch): Quyết định những gì cần chạy và khi nào.
- Executor (Trình thực thi): Chạy các tác vụ (cục bộ, với Celery, Kubernetes, v.v.).
- UI: Trung tâm điều khiển của bạn cho các lần chạy, nhật ký, thử lại và dòng dõi.
Bắt đầu với các hướng dẫn chính thức sau khi bạn đã cài đặt Airflow; chúng cung cấp cho bạn bức tranh toàn cảnh một cách nhanh chóng.
Cài Đặt và Chạy Airflow Đúng Cách
Airflow rất linh hoạt. Chọn đường dẫn phù hợp với giai đoạn của bạn:
- Phát triển cục bộ (khởi động nhanh):
- Sử dụng Docker Compose khởi động nhanh do dự án cung cấp. Nó khởi động webserver, scheduler, database và hơn thế nữa với các giá trị mặc định hợp lý.
- Tuyệt vời để học và lặp lại trên DAG.
- Celery Executor hoặc Kubernetes Executor với Postgres được quản lý.
- Lưu trữ nhật ký trong S3/GCS và đóng gói các phụ thuộc bằng image hoặc
requirements.txt.
- Kubernetes Executor để co giãn hoặc Celery Executor với autoscaling workers.
- External secrets (Vault), khả năng quan sát mạnh mẽ (nhật ký + metrics) và blue/green deploys để nâng cấp.
Mẹo: Giữ codebase Airflow của bạn được kiểm soát phiên bản, container hóa và kiểm tra trước khi quảng bá. Trang “Best Practices” phác thảo các mẫu sẵn sàng cho sản xuất.
Các Khái Niệm Cốt Lõi Bạn Sẽ Sử Dụng Hàng Ngày
DAG: Quy Trình Làm Việc Của Bạn Dưới Dạng Mã
DAG là một tệp Python định nghĩa:
- Metadata DAG: id, schedule, start date, tags.
- Default args: retries, owners, SLAs.
- Các tác vụ và sự phụ thuộc của chúng.
Hãy nghĩ về DAG như “cái gì” và “khi nào”, và các tác vụ là “như thế nào”.
Các Tác Vụ và Toán Tử
Operators là các bản dựng sẵn cho các tác vụ phổ biến. Ví dụ:
- PythonOperator / TaskFlow
@task cho mã Python
- BashOperator cho các lệnh shell
- SimpleHttpOperator cho API
- KubernetesPodOperator cho các công việc container hóa
- Các nhà cung cấp SQL (ví dụ: Snowflake, BigQuery, Postgres) cho công việc kho dữ liệu
TaskFlow API: Cách Tiếp Cận Hiện Đại, Dựa Trên Python
TaskFlow API cho phép bạn viết các tác vụ dưới dạng các hàm Python với @task, trả về các giá trị truyền qua XCom và kết hợp chúng một cách rõ ràng. Nó giảm boilerplate và cải thiện khả năng đọc—rất được khuyến khích.
DAG Airflow Đầu Tiên Của Bạn (Phiên Bản TaskFlow)
Dưới đây là một ví dụ tối thiểu theo kiểu ETL để minh họa các ý tưởng chính: lập lịch, TaskFlow, các phụ thuộc và truyền dữ liệu XCom.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(
dag_id="weather_etl_example",
start_date=datetime(2024, 1, 1),
schedule="0 * * * *", # hourly
catchup=False,
tags=.
Lập Lịch, Bắt Kịp và Backfill
schedule: Cron hoặc các cài đặt trước (@daily, @hourly).
start_date + catchup: Nếu catchup=True, Airflow sẽ backfill các lần chạy từ ngày bắt đầu. Đối với các pipeline kiểu streaming, hãy đặt catchup=False.
- Backfill thủ công: Sử dụng UI hoặc CLI để chạy lại các khoảng thời gian lịch sử.
Quy tắc thực tế: bật catchup cho các công việc hàng loạt xác định; tắt cho các pipeline giới hạn tốc độ API hoặc thời gian thực.
Truyền Dữ Liệu Giữa Các Tác Vụ (XCom) Một Cách An Toàn
- Các đối tượng nhỏ: trả về các giá trị bằng TaskFlow là tốt.
- Payload lớn: lưu trữ trong bộ nhớ đối tượng (S3/GCS) với một key trong XCom.
- Tránh dữ liệu nhạy cảm trong XCom; sử dụng các secrets backends (ví dụ: Vault) và các biến môi trường.
Ánh Xạ Tác Vụ Động và Khối Lượng Công Việc Fan-out
Airflow có thể tạo các tác vụ động tại thời điểm chạy dựa trên đầu vào—lý tưởng cho các tập dữ liệu được phân vùng hoặc các công việc đa người thuê.
- Giữ cho DAG xác định và idempotent.
- Tách biệt điều phối (Airflow) khỏi tính toán (Spark, dbt, kho dữ liệu).
- Sử dụng TaskFlow API để rõ ràng và vệ sinh XCom.
- Tham số hóa DAG; sử dụng các biến một cách thận trọng.
- Giám sát, cảnh báo và ghi lại các pipeline của bạn.
Cách Làm Việc Với Kho Dữ Liệu và ML
- Kho dữ liệu: Sử dụng các toán tử nhà cung cấp (ví dụ: SnowflakeOperator, BigQueryInsertJobOperator) cho các công việc SQL. Lưu trữ SQL trong các tệp hoặc các module được kiểm soát phiên bản.
- dbt: Kích hoạt dbt thông qua Bash/KubernetesPodOperator hoặc các toán tử dbt chuyên dụng trong các nhà cung cấp.
- ML: Điều phối tạo feature, đào tạo và suy luận hàng loạt như các tác vụ riêng biệt; lưu trữ các artifacts trong bộ nhớ và ghi lại các metrics.
Lập Lịch Nâng Cao: Tập Dữ Liệu và Các Phụ Thuộc Chéo DAG
- Datasets cho phép một DAG tạo ra một tập dữ liệu logic kích hoạt một DAG khác khi được cập nhật—sạch hơn so với các trình kích hoạt ad-hoc.
- Đối với các mẫu cũ, ExternalTaskSensor hoạt động, nhưng datasets mang tính khai báo hơn.
Bảo Mật và Tuân Thủ
- Sử dụng kiểm soát truy cập dựa trên vai trò (RBAC) trong UI.
- Cô lập môi trường cho mỗi nhóm hoặc ranh giới tin cậy.
- Giữ audit trails thông qua nhật ký và lịch sử thay đổi kết nối.
Nâng Cấp và Kiểm Soát Phiên Bản
- Kiểm tra nâng cấp trong dàn dựng với khối lượng công việc giống như sản xuất.
- Ghim và nâng cấp các nhà cung cấp một cách có chủ ý.
- Đọc release notes để biết các thay đổi và các deprecated liên quan đến executor.
Danh Sách Kiểm Tra Nhanh Cho DAG Sản Xuất Đầu Tiên Của Bạn
- Quyền sở hữu rõ ràng (tag
owner) và các cảnh báo được định cấu hình.
- Đã đặt
retries với backoffs hợp lý.
- Các tác vụ Idempotent và các phụ thuộc rõ ràng.
- Payload XCom nhỏ; dữ liệu lớn trong bộ nhớ.
- Nhật ký được chuyển đến bộ nhớ lâu bền; các metrics được xuất.
- Kế hoạch triển khai (canary hoặc blue/green) và các bước rollback.
Ví dụ: DAG Tải Kho Dữ Liệu Thực Tế
Mẫu này trích xuất các tệp hàng ngày, xác thực chúng và tải chúng vào một bảng kho dữ liệu, với ánh xạ động cho mỗi phân vùng và các deferrable sensors.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- Xem lại Best Practices trước khi quảng bá lên sản xuất.
- Khám phá tài liệu nhà cung cấp cho các hệ thống của bạn (kho dữ liệu, đám mây, công cụ ML).
Nhân tiện: Tăng tốc soạn thảo với một trợ lý AI
Điều đáng chú ý: nếu bạn phác thảo nhiều DAG, một trợ lý AI hiểu mã có thể tăng tốc boilerplate, tạo các stubs TaskFlow và thậm chí đề xuất các bản sửa lỗi phụ thuộc. Nếu bạn muốn một trình trợ giúp nhẹ bên cạnh trình soạn thảo và trình duyệt của mình, Sider.AI có thể hữu ích cho việc viết lại và giải thích mã nhanh chóng trong quá trình phát triển. Những Điểm Chính
- Sử dụng Airflow để điều phối, không phải tính toán.
- Ưu tiên TaskFlow API cho các DAG sạch sẽ, có thể kiểm tra.
- Giữ dữ liệu bên ngoài XCom; thay vào đó hãy truyền các tham chiếu.
- Sử dụng các deferrable sensors/operators để tiết kiệm slots.
- Container hóa, kiểm tra và quảng bá thông qua các môi trường.
- Dựa vào các hướng dẫn chính thức và các phương pháp hay nhất như là kim chỉ nam của bạn.
FAQ
Q1: Cách dễ nhất để học cách sử dụng Airflow là gì?
Bắt đầu với Hướng dẫn chính thức để hiểu DAG, các tác vụ, lập lịch và UI. Sau đó, xây dựng một pipeline nhỏ dựa trên TaskFlow và lặp lại với hướng dẫn các phương pháp hay nhất để sẵn sàng cho sản xuất.
Q2: Tôi có nên sử dụng TaskFlow API hay các toán tử cổ điển trong Airflow?
Sử dụng TaskFlow API cho hầu hết các pipeline Pythonic vì nó sạch hơn và xử lý các XCom returns một cách tự nhiên. Các toán tử cổ điển vẫn rất tốt cho các tác vụ không phải Python như Bash, SQL hoặc các công việc container.
Q3: Làm cách nào để truyền dữ liệu lớn giữa các tác vụ Airflow?
Tránh đưa các payload lớn vào XCom. Lưu trữ dữ liệu trong S3/GCS hoặc cơ sở dữ liệu và chỉ truyền các tham chiếu hoặc URI thông qua XCom để giữ cho các tác vụ nhanh chóng và đáng tin cậy.
Q4: Tôi nên chọn executor nào cho Airflow trong sản xuất?
Để co giãn và cô lập, Kubernetes Executor là một mặc định mạnh mẽ. Đối với các thiết lập đơn giản hơn, Celery Executor hoạt động tốt—chỉ cần đảm bảo autoscaling, ghi nhật ký mạnh mẽ và externalized secrets.
Q5: Làm cách nào để xử lý các phụ thuộc trên nhiều DAG Airflow?
Sử dụng Datasets cho các trình kích hoạt chéo DAG khai báo khi một pipeline tạo ra dữ liệu cho một pipeline khác. Ngoài ra, ExternalTaskSensor có thể điều phối các lần chạy, nhưng Datasets sạch hơn cho việc điều phối dựa trên dữ liệu.
