# 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

> Vòng lặp đọc dữ liệu chỉ có vài dòng code, nhưng nếu chọn sai cách phân trang, script sẽ chạy cả đêm hoặc trả về dữ liệu trùng mà bạn không hề hay biết.

Bản gốc: https://fdetimes.net/vi/bach-khoa/thuc-hanh-keo-du-lieu-lon-qua-api-phan-trang/

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.

```python
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.

**Điểm mấu chốt:** Offset chỉ an toàn khi dữ liệu đứng yên. Hệ thống production của khách thì hiếm khi đứng yên.

## 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.

```python
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`.

```python
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.

```python
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.

```python
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.

**Thử ngay tuần này:**

- Mở tài liệu pagination của Stripe, Slack và Confluence, ghi lại vào một bảng tham số trang, giới hạn limit và tín hiệu dừng của từng bên.
- Viết hàm pull_all() có limit tường minh, retry khi gặp 429 và lưu cursor sau mỗi batch đã ghi xuống đĩa, rồi kill giữa chừng và chạy lại để xem số ID duy nhất có khớp không.
- Tự dựng page drift trên một bảng local: vừa đọc theo offset vừa chèn dòng mới, rồi đếm số ID bị trùng.

## Nguồn

- [Everything You Need to Know About API Pagination](https://nordicapis.com/everything-you-need-to-know-about-api-pagination/)

- [Pagination in the REST API - Atlassian](https://developer.atlassian.com/server/confluence/pagination-in-the-rest-api/)

- [Pagination | Stripe API Reference](https://docs.stripe.com/api/pagination)

- [Pagination | Slack Developer Docs](https://docs.slack.dev/apis/web-api/pagination)

- [Unlocking the Power of API Pagination: Best Practices and Strategies](https://dev.to/pragativerma18/unlocking-the-power-of-api-pagination-best-practices-and-strategies-4b49)

- [How to Implement Filtering and Sorting in REST APIs](https://oneuptime.com/blog/post/2026-01-26-rest-api-filtering-sorting/view)

- [Describing Parameters - OpenAPI Guide](https://swagger.io/docs/specification/describing-parameters/)