Kéo một triệu bản ghi qua API phân trang của khách hàng mà không mất dòng nào

Thử tính trước khi gõ code. Một hệ thống có 1.000.000 bản ghi, mỗi trang lấy 100 dòng, nghĩa là bạn cần 10.000 request. Nếu quên đặt limit mà API mặc định trả 10 dòng như Stripe thì con số đó thành 100.000.

Còn nếu API phân trang bằng offset, ở những trang cuối server phải quét qua gần một triệu dòng rồi bỏ đi, chỉ để trả về 100 dòng bạn cần.

Thử hình dung việc đầu tiên bạn được giao ở chỗ khách là kéo ticket, giao dịch hay tin nhắn từ hệ thống của họ để dựng pipeline hoặc đưa vào một agent. Khi đó mọi bước phía sau đều dựa vào vòng lặp đọc API này.

Nếu nó âm thầm làm trùng 5 dòng hoặc bỏ sót 5 dòng thì mọi con số trong buổi demo đều sai.

Bài này hướng dẫn viết một bộ kéo dữ liệu bằng Python cho ba kiểu phân trang phổ biến, có bộ lọc, batch, xử lý rate limit và checkpoint. Bạn cần Python 3, thư viện requests và một API key thử nghiệm. Phần code dùng một API giả định với tên trường đã được đơn giản hoá. Khi làm thật, bạn thay bằng tên trường trong tài liệu của khách.

Bước 1: Đọc tài liệu để biết API thuộc kiểu nào

Trước khi viết vòng lặp, cần trả lời hai câu: API đi sang trang sau bằng cách nào, và làm sao biết đã hết dữ liệu. Ba API công khai dưới đây đại diện cho ba cách làm bạn sẽ gặp nhiều nhất.

API	Cách đi tiếp	Kích thước trang	Tín hiệu dừng
Confluence (Atlassian)	Offset với start và limit	Atlassian khuyên luôn đặt limit tường minh	Response không còn link next
Stripe	Cursor theo ID đối tượng, starting_after / ending_before	limit từ 1 đến 100, mặc định 10	has_more là false
Slack	Cursor do server cấp	Slack khuyên 100 đến 200 mỗi lần	next_cursor rỗng, null hoặc không có

Điểm chung của cả ba là tín hiệu dừng do server quyết định, không phải client. Đây là quy tắc đầu tiên: đừng bao giờ tính trước “có 50 trang” rồi lặp 50 lần.

Xong bước này, bạn phải nói được ba điều về API của khách: tham số dùng để sang trang tên là gì, limit tối đa bao nhiêu, và trường nào báo đã hết dữ liệu.

Bước 2: Vòng lặp offset, có limit tường minh

Với API kiểu Confluence, vòng lặp đi theo link next thay vì tự cộng start. Đoạn dưới đây đã được đơn giản hoá: links.next là tên trường giả định, bạn cần đối chiếu với response thật.

import requests

def iter_offset(session, url, limit=100):
    params = {"start": 0, "limit": limit}  # luôn đặt limit tường minh
    while url:
        resp = session.get(url, params=params)
        resp.raise_for_status()
        page = resp.json()
        yield from page["results"]
        url = page.get("links", {}).get("next")  # hết link next nghĩa là hết dữ liệu
        params = None  # link next thường đã chứa sẵn start/limit

Kiểm tra: in ra số dòng của mỗi trang. Nếu trang nào cũng có đúng 25 dòng dù bạn xin 100 thì server đang áp giới hạn riêng. Đó là lý do Atlassian khuyên đặt limit để chắc chắn về số kết quả mỗi trang.

Vì sao offset hỏng khi dữ liệu đang chạy?

Offset có hai điểm yếu. Cái dễ thấy là tốc độ: offset càng lớn, database càng phải quét và bỏ đi nhiều dòng. Cái nguy hiểm hơn gọi là page drift: bảng có thêm bản ghi mới trong lúc bạn đang lật trang.

Thử hình dung bạn đọc ticket theo thứ tự mới nhất trước. Trang 1 trả về dòng 1 đến 100. Trong lúc đó có 5 ticket mới được tạo, và vì mới nhất nên chúng chen lên đầu danh sách, đẩy mọi dòng cũ lùi xuống 5 vị trí.

