Khi đội nguồn của khách lặng lẽ đổi schema: thực hành viết hợp đồng dữ liệu để pipeline không gãy

Một schema chưa phải là hợp đồng. Trong bài phân tích về Kafka data contracts, Conduktor định nghĩa hợp đồng là schema, cộng thêm các quy tắc về cách nó được phép thay đổi và một người chịu trách nhiệm về nó. Thiếu hai phần sau, bạn chỉ có một bản mô tả dữ liệu tại một thời điểm.

Ở vị trí FDE, bạn sẽ va vào khoảng trống này rất sớm. Bạn dựng pipeline trên dữ liệu của khách, còn đội phát triển hệ thống nguồn của khách thì không báo cáo cho bạn. Họ có thể đổi tên một field vào chiều thứ Sáu, và đến sáng thứ Hai dashboard đã sai mà không ai biết lý do.

Bài này hướng dẫn bạn dựng bốn thứ trên laptop. Đó là một file hợp đồng, một test phía consumer, một cấu hình kiểm tra ở ranh giới và một quy trình đổi field hai giai đoạn. Ví dụ xuyên suốt là giả định: một chuỗi bán lẻ lưu đơn hàng trong collection don_hang của một document database.

Vì sao document database dễ “trôi” schema?

Document database không lưu dữ liệu theo hàng và cột cố định mà lưu thành các document linh hoạt. Tài liệu của MongoDB nói rõ các document trong cùng một collection không bắt buộc phải có cùng field. Wikipedia cũng mô tả các document cùng collection có thể khác nhau về field lẫn cấu trúc.

Vì vậy bạn không thể nhìn một document mẫu rồi kết luận đó là schema. Có thể bản ghi năm ngoái có total_amount còn bản ghi tuần này lại không có. Muốn biết, bạn phải đọc dữ liệu thật trước khi viết bất kỳ quy tắc nào.

Chuẩn bị: Python 3, một trình soạn thảo, quyền đọc một mẫu dữ liệu từ nguồn của khách hoặc vài document JSON bạn tự tạo để tập.

Bước 1: chỉ đưa vào hợp đồng những field bạn thật sự đọc

Mở code consumer của bạn ra và liệt kê các field mà nó truy cập, không phải tất cả field có trong nguồn. Trong ví dụ, pipeline chỉ dùng order_id, total_amount và status, nên hợp đồng chỉ cần ba field này.

Quy tắc schema validation của MongoDB không cần bao phủ mọi field trong document. Bạn có thể bắt đầu với một hợp đồng nhỏ, chặt ở đúng những chỗ quan trọng và để phần còn lại linh hoạt.

Kiểm tra: với mỗi field trong danh sách, bạn phải chỉ ra được dòng code đang đọc nó. Field nào không chỉ ra được thì bỏ khỏi danh sách.

Bước 2: viết file hợp đồng đủ ba phần

File dưới đây theo định dạng tự đặt, không theo chuẩn của công cụ nào. Điều cần chú ý là nó có đủ ba phần trong định nghĩa ở đầu bài.

# contract.yaml — định dạng tự đặt, dùng để thống nhất với đội nguồn
dataset: don_hang
owner: "<tên người bên đội nguồn của khách>"
consumer_contact: "<tên bạn>"
fields:
  order_id:     { type: string, required: true }
  total_amount: { type: number, required: true, min: 0 }
  status:       { type: string, required: true, enum: [new, paid, cancelled] }
evolution:
  them_field_moi: "được phép, báo trước"
  xoa_hoac_doi_nghia_field: "chỉ làm theo hai giai đoạn"
  doi_ten_field: "coi như xoá field cũ + thêm field mới"

Dòng owner quan trọng nhất trong file. Khi schema đổi lúc nửa đêm, nó cho biết bạn cần gọi ai. Nếu khách không chỉ định được người nào, bạn nên ghi điều đó vào biên bản họp như một rủi ro của dự án, đừng tự điền tên mình vào.

Kiểm tra: đưa file cho đội nguồn đọc và xin họ xác nhận bằng văn bản. Một hợp đồng chỉ có bạn ký thì chưa có giá trị.

Bước 3: test phía consumer

Theo cách tiếp cận consumer-driven contract testing, consumer tự viết test để nói mình chờ đợi gì từ provider. Pact là framework phổ biến cho việc này. Đoạn dưới đây là bản Python thuần đã đơn giản hoá, dùng để hiểu ý tưởng, không phải Pact.

CONTRACT = {"order_id": str, "total_amount": (int, float), "status": str}
ALLOWED_STATUS = {"new", "paid", "cancelled"}

def check(doc):
    errors = []
    for field, typ in CONTRACT.items():
        if field not in doc:
            errors.append(f"thiếu {field}")
        elif not isinstance(doc[field], typ):
            errors.append(f"{field} sai kiểu")
    if isinstance(doc.get("total_amount"), (int, float)) and doc["total_amount"] < 0:
        errors.append("total_amount âm")
    # chỉ kiểm tra enum khi status có mặt và đúng kiểu, tránh báo lỗi trùng
    if isinstance(doc.get("status"), str) and doc["status"] not in ALLOWED_STATUS:
        errors.append("status lạ")
    return errors

