Series dbt · Bài 7/10. Khi bảng có nhiều triệu dòng, xây lại toàn bộ sau mỗi lần chạy có thể tốn thời gian và tài nguyên. Incremental model chỉ xử lý một tập ứng viên mới hoặc thay đổi, rồi cập nhật relation đích theo chiến lược đã chọn. Lợi ích này đi kèm trách nhiệm: chọn đúng watermark, xử lý cập nhật cũ, dữ liệu đến muộn và xóa.

Tiếp tục dbt-retail sau bài 6 với dữ liệu đã hoàn nguyên. Bài tạo model int_orders_incremental riêng ở grain đơn hàng, không thay fct_sales đang có. Giữ Core 1.9.8, dbt-postgres 1.9.1 và PostgreSQL 16.6 của lab; không dùng tính năng chỉ có ở runtime mới. Khi triển khai thật cần chọn phiên bản còn được hỗ trợ và kiểm thử adapter/strategy.

1. Không chỉ là thêm WHERE ngày hôm nay

Một đơn tạo tháng trước có thể đổi từ pending thành paid hôm nay. Nếu lọc theo order_date mới, bạn bỏ sót cập nhật đó. Vì vậy, cần phân biệt thời gian nghiệp vụ của đơn, thời gian bản ghi thay đổi và thời gian ingestion đưa bản ghi tới warehouse. Lab bổ sung updated_at làm đồng hồ thay đổi do nguồn cung cấp.

unique_key nói cho chiến lược biết dòng nào tương ứng với dòng cũ. Nó không tự bảo đảm dữ liệu duy nhất và không mặc định tạo constraint vật lý. Nếu batch mới có nhiều bản cùng khóa, cần chọn một phiên bản theo quy tắc rõ ràng trước khi ghi. Lab nguồn có primary key order_id nên mỗi lần đọc chỉ có một bản hiện tại.

2. Bổ sung metadata thay đổi cho nguồn lab

Tạo init/02_incremental_columns.sql. File bổ sung cột nếu chưa có, không xóa dữ liệu. Mốc thời gian cố định dùng để mọi người tái hiện cùng thí nghiệm, không phải thời điểm ingestion thật. Chỉ chạy trên database lab.

alter table raw.orders
    add column if not exists updated_at timestamptz
    not null default '2026-01-01 00:00:00+00';

alter table raw.orders
    add column if not exists is_deleted boolean
    not null default false;

create index if not exists orders_updated_at_idx
    on raw.orders(updated_at);

docker compose exec -T postgres psql -v ON_ERROR_STOP=1 \
  -U dbt_lab -d retail < init/02_incremental_columns.sql

Trong hệ thống thật, nguồn hoặc ingestion phải cập nhật updated_at đúng mỗi khi các cột cần phân tích đổi. Thêm cột default vào warehouse không tự tạo cơ chế CDC. Index có thể hỗ trợ đọc theo timestamp, nhưng trên năm dòng PostgreSQL vẫn có thể chọn sequential scan; đó không tự là lỗi hiệu năng.

3. Viết incremental model ở grain order_id

models/intermediate/int_orders_incremental.sql

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='delete+insert',
    on_schema_change='fail'
) }}

select
    order_id,
    customer_id,
    ordered_at::date as order_date,
    lower(trim(status)) as order_status,
    updated_at,
    is_deleted
from {{ source('shop', 'orders') }}
{% if is_incremental() %}
where updated_at >= (
    select coalesce(max(updated_at), timestamptz '1900-01-01 00:00:00+00')
    from {{ this }}
) - interval '2 days'
{% endif %}

config trong file ghi đè materialized=view của folder intermediate. Relation đích vẫn nằm ở dev_intermediate. this là relation hiện tại của chính model; is_incremental() chỉ đúng khi relation đích đã tồn tại ở dạng phù hợp, model cấu hình incremental và lệnh không yêu cầu full refresh. Vì vậy SQL phải hợp lệ cả khi khối WHERE được render và khi nó không được render.

Với delete+insert, các khóa xuất hiện trong tập ứng viên được xóa khỏi đích rồi chèn lại bằng bản mới theo cơ chế adapter. Đây là cập nhật trạng thái hiện tại, không lưu nhiều phiên bản lịch sử. Những khóa không có trong tập ứng viên vẫn nằm nguyên ở đích. Chạy lại cùng batch không được làm tăng số dòng nếu nguồn và unique_key đúng.

Không lọc is_deleted=false trước bước ghi incremental. Nếu một đơn đã có ở đích rồi bị đánh dấu xóa mềm, loại nó khỏi tập ứng viên sẽ khiến dòng cũ không bao giờ được cập nhật thành deleted. Model giữ cả cờ xóa; model tiêu thụ mới lọc những đơn đang hoạt động theo quy tắc riêng.