Khi bạn xin trang 2 với offset 100, server trả về những dòng trước đó ở vị trí 96 đến 195, nên 5 dòng bị lặp lại. Nếu là xoá thay vì thêm thì ngược lại, 5 dòng sẽ bị bỏ qua.

Bước 3: Vòng lặp cursor, dừng theo server

Cursor tránh được page drift vì nó đánh dấu “sau bản ghi này” chứ không đánh dấu “sau vị trí thứ N”. Stripe dùng chính ID của đối tượng cuối trang làm starting_after. Đoạn code dưới đây viết theo kiểu Stripe, trong đó items là tên trường giả định.

def iter_cursor(session, url, limit=100):
    params = {"limit": limit}  # Stripe cho phép 1-100
    while True:
        resp = session.get(url, params=params)
        resp.raise_for_status()
        page = resp.json()
        items = page["items"]
        yield from items
        if not page["has_more"] or not items:
            break
        params["starting_after"] = items[-1]["id"]

Với API kiểu Slack, bạn đổi điều kiện dừng: kiểm tra next_cursor rỗng, null hoặc không tồn tại, và nhớ xử lý cả ba trường hợp này. Stripe cũng có thư viện client hỗ trợ auto-pagination. Nếu khách cho phép dùng thì nên dùng, nhưng vẫn nên hiểu vòng lặp bên dưới để debug khi có sự cố.

Cursor cũng có rủi ro riêng. Với cách phân trang kiểu seek/keyset, nếu bản ghi làm mốc bị xoá thì ID đó có thể không còn hợp lệ, và request tiếp theo sẽ lỗi. Vì vậy bạn phải log lại cursor cuối cùng mỗi khi gặp lỗi, đừng chỉ log mã HTTP.

Bước 4: Dùng bộ lọc để thu hẹp phạm vi cần kéo

Đừng kéo cả triệu dòng nếu bạn chỉ cần dữ liệu quý vừa rồi. Bộ lọc theo thời gian hay trạng thái giúp cắt bớt phạm vi ngay từ request đầu tiên.

Nếu phần còn lại vẫn lớn, bạn có thể tự chia thành nhiều cửa sổ thời gian, mỗi cửa sổ chạy một vòng lặp cursor riêng; đây là cách tổ chức việc ở phía client, không phải một kiểu phân trang.

Keyset pagination là chuyện khác: nó lấy giá trị lọc của trang trước để xác định trang sau. Nếu API của khách không cấp cursor mà chỉ có bộ lọc và sắp xếp, bạn tự dựng keyset bằng cách lấy created_at và id của dòng cuối trang làm điều kiện cho request kế tiếp.

Dù đi theo cách nào, thứ tự sắp xếp phải ổn định. Chỉ sắp theo created_at là chưa đủ: hai bản ghi trùng timestamp có thể đổi chỗ giữa hai request, và nếu chúng nằm ở ranh giới trang thì một dòng bị lặp, một dòng bị sót. Thêm id làm tiêu chí phụ để hai dòng không bao giờ “hoà” nhau.

Còn hai chi tiết thường làm mất cả buổi chiều. Cách truyền mảng vào query không thống nhất: theo hướng dẫn OpenAPI, kiểu form có thể là ?color=blue,green,red hoặc ?color=blue&color=green, nên bạn phải xem API của khách chờ dạng nào. Ngoài ra, nhiều API dùng tiền tố - để sắp xếp giảm dần, ví dụ sort=-created_at.

params = {
    "limit": 100,
    "status": "closed,resolved",   # hoặc lặp tham số, tuỳ API
    "sort": "created_at,id",       # tăng dần, id phá thế hoà (cú pháp nhiều trường là giả định)
}

Bộ lọc và sắp xếp chỉ nhanh khi phía sau có index trong database. Nếu một bộ lọc làm mỗi request chậm hẳn đi, hãy hỏi đội kỹ thuật của khách xem trường đó có index không, trước khi kết luận là API “chậm”.

Bước 5: Batch, 429 và checkpoint

Kéo dữ liệu lớn thì sớm muộn cũng chạm rate limit. API làm đúng chuẩn sẽ trả HTTP 429 Too Many Requests, và script của bạn phải coi đó là tín hiệu để chờ chứ không phải để dừng hẳn. Đoạn dưới đây là cách làm đơn giản hoá với thời gian chờ tăng dần; nếu tài liệu của khách quy định cách retry riêng thì làm theo tài liệu.