print(check({"order_id": "A1", "total_amount": 250000, "status": "paid"}))
print(check({"order_id": "A2", "amount_vnd": 250000, "status": "paid"}))

Lần gọi đầu trả về []. Lần gọi thứ hai trả về ['thiếu total_amount'], đúng tình huống đội nguồn đổi tên field mà không báo. Một document thiếu hẳn status sẽ chỉ nhận một lỗi thiếu status, không bị báo thêm status lạ.

Kiểm tra: chạy hàm trên mẫu dữ liệu thật. Nếu có document cũ thiếu field, bạn vừa phát hiện chỗ schema đã trôi trong quá khứ. Hãy ghi lại trước khi bật chặn ở bước sau.

Bước 4: thực thi ở ranh giới, bắt đầu bằng cảnh báo

Conduktor viết rằng một hợp đồng không thực thi được chỉ là lời nhắc nhở, không ràng buộc được ai. Test ở bước 3 chạy phía bạn. Lớp chặn thứ hai cần đặt ngay tại nơi dữ liệu được ghi vào.

Với MongoDB, bạn có thể bật schema validation để khoá schema khi cần, bằng các quy tắc về kiểu dữ liệu và khoảng giá trị. Mặc định, mọi insert hoặc update tạo ra document không hợp lệ đều bị từ chối. Bạn cũng có thể cấu hình để document không hợp lệ vẫn được ghi, kèm một dòng cảnh báo trong log.

# MÔ PHỎNG Ý TƯỞNG — không phải cú pháp MongoDB.
# Tra "Schema Validation" trong MongoDB Manual để có cú pháp chính xác.
collection: don_hang
rules: order_id string bắt buộc; total_amount number >= 0; status thuộc enum
khi_vi_pham: warn     # giai đoạn 1: chỉ ghi log
# sau khi log sạch một thời gian: chuyển về mặc định (reject)

Nên bắt đầu bằng cảnh báo vì đội nguồn có thể đang có những luồng ghi mà cả hai bên chưa biết. Nếu bật reject ngay, bạn có thể làm hỏng hệ thống vận hành của khách, và đó là cách nhanh nhất để mất lòng tin của họ.

Kiểm tra: cho log cảnh báo chạy một thời gian, xem lại từng loại vi phạm cùng đội nguồn, sửa nguồn hoặc sửa hợp đồng, rồi mới chuyển sang reject.

Bước 5: khi đội nguồn muốn đổi tên field thì làm thế nào?

Giả sử đội nguồn muốn đổi total_amount thành amount_vnd. Nguyên tắc Conduktor đưa ra rất rõ: không được xoá hay đổi mục đích một field khi vẫn còn hệ thống đọc nó. Cách làm là xoá theo hai giai đoạn, chuyển field sang tuỳ chọn trước rồi mới xoá.

Áp vào ví dụ, giai đoạn một là đội nguồn ghi cả hai field. Bạn sửa consumer để đọc amount_vnd và chỉ dùng total_amount khi thiếu field mới, đồng thời cập nhật contract.yaml để total_amount thành tuỳ chọn. Giai đoạn hai bắt đầu khi bạn xác nhận không còn consumer nào đọc field cũ. Lúc đó đội nguồn mới ngừng ghi total_amount.

Chữ BACKWARD hay bị hiểu nhầm chiều kiểm tra. BACKWARD compatibility nghĩa là schema mới đọc được dữ liệu cũ, nên phía consumer phải nâng cấp trước. Thứ tự trong ví dụ trên đúng như vậy: bạn sửa consumer xong thì đội nguồn mới được bỏ field cũ.

Những lỗi hay gặp

Lỗi thứ nhất là chép toàn bộ document mẫu vào hợp đồng. Hợp đồng sẽ phình to, và mỗi thay đổi vô hại ở một field bạn không dùng cũng thành một sự cố. Lỗi thứ hai là bật reject từ ngày đầu, trước khi biết dữ liệu cũ bẩn đến mức nào.

Lỗi thứ ba khó thấy hơn: coi việc đổi tên là thay đổi nhỏ. Với consumer, đổi tên chính là xoá một field. Lỗi cuối cùng là để trống ô owner, và khi đó hợp đồng chỉ còn là tài liệu.

Kỹ năng này xuất hiện thế nào ở chỗ khách?

Trong tuần đầu của một deployment, hợp đồng dữ liệu là thứ mở ra cuộc nói chuyện với đội nguồn. Bạn không đến để xin quyền truy cập toàn bộ database. Bạn đến với ba field, một quy tắc thay đổi và câu hỏi ai sẽ ký tên.

Khi đọc JD, hãy để ý các cụm như “data quality”, “schema evolution” hay “integrate with customer systems”. Trong CV, đừng viết “có kinh nghiệm MongoDB”. Hãy viết rằng bạn đã đổi tên một field trên dữ liệu production qua hai giai đoạn và pipeline không phải dừng.

Trước buổi phỏng vấn FDE, hãy chuẩn bị một câu chuyện thật thay vì định nghĩa schema: lần gần nhất dữ liệu của khách thay đổi, bạn phát hiện trước hay khách phát hiện trước?