Proxy Apache Airflow: 5 Bước Xây Data Pipeline Ổn Định

SEO Administrator 08-11-2026 Đang tính... phút đọc

Proxy Apache Airflow giúp Data Engineer điều phối hoạt động thu thập dữ liệu, phân bổ kết nối và xử lý lỗi có hệ thống. Trong bài viết này, ProxyVN sẽ cùng anh em xây dựng pipeline e-commerce qua 5 bước rõ ràng. Qua đó, anh em có thể giảm lỗi 429, bảo vệ thông tin proxy và duy trì chất lượng dữ liệu khi khối lượng request tăng cao.

Proxy Apache Airflow giải quyết vấn đề gì?

Một pipeline thu thập dữ liệu có thể hoạt động ổn định trong giai đoạn thử nghiệm. Tuy nhiên, khi số URL và tần suất cập nhật tăng lên, hệ thống thường bắt đầu xuất hiện lỗi.

Lỗi quá tải dữ liệu

Hai mã trạng thái phổ biến nhất là 403 Forbidden và 429 Too Many Requests. Lỗi 403 cho biết máy chủ hiểu request nhưng từ chối xử lý. Trong khi đó, lỗi 429 thường xuất hiện khi client gửi quá nhiều request trong một khoảng thời gian.

Nếu chỉ chạy một script Python bằng cron, anh em sẽ khó kiểm soát đồng thời lịch chạy, số lần thử lại, trạng thái từng tác vụ và chất lượng dữ liệu đầu ra. Đây là lúc Airflow phát huy vai trò của một nền tảng điều phối workflow.

Khi kết hợp với proxy, Airflow có thể hỗ trợ:

Lập lịch thu thập dữ liệu theo giờ, ngày hoặc sự kiện.
Giới hạn số tác vụ chạy đồng thời.
Phân phối kết nối qua các endpoint proxy được cấp phép.
Thử lại tác vụ khi gặp lỗi mạng tạm thời.
Ghi log rõ ràng cho từng lần chạy.
Cảnh báo khi tỷ lệ lỗi vượt ngưỡng.
Tách dữ liệu thô, dữ liệu đã làm sạch và dữ liệu phục vụ phân tích.

Proxy không phải công cụ để bỏ qua điều khoản truy cập của website. Anh em chỉ nên thu thập dữ liệu công khai hoặc dữ liệu mình được phép sử dụng, đồng thời tuân thủ API policy, robots.txt và giới hạn của nguồn.

Như vậy, giá trị chính của mô hình này không nằm ở việc gửi request nhiều hơn. Giá trị nằm ở khả năng kiểm soát request tốt hơn, phục hồi có điều kiện và đo lường được từng điểm lỗi.

Nút thắt của pipeline e-commerce quy mô lớn

Khi pipeline mở rộng, lỗi thường không đến từ một nguyên nhân riêng lẻ. Hệ thống có thể gặp vấn đề ở máy chủ đích, proxy pool, Airflow Worker hoặc chính cách thiết kế luồng dữ liệu.

Rate limit và lỗi kết nối

Máy chủ e-commerce thường áp dụng giới hạn theo IP, tài khoản, token hoặc endpoint. Nếu vượt ngưỡng, hệ thống có thể trả về mã 429 kèm header Retry-After.

Trong trường hợp này, retry ngay lập tức chỉ khiến tỷ lệ lỗi tăng thêm. Pipeline nên đọc thời gian chờ do server cung cấp hoặc áp dụng exponential backoff để giãn cách các lần thử.

Lỗi 403 cần được xử lý thận trọng hơn. Đây có thể là dấu hiệu request thiếu quyền, token hết hạn, endpoint thay đổi hoặc hoạt động thu thập không phù hợp với chính sách của nguồn. Vì vậy, không nên mặc định đổi IP rồi tiếp tục gửi request khi gặp 403.

Cạn tài nguyên proxy

Một pool nhỏ khó phục vụ hàng nghìn kết nối đồng thời. Khi số tác vụ vượt quá năng lực của pool, pipeline có thể gặp timeout, kết nối chậm hoặc sử dụng lặp lại cùng một IP quá dày.

Cạn tài nguyên proxy

Ngoài số lượng IP, anh em còn cần theo dõi:

Tỷ lệ request thành công theo từng endpoint.
Thời gian phản hồi trung bình.
Số lần timeout.
Lượng băng thông đã sử dụng.
Số IP đang hoạt động.
Tỷ lệ lỗi 403, 429 và 5xx.

Các chỉ số này giúp team biết vấn đề nằm ở proxy, website nguồn hay cấu hình workflow.

Thiếu trạng thái dùng chung

Nếu Worker A phát hiện một endpoint proxy đang lỗi nhưng Worker B vẫn tiếp tục sử dụng endpoint đó, tài nguyên sẽ bị lãng phí. Do đó, trạng thái proxy cần được lưu tại nơi các Worker có thể cùng truy cập, chẳng hạn Redis hoặc một dịch vụ quản lý proxy riêng.

Airflow nên đảm nhiệm việc điều phối. Việc kiểm tra sức khỏe, chấm điểm và phân phối IP nên được tách thành một lớp dịch vụ rõ ràng khi hệ thống phát triển lớn hơn.

Lợi ích của Proxy Apache Airflow trong Data Pipeline

Airflow biến quy trình thu thập dữ liệu thành một DAG gồm các tác vụ có quan hệ phụ thuộc. Nhờ đó, anh em có thể quan sát toàn bộ luồng từ lúc gửi request đến khi dữ liệu được đưa vào kho phân tích.

Quản lý retry có điều kiện

Airflow hỗ trợ cấu hình số lần thử lại, thời gian chờ và exponential backoff. Cơ chế này phù hợp với các lỗi tạm thời như timeout, mất kết nối hoặc phản hồi 5xx.

Tuy nhiên, retry cần đi kèm phân loại lỗi. Ví dụ:

Retry khi gặp timeout hoặc lỗi 502, 503, 504.
Tôn trọng Retry-After khi gặp 429.
Dừng và kiểm tra quyền truy cập khi gặp 401 hoặc 403.
Không retry dữ liệu sai cấu trúc mà chưa xác định nguyên nhân.

Cách phân loại này giúp giảm request thừa và tránh làm Airflow Worker bận với những tác vụ không thể tự phục hồi.

Điều phối tác vụ song song

Airflow Pools cho phép giới hạn số tác vụ cùng sử dụng một tài nguyên. Theo tài liệu chính thức về Airflow Pools, khi pool hết slot, các tác vụ tiếp theo sẽ chuyển sang hàng đợi.

Đây là lớp bảo vệ quan trọng cho cả proxy pool lẫn hệ thống nguồn. Thay vì kích hoạt 100 kết nối cùng lúc, anh em có thể giới hạn còn 5 hoặc 10 kết nối, sau đó điều chỉnh dựa trên số liệu đo được.

Theo dõi và chạy lại dễ dàng

Mỗi task trong Airflow đều có trạng thái, thời gian bắt đầu, thời gian kết thúc và log riêng. Khi một bước thất bại, team có thể chạy lại đúng task đó thay vì khởi động toàn bộ pipeline.

Nhờ vậy, quá trình điều tra lỗi trở nên rõ ràng hơn. Anh em cũng dễ xác định dữ liệu nào đã được xử lý, dữ liệu nào cần thu thập lại và tác vụ nào đang làm chậm toàn bộ DAG.

5 bước thiết lập Proxy Apache Airflow cho e-commerce

Quy trình dưới đây tập trung vào khả năng kiểm soát, bảo mật và chất lượng dữ liệu. Anh em có thể điều chỉnh số Worker, lịch chạy và loại proxy dựa trên quy mô dự án.

Bước 1: Thiết kế kiến trúc dữ liệu theo từng lớp

Trước khi viết DAG, anh em cần xác định dữ liệu đi qua những giai đoạn nào. Một cấu trúc dễ quản lý là mô hình Bronze, Silver và Gold.

Extract: Gửi request qua kết nối được cấp phép và nhận HTML hoặc JSON.
Bronze: Lưu dữ liệu gốc vào S3, MinIO hoặc object storage tương đương.
Silver: Làm sạch, loại bản ghi trùng và chuẩn hóa trường dữ liệu.
Gold: Tổng hợp dữ liệu phục vụ dashboard, báo cáo hoặc mô hình phân tích.

Điểm quan trọng là không đưa toàn bộ HTML vào XCom. Tài liệu XCom của Apache Airflow nêu rõ XCom được thiết kế cho lượng dữ liệu nhỏ, không phù hợp để truyền các giá trị lớn như dataframe.

Vì vậy, task thu thập nên lưu dữ liệu thô lên object storage. Sau đó, task chỉ trả về URI, mã checksum, thời gian thu thập và một số metadata cần thiết.

Kinh nghiệm triển khai: Khi chỉ truyền URI qua XCom, metadata database nhẹ hơn và team cũng dễ chạy lại bước xử lý mà không cần gọi lại website nguồn.

Bước 2: Lưu thông tin proxy trong Airflow Connections

Không nên viết username, password hoặc proxy URL trực tiếp trong file DAG. Nếu mã nguồn được đưa lên Git, thông tin xác thực có thể bị lộ qua lịch sử commit.

Airflow cung cấp Connections để lưu hostname, port, login, password và các tham số bổ sung. Tài liệu Connections & Hooks mô tả đây là cơ chế dành riêng cho thông tin kết nối tới hệ thống bên ngoài.

Anh em có thể cấu hình như sau:

Mở Airflow UI.
Truy cập mục quản lý Connections.
Tạo Connection loại HTTP.
Điền host, port, username và password.
Đặt Connection ID là my_residential_proxy.
Kiểm tra quyền xem và sửa Connection của từng tài khoản Airflow.

Với môi trường doanh nghiệp, anh em có thể kết hợp Secrets Backend như HashiCorp Vault hoặc dịch vụ quản lý bí mật của nền tảng đám mây. Cách này giúp xoay vòng mật khẩu mà không phải sửa DAG.

Bước 3: Chọn chiến lược xoay IP phù hợp

Không phải tác vụ nào cũng cần thay IP sau mỗi request. Chiến lược xoay phụ thuộc vào trạng thái phiên, giới hạn của nguồn và quyền truy cập được cấp.

Per-request rotation phù hợp với các request độc lập. Ví dụ, pipeline đọc danh mục sản phẩm công khai, trong đó mỗi URL không phụ thuộc cookie của URL trước.

Sticky session phù hợp với quy trình nhiều bước. Một IP được giữ trong một khoảng thời gian để duy trì tính nhất quán của phiên, cookie và vị trí địa lý.

Anh em có thể chọn theo nguyên tắc:

Dùng IP xoay cho dữ liệu công khai, không có trạng thái.
Dùng sticky session cho quy trình nhiều bước được cho phép.
Không đổi quốc gia giữa các bước trong cùng một phiên.
Không dùng chung một session cho nhiều tài khoản.
Đặt giới hạn request cho từng domain.
Loại proxy khỏi pool khi timeout liên tiếp vượt ngưỡng.

Nếu nguồn cho phép 5 request/phút/IP và pipeline cần xử lý trung bình 100.000 request/ngày, tải trung bình vào khoảng 69,4 request/phút. Về mặt số học, hệ thống cần ít nhất 14 IP hoạt động đồng thời.

Tuy nhiên, con số triển khai nên có thêm dung lượng dự phòng cho timeout, thời điểm tải tăng và IP tạm ngừng. Khoảng 17–20 IP có thể là điểm bắt đầu để kiểm thử, không phải con số dùng chung cho mọi website.

Bước 4: Viết TaskFlow DAG với retry và proxy

Với Airflow 3, tác giả DAG nên sử dụng namespace airflow.sdk. Đây là giao diện công khai được Apache khuyến nghị trong tài liệu Public Interface của Airflow 3.

Cài các thư viện cần thiết trong môi trường Airflow:

pip install apache-airflow curl-cffi

DAG minh họa dưới đây lấy Connection từ Airflow, gửi request qua proxy và trả về một URI giả lập thay vì đẩy HTML vào XCom:

from datetime import timedelta

import pendulum

from airflow.sdk import BaseHook, dag, task

from curl_cffi import requests

@dag(

dag_id="ecommerce_proxy_pipeline",

schedule="@hourly",

start_date=pendulum.datetime(2026, 6, 1, tz="UTC"),

catchup=False,

default_args={

"owner": "data_engineer",

"retries": 4,

"retry_delay": timedelta(minutes=1),

"retry_exponential_backoff": True,

"max_retry_delay": timedelta(minutes=15),

tags=["ecommerce", "proxy"],

)

def ecommerce_proxy_pipeline():

@task(pool="ecommerce_api_pool")

def fetch_data(target_url: str, proxy_conn_id: str) -> str:

conn = BaseHook.get_connection(proxy_conn_id)

proxy_url = (

f"http://{conn.login}:{conn.password}"

f"@{conn.host}:{conn.port}"

)

proxies = {

"http": proxy_url,

"https": proxy_url,

}

response = requests.get(

target_url,

proxies=proxies,

impersonate="chrome",

timeout=30,

)

if response.status_code == 429:

raise RuntimeError("Rate limit: kiểm tra Retry-After")

if response.status_code in (401, 403):

raise PermissionError(

"Nguồn từ chối truy cập: kiểm tra quyền và chính sách"

)

response.raise_for_status()

object_uri = "s3://data-lake/bronze/products/run_id.json"

# Lưu response.content lên object storage tại đây.

return object_uri

@task

def validate_bronze(object_uri: str) -> dict:

return {

"object_uri": object_uri,

"status": "validated",

}

bronze_uri = fetch_data(

target_url="https://api.example.com/v1/products",

proxy_conn_id="my_residential_proxy",

)

validate_bronze(bronze_uri)

ecommerce_proxy_pipeline()

Tham số impersonate="chrome" của curl_cffi giúp client sử dụng một cấu hình dấu vân tay được thư viện hỗ trợ. Tuy nhiên, hướng dẫn của curl_cffi cũng cho thấy thư viện không tạo được JavaScript fingerprint như trình duyệt thật.

Do đó, tính năng này không bảo đảm request được chấp nhận. Quyền truy cập, chất lượng IP, tần suất gửi, cookie và chính sách của nguồn vẫn là những yếu tố quan trọng.

Trong môi trường thật, nhánh xử lý 429 nên đọc Retry-After và trì hoãn đúng thời gian. Với 401 hoặc 403, task nên dừng, ghi nhận nguyên nhân và chuyển sang hàng đợi kiểm tra thay vì đổi IP liên tục.

Bước 5: Thiết lập Data Quality và Monitoring

Một request trả về mã 200 chưa có nghĩa dữ liệu đã đúng. Website có thể trả về trang thông báo, CAPTCHA hoặc một cấu trúc JSON khác với dự kiến.

Vì vậy, pipeline cần kiểm tra cả trạng thái HTTP lẫn nội dung:

Xác nhận Content-Type đúng định dạng.
Kiểm tra các trường bắt buộc.
Xác thực giá là số và lớn hơn 0.
Phát hiện HTML xuất hiện trong response JSON.
Loại bản ghi trùng bằng khóa sản phẩm.
Gắn thời gian thu thập và nguồn dữ liệu.
Đưa bản ghi bất thường vào vùng quarantine.

Đối với monitoring, anh em nên theo dõi tỷ lệ thành công theo domain và proxy endpoint. Nếu task duration tăng nhưng lưu lượng không đổi, đây có thể là dấu hiệu proxy chậm, nguồn đang throttling hoặc Worker thiếu tài nguyên.

Các cảnh báo nên được kích hoạt khi:

Tỷ lệ lỗi vượt ngưỡng trong 5–15 phút.
Số proxy hoạt động thấp hơn mức dự phòng.
Tỷ lệ 429 tăng nhanh.
Schema dữ liệu thay đổi.
Dung lượng Bronze Layer tăng bất thường.
DAG trễ hơn lịch vận hành dự kiến.

Airflow cũng hỗ trợ callback và cơ chế notifier. Team có thể gửi thông tin gồm DAG ID, Task ID, thời gian lỗi và đường dẫn log sang Slack, email hoặc hệ thống cảnh báo nội bộ.

Case study: Từ cron job đến pipeline có kiểm soát

Một tình huống thường gặp là team chạy script mỗi giờ để lấy giá sản phẩm. Khi danh sách tăng từ 2.000 lên 30.000 URL, thời gian chạy kéo dài và nhiều request bị lặp.

Nếu chuyển sang Airflow nhưng chỉ tăng số Worker, tỷ lệ lỗi có thể vẫn cao. Nguyên nhân là hệ thống chưa giới hạn concurrency theo domain và chưa phân biệt lỗi tạm thời với lỗi quyền truy cập.

Cách cải thiện phù hợp gồm:

Chia URL thành các batch nhỏ.
Tạo pool riêng cho từng nguồn.
Áp dụng retry cho timeout và lỗi 5xx.
Tôn trọng Retry-After khi gặp 429.
Lưu dữ liệu thô lên object storage.
Chỉ truyền URI giữa các task.
Cô lập response sai cấu trúc.
Theo dõi chi phí trên mỗi bản ghi thành công.

Sau thay đổi, chỉ số cần ưu tiên không phải tổng số request. Chỉ số có ý nghĩa hơn là số bản ghi hợp lệ trên mỗi phút và chi phí cho mỗi bản ghi hợp lệ.

Góc nhìn này giúp team tránh tình trạng mua thêm proxy hoặc tăng Worker trong khi nguyên nhân nằm ở schema, lịch chạy hoặc request bị trùng.

Nguyên tắc vận hành an toàn và bền vững

Proxy nên được xem là một thành phần mạng trong kiến trúc, không phải giải pháp duy nhất cho rate limit. Trước khi mở rộng proxy pool, anh em nên kiểm tra khả năng dùng API chính thức, feed dữ liệu hoặc thỏa thuận truy cập với chủ sở hữu nguồn.

Vận hành proxy pool

Google Search Central giải thích robots.txt chủ yếu dùng để quản lý lưu lượng crawler và quyền truy cập URL. Tuy nhiên, robots.txt không thay thế điều khoản dịch vụ hay cơ chế cấp quyền.

Một pipeline phù hợp nên:

Chỉ truy cập dữ liệu công khai hoặc được cấp phép.
Nhận diện rõ User-Agent khi chính sách yêu cầu.
Giới hạn tốc độ theo domain.
Cache dữ liệu để tránh request lặp.
Không thu thập thông tin cá nhân không cần thiết.
Mã hóa credential và hạn chế quyền truy cập.
Xây cơ chế dừng khẩn cấp khi tỷ lệ lỗi tăng.
Ghi lại nguồn, thời gian và mục đích sử dụng dữ liệu.

Đây cũng là cách giảm rủi ro kỹ thuật. Khi nguồn thay đổi chính sách hoặc API, team có thể dừng đúng DAG và đánh giá lại thay vì để hàng nghìn request tiếp tục chạy.

Câu hỏi thường gặp

Proxy dân cư và proxy datacenter khác nhau thế nào?

Proxy datacenter sử dụng IP thuộc hạ tầng trung tâm dữ liệu. Loại này thường có tốc độ tốt và chi phí dễ kiểm soát. Proxy dân cư sử dụng IP do ISP cấp cho kết nối dân cư nên đặc điểm mạng khác với IP máy chủ.

Việc lựa chọn phụ thuộc vào nguồn, vị trí, ngân sách và quyền truy cập. Anh em nên thử trên một tập dữ liệu nhỏ, sau đó so sánh tỷ lệ thành công, độ trễ và chi phí trên dữ liệu hợp lệ.

Có nên retry mọi lỗi 403 và 429 không?

Không. Với 429, pipeline nên tôn trọng Retry-After hoặc tăng thời gian chờ. Với 403, team cần kiểm tra quyền, token, endpoint và chính sách truy cập.

Retry liên tục một lỗi không thể tự phục hồi chỉ làm tăng tải và chi phí. Vì vậy, pipeline cần phân loại lỗi trước khi quyết định thử lại.

curl_cffi có giúp mọi request thành công không?

Không. Thư viện hỗ trợ một số cấu hình TLS và HTTP/2 tương ứng với các profile trình duyệt được công bố. Tuy nhiên, nó không mô phỏng toàn bộ môi trường trình duyệt và không thay thế quyền truy cập hợp lệ.

Hiệu quả cuối cùng còn phụ thuộc proxy, session, cookie, tần suất request và cơ chế bảo vệ của nguồn.

Có nên trả Raw HTML trực tiếp qua XCom không?

Không nên. XCom phù hợp với giá trị nhỏ như URI, ID, trạng thái hoặc metadata. Raw HTML và dataframe nên được lưu vào S3, MinIO hoặc một object storage tương đương.

Cách làm này giảm tải metadata database và giúp pipeline xử lý lại dữ liệu mà không phải gửi request mới.

Chọn bao nhiêu slot cho Airflow Pool?

Hãy bắt đầu từ giới hạn thấp, chẳng hạn 5–10 slot cho một domain. Sau đó, anh em theo dõi tỷ lệ 429, task duration, timeout và năng lực proxy pool để điều chỉnh.

Số slot không nên được chọn chỉ dựa trên số Worker. Nó cần phản ánh giới hạn của nguồn, năng lực mạng và mục tiêu chất lượng dữ liệu.

Proxy Apache Airflow giúp anh em kiểm soát lịch chạy, proxy credential, concurrency, retry và chất lượng dữ liệu trong cùng một workflow. Khi triển khai, hãy ưu tiên quyền truy cập hợp lệ, Airflow Pools, object storage và giám sát theo domain. Đồng hành cùng Proxy.vn - Nhà cung cấp dịch vụ proxy chất lượng hàng đầu Việt Nam, anh em có thể xây nền tảng dữ liệu ổn định hơn với Proxy Apache Airflow.

Proxy Apache Airflow: 5 Bước Xây Data Pipeline Ổn Định

Proxy Apache Airflow giải quyết vấn đề gì?