import time, json, os

def get_with_retry(session, url, params, max_tries=6):
    for attempt in range(max_tries):
        resp = session.get(url, params=params)
        if resp.status_code != 429:
            resp.raise_for_status()
            return resp.json()
        time.sleep(2 ** attempt)  # chờ 1, 2, 4, 8... giây
    raise RuntimeError("Vẫn bị 429 sau nhiều lần thử")

def save_checkpoint(cursor, path="checkpoint.json"):
    with open(path, "w") as f:
        json.dump({"cursor": cursor}, f)

def load_checkpoint(path="checkpoint.json"):
    if not os.path.exists(path):
        return None
    with open(path) as f:
        return json.load(f).get("cursor")

Hai hàm này chỉ có ích khi được nối vào vòng lặp cursor. Nguyên tắc quan trọng nhất: chỉ lưu cursor sau khi batch đã nằm trên đĩa. Nếu lưu cursor trước rồi script chết lúc đang ghi file, lần chạy sau sẽ bắt đầu sau những dòng chưa bao giờ được ghi.

def flush(batch, out_dir):
    last_id = batch[-1]["id"]
    with open(f"{out_dir}/batch_{last_id}.jsonl", "w") as f:
        for row in batch:
            f.write(json.dumps(row) + "\n")
    save_checkpoint(last_id)  # lưu cursor SAU khi dữ liệu đã ghi xong

def pull_all(session, url, out_dir, limit=100, batch_size=1000):
    params = {"limit": limit}
    cursor = load_checkpoint()
    if cursor:
        params["starting_after"] = cursor  # chạy tiếp từ chỗ đã lưu
    batch = []
    while True:
        page = get_with_retry(session, url, params)
        items = page["items"]
        batch.extend(items)
        if len(batch) >= batch_size:
            flush(batch, out_dir)
            batch = []
        if not page["has_more"] or not items:
            break
        params["starting_after"] = items[-1]["id"]
    if batch:
        flush(batch, out_dir)

Thử hình dung script chết ở request thứ 7.000 trong 10.000. Cursor trong file là ID cuối của batch gần nhất đã ghi xong, nên lần chạy sau kéo lại vài trang chưa kịp ghi rồi đi tiếp, không phải làm lại từ đầu và không bỏ sót dòng nào.

Tên file theo ID cuối batch cũng giúp một lần chạy lại không ghi đè lên dữ liệu cũ.

Với các lần đồng bộ lặp lại, nên kiểm tra xem API có hỗ trợ cache bằng ETag hoặc Last-Modified không, để khỏi kéo lại những gì chưa thay đổi.

Kiểm tra cuối: so số ID duy nhất với tổng số dòng đã ghi. Hai số này phải bằng nhau. Nếu lệch, bạn đang gặp page drift, thứ tự sắp xếp không ổn định hoặc retry ghi trùng, và phải xử lý trước khi đưa dữ liệu cho ai dùng.

Kỹ năng này hiện ra thế nào ở chỗ khách hàng?

Khách sẽ không hỏi bạn “có biết cursor pagination không”. Họ hỏi “vì sao dashboard thiếu 312 ticket so với hệ thống gốc”. Người trả lời được câu đó trong một giờ, nhờ đã log cursor, đếm ID duy nhất và biết page drift là gì, sẽ được tin tưởng giao phần việc tiếp theo.

Khi đọc JD của các vị trí FDE, hãy để ý những yêu cầu liên quan đến tích hợp với hệ thống của khách hoặc dựng pipeline dữ liệu từ hệ thống bên thứ ba: đó là chỗ kỹ năng này được dùng hằng ngày.

Trong CV, thay vì ghi “tích hợp API”, hãy viết cụ thể: kéo bao nhiêu bản ghi, qua kiểu phân trang nào, xử lý rate limit ra sao, và đã kiểm tra tính toàn vẹn của dữ liệu bằng cách nào.

Một dự án nhỏ trên GitHub kéo dữ liệu từ Slack hoặc Stripe sandbox, có checkpoint và bước đối soát số dòng, sẽ thuyết phục hơn một dòng mô tả chung chung.

Script chạy được trên laptop của bạn mới là điểm xuất phát. Thứ khách thật sự kiểm tra là số dòng có khớp với hệ thống gốc hay không, kể cả sau khi script bị ngắt và chạy lại lúc nửa đêm.