models/intermediate/_incremental.yml

version: 2
models:
  - name: int_orders_incremental
    description: 'Current order state including soft-delete flag'
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: updated_at
        tests: [not_null]
      - name: is_deleted
        tests: [not_null]

4. Lần đầu đọc đủ, lần sau đọc ứng viên

dbt run --profiles-dir profiles --select int_orders_incremental
dbt test --profiles-dir profiles --select int_orders_incremental

docker compose exec -T postgres psql -U dbt_lab -d retail \
  -c "select * from dev_intermediate.int_orders_incremental order by order_id;"

Lần đầu model chưa có nên khối is_incremental không tạo WHERE; cả năm đơn được đọc. Chạy lại ngay có thể vẫn đọc cả năm vì dữ liệu đều nằm trong lookback. Với dữ liệu nhỏ, incremental không tự nhanh hơn full table; mục tiêu lab là chứng minh tính đúng, không tạo benchmark tốc độ.

Lookback hai ngày ở đây tính từ max updated_at trong đích, không phải hai ngày tính từ đồng hồ máy. Nếu không có dữ liệu mới, cùng cửa sổ có thể được đọc lại. Dùng >= giúp xử lý lại các timestamp bằng ranh giới, với unique_key để tránh append lặp. Điều này không giải quyết bản ghi ngoài cửa sổ hoặc timestamp không đáng tin.

5. Thử thay đổi: update, insert và soft delete

5.1. Đơn cũ đổi trạng thái

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
update raw.orders
set status = 'paid', updated_at = '2026-01-03 10:00:00+00'
where order_id = 1004;
SQL

dbt run --profiles-dir profiles --select int_orders_incremental

Đơn 1004 tạo ngày 2 nhưng updated_at ngày 3 nên được chọn. Đích vẫn có năm dòng và 1004 chuyển paid. Đây là cách quan sát vì sao lọc theo updated_at khác lọc theo ngày tạo. Các mart cũ không được tự cập nhật chỉ vì bạn chạy model này; nó đang là một nhánh thí nghiệm riêng.

5.2. Đơn mới có ngày nghiệp vụ cũ

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
insert into raw.orders
    (order_id, customer_id, ordered_at, status, updated_at, is_deleted)
values
    (1006, 3, '2026-01-01', 'pending', '2026-01-04 12:00:00+00', false)
on conflict (order_id) do update set
    status = excluded.status,
    updated_at = excluded.updated_at,
    is_deleted = excluded.is_deleted;
SQL

dbt run --profiles-dir profiles --select int_orders_incremental

Đích tăng thành sáu dòng. Đơn 1006 có ngày nghiệp vụ cũ nhưng timestamp thay đổi mới, nên vẫn được nhận. Đơn pending chưa có item chỉ là dữ liệu thử, không được coi là doanh thu paid. Bài không thêm dòng hàng cho nó để giữ phép thí nghiệm đơn giản.

5.3. Đánh dấu xóa mềm

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
update raw.orders
set is_deleted = true, updated_at = '2026-01-05 12:00:00+00'
where order_id = 1005;
SQL

dbt run --profiles-dir profiles --select int_orders_incremental

docker compose exec -T postgres psql -U dbt_lab -d retail \
  -c "select order_id, order_status, is_deleted from dev_intermediate.int_orders_incremental order by order_id;"

Đích vẫn sáu dòng, nhưng 1005 có is_deleted=true. Truy vấn trạng thái đang hoạt động phải thêm WHERE NOT is_deleted. Đừng so doanh thu của nhánh này với star schema cũ rồi kết luận mâu thuẫn: fct_sales cũ chưa được thiết kế để dùng cờ xóa mới. Khi chuyển đổi production phải đưa cùng định nghĩa xóa vào toàn bộ luồng liên quan.

6. Chứng minh giới hạn của cửa sổ lookback

Watermark lúc này là ngày 5; lookback bắt đầu khoảng ngày 3. Cố ý sửa đơn 1002 với updated_at rất cũ để mô phỏng một bản đến muộn có timestamp nguồn nằm ngoài cửa sổ. Đây là dữ liệu sai thời gian có chủ đích để học, không phải cách cập nhật nguồn thông thường.

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
update raw.orders
set status = 'pending', updated_at = '2025-12-20 10:00:00+00'
where order_id = 1002;
SQL

dbt run --profiles-dir profiles --select int_orders_incremental

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
select order_id, status, updated_at from raw.orders where order_id = 1002;
select order_id, order_status, updated_at
from dev_intermediate.int_orders_incremental where order_id = 1002;
SQL

Nguồn đã pending nhưng đích có thể vẫn cancelled với bản cũ. Model chạy thành công mà dữ liệu chưa đồng bộ: đó là giới hạn của tiêu chí chọn ứng viên. Mở lookback lớn hơn chỉ mở rộng phạm vi bắt được, không tạo bảo đảm vô hạn. Khi nguồn có độ trễ không chặn trên, cần watermark ingestion/CDC phù hợp hoặc cơ chế backfill và đối soát có chủ đích.

7. Đối soát và sửa bằng full refresh có kiểm soát

tests/assert_incremental_orders_match_source.sql

with expected as (
    select
        order_id, customer_id, ordered_at::date as order_date,
        lower(trim(status)) as order_status, updated_at, is_deleted
    from {{ source('shop', 'orders') }}
),
actual as (
    select order_id, customer_id, order_date, order_status, updated_at, is_deleted
    from {{ ref('int_orders_incremental') }}
),
missing_or_changed as (
    select * from expected
    except
    select * from actual
),
extra_or_old as (
    select * from actual
    except
    select * from expected
)
select 'source_only' as issue, * from missing_or_changed
union all
select 'target_only' as issue, * from extra_or_old

dbt test --profiles-dir profiles --select assert_incremental_orders_match_source
dbt run --profiles-dir profiles --full-refresh --select int_orders_incremental
dbt test --profiles-dir profiles --select assert_incremental_orders_match_source

Test phải fail ở tình huống cố ý bỏ sót và pass sau full refresh. Full refresh bỏ nhánh incremental để đọc lại nguồn đầy đủ, nên có thể tốn tài nguyên và ảnh hưởng relation đích. Không biến nó thành câu trả lời mặc định cho mọi lỗi production; cần ước lượng tải, thời gian, quyền và cách phục hồi trước.

Test so toàn nguồn/đích phù hợp lab nhỏ nhưng có thể đắt với bảng lớn. Production có thể đối soát theo partition, khóa bị ảnh hưởng hoặc tổng hợp kiểm soát, kèm một lịch kiểm tra rộng hơn. Tối ưu kiểm tra không được âm thầm bỏ các trường hợp late arrival mà chính mô hình incremental có nguy cơ bỏ sót.

8. Khi mở rộng sang fact, phải xét mọi đầu vào thay đổi

Một fact dòng hàng có thể phụ thuộc order, item và dimension. Chỉ lọc item.updated_at sẽ bỏ sót trường hợp order_status đổi mà item không đổi. Cần xác định tập order_id bị tác động từ cả orders và items, sau đó dựng lại đủ các dòng của những đơn đó. Nếu dimension lịch sử đổi, cũng phải xem tác động tới khóa phiên bản và thời gian join.

Hard delete là trường hợp khác: một dòng biến mất khỏi source sẽ không còn nằm trong SELECT ứng viên, nên delete+insert không tự biết để xóa dòng cũ. Cần tombstone, soft delete, CDC hoặc một cơ chế đối chiếu/xóa riêng. on_schema_change=fail cũng chỉ làm thay đổi schema trở nên rõ ràng, không tự backfill giá trị mới cho lịch sử.

9. Hoàn nguyên để các bài sau chạy nhất quán

Khôi phục ba đơn đã sửa, bỏ đúng đơn pending 1006 do bài này tạo, sau đó full refresh model riêng. Không xóa volume hoặc các bảng khác. Những câu lệnh này chỉ dành cho lab với ID đã nêu.

docker compose exec -T postgres psql -U dbt_lab -d retail <<'SQL'
update raw.orders set status = 'pending', updated_at = '2026-01-06 12:00:00+00'
where order_id = 1004;
update raw.orders set status = 'cancelled', updated_at = '2026-01-06 12:00:00+00'
where order_id = 1002;
update raw.orders set is_deleted = false, updated_at = '2026-01-06 12:00:00+00'
where order_id = 1005;
delete from raw.orders where order_id = 1006;
SQL

dbt run --profiles-dir profiles --full-refresh --select int_orders_incremental
dbt build --profiles-dir profiles

Nghiệm thu: model incremental có năm đơn, mọi cờ xóa false; test parity pass; các mart cũ trở lại 400.000 VND. Giữ updated_at sau thí nghiệm như trên, không cần quay ngược về ngày 1. Bài tiếp theo dùng snapshot khách hàng riêng để phân biệt trạng thái hiện tại với lịch sử phiên bản.

Tài liệu tham khảo

• Incremental models và is_incremental
• Các chiến lược incremental, chọn đúng adapter/phiên bản
• Ý nghĩa unique_key
• Relation this
• on_schema_change