Project cuối chuỗi kết hợp các kỹ năng SQL thành một hệ thống phân tích bán hàng có thể chạy lại: nạp staging, kiểm tra dữ liệu, xây mart ngày, đối chiếu số liệu và xuất truy vấn KPI. Mục tiêu là một repository có thể trình bày trong portfolio, không phải một file SELECT dài.

Mục lục

1. Bài toán thực tế
2. Sản phẩm hoàn thành
3. Kiến thức cần dùng
4. Chuẩn bị môi trường
5. Dữ liệu thực hành
6. Xây dựng từng bước
7. Chương trình hoàn chỉnh
8. Kiểm tra kết quả
9. Lỗi thường gặp
10. Đưa vào thực tế
11. Bài tập thực hành
12. Project mở rộng
13. Tổng kết
14. Tài liệu tham khảo

1. Bài toán thực tế

Doanh nghiệp nhận bốn CSV mỗi ngày nhưng báo cáo được làm bằng copy-paste trong Excel. Khi file đến trễ hoặc job chạy lại, dữ liệu có thể bị nạp hai lần. Không có audit cho biết batch nào đã xử lý và bao nhiêu dòng lỗi.

Dashboard cần doanh thu thuần, số đơn, số khách, sản phẩm bán ra và AOV theo cửa hàng-ngày. Số liệu chỉ được công bố khi tổng mart đối chiếu với nguồn hợp lệ.

2. Sản phẩm hoàn thành

Sản phẩm gồm schema raw, staging và mart; bảng etl_runs; bảng quarantine; mart_daily_store_sales; view KPI tháng và bộ quality checks.

Pipeline nhận batch_id, chạy trong transaction, có thể chạy lại mà không nhân dữ liệu và ghi trạng thái thành công hoặc thất bại.

• Thiết kế project theo layer và grain.
• Nạp idempotent bằng khóa batch.
• Tách dữ liệu lỗi vào quarantine.
• Xây mart incremental theo ngày.
• Thực hiện reconciliation và audit.

3. Kiến thức cần dùng

Idempotent nghĩa là chạy lại cùng đầu vào cho cùng kết quả, không tạo duplicate. Có thể đạt bằng unique key, upsert hoặc xóa rồi nạp lại đúng partition trong transaction.

Staging giữ dữ liệu gần nguồn và thêm metadata. Mart định hình dữ liệu theo câu hỏi phân tích. Không cho dashboard truy cập trực tiếp raw vì logic sẽ bị lặp ở nhiều báo cáo.

Reconciliation so sánh số dòng và tổng tiền giữa các lớp. Sai lệch phải được giải thích bằng quarantine hoặc quy tắc nghiệp vụ.

4. Chuẩn bị môi trường

Repository gồm sql/01_schema.sql, 02_load_staging.sql, 03_validate.sql, 04_build_mart.sql, 05_checks.sql và README. Dữ liệu mẫu nhỏ được commit; dữ liệu nhạy cảm không được đưa vào Git.

Mỗi lần chạy tạo etl_run_id và ghi source_date, started_at, status. Nếu transaction rollback, cập nhật failure ở transaction riêng hoặc orchestration layer.

CREATE SCHEMA IF NOT EXISTS staging;
CREATE SCHEMA IF NOT EXISTS mart;
CREATE TABLE etl_runs(
 etl_run_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
 source_date date NOT NULL,batch_id text NOT NULL UNIQUE,
 started_at timestamptz NOT NULL DEFAULT now(),finished_at timestamptz,status text NOT NULL
);

5. Dữ liệu thực hành

orders và order_items dùng cùng định nghĩa các bài trước. Staging bổ sung batch_id, source_file và loaded_at. Grain mart là sale_date + store_id.

Đơn completed được tính doanh thu; cancelled bị loại; dòng item số lượng không dương, giá âm hoặc discount vượt gross vào quarantine.

CREATE TABLE mart.daily_store_sales(
 sale_date date NOT NULL,store_id bigint NOT NULL,
 order_count bigint NOT NULL,customer_count bigint NOT NULL,units_sold bigint NOT NULL,
 gross_revenue numeric(16,2) NOT NULL,discount_amount numeric(16,2) NOT NULL,net_revenue numeric(16,2) NOT NULL,
 batch_id text NOT NULL,updated_at timestamptz NOT NULL DEFAULT now(),
 PRIMARY KEY(sale_date,store_id)
);

6. Xây dựng từng bước

Bước 1: Đăng ký batch

Unique batch_id chặn việc cùng file được xử lý như hai batch khác nhau.

INSERT INTO etl_runs(source_date,batch_id,status)
VALUES(:source_date,:batch_id,'running')
RETURNING etl_run_id;

Kết quả mong đợi: Một run mới hoặc lỗi unique khi batch đã tồn tại.

Bước 2: Phân loại dòng item

Tính gross và error_reason trong CTE. Dữ liệu lỗi không đi vào mart.

CASE WHEN quantity<=0 THEN 'invalid_quantity'
 WHEN unit_price<0 THEN 'invalid_price'
 WHEN discount_amount<0 THEN 'negative_discount'
 WHEN discount_amount>quantity*unit_price THEN 'discount_exceeds_gross'
 ELSE NULL END AS error_reason

Kết quả mong đợi: Mỗi dòng hợp lệ có error_reason NULL.

Bước 3: Ghi quarantine

Lưu khóa và raw_record giúp nhóm nguồn sửa. Không chỉ ghi tổng số lỗi.

INSERT INTO quarantine_order_items(batch_id,order_id,product_id,error_reason,raw_record)
SELECT :batch_id,order_id,product_id,error_reason,to_jsonb(v)
FROM validated_items v WHERE error_reason IS NOT NULL;

Kết quả mong đợi: Tất cả item lỗi của batch có lý do.

Bước 4: Tổng hợp mart theo ngày

Chỉ item hợp lệ thuộc đơn completed được tổng hợp. COUNT DISTINCT bảo vệ số đơn.

SELECT o.order_date::date sale_date,o.store_id,
 COUNT(DISTINCT o.order_id) order_count,COUNT(DISTINCT o.customer_id) customer_count,
 SUM(v.quantity) units_sold,SUM(v.quantity*v.unit_price) gross_revenue,
 SUM(v.discount_amount) discount_amount,SUM(v.quantity*v.unit_price-v.discount_amount) net_revenue
FROM orders o JOIN validated_items v ON v.order_id=o.order_id
WHERE o.status='completed' AND v.error_reason IS NULL
GROUP BY o.order_date::date,o.store_id;

Kết quả mong đợi: Một dòng mỗi cửa hàng-ngày.

Bước 5: Upsert để chạy lại

Primary key xác định partition logic. ON CONFLICT thay giá trị thay vì chèn trùng.

INSERT INTO mart.daily_store_sales (
 sale_date,store_id,order_count,customer_count,units_sold,
 gross_revenue,discount_amount,net_revenue,batch_id
)
SELECT sale_date,store_id,order_count,customer_count,units_sold,
 gross_revenue,discount_amount,net_revenue,:batch_id
FROM daily_aggregated
ON CONFLICT(sale_date,store_id) DO UPDATE SET
 order_count=EXCLUDED.order_count,
 customer_count=EXCLUDED.customer_count,
 units_sold=EXCLUDED.units_sold,
 gross_revenue=EXCLUDED.gross_revenue,
 discount_amount=EXCLUDED.discount_amount,
 net_revenue=EXCLUDED.net_revenue,
 batch_id=EXCLUDED.batch_id,updated_at=now();

Kết quả mong đợi: Chạy lại cùng ngày cập nhật kết quả, không tăng số dòng.

Bước 6: Hoàn tất audit

Chỉ đánh success sau khi check quan trọng đạt. Transaction bảo đảm mart và audit nhất quán.

UPDATE etl_runs SET status='success',finished_at=now()
WHERE batch_id=:batch_id;

Kết quả mong đợi: Run có thời gian kết thúc và trạng thái.

Thiết kế lần chạy thất bại

Giả sử job ghi được nửa mart rồi mất kết nối. Transaction bảo đảm rollback, nhưng file nguồn vẫn tồn tại và orchestrator sẽ retry. batch_id cùng logic upsert giúp lần retry không tạo dữ liệu kép.

Nếu dữ liệu quá lớn để một transaction, chia theo partition có trạng thái và checkpoint. Mỗi partition vẫn cần commit boundary và reconciliation riêng.

Reconciliation như hợp đồng bàn giao

Báo cáo trước khi publish phải có source rows, accepted rows, quarantined rows, source amount và mart amount. Chênh lệch hợp lệ như cancelled phải xuất hiện thành một dòng giải thích, không được ẩn trong WHERE.

Lưu kết quả reconciliation theo etl_run_id để điều tra lịch sử. Khi số dashboard bị hỏi sau ba tháng, nhóm có thể tái hiện batch và rule thay vì chỉ nhìn dữ liệu hiện tại.

Portfolio cần chứng minh điều gì

README nên có kiến trúc, grain từng bảng, lệnh chạy, ảnh kết quả và quyết định trade-off. Một nhà tuyển dụng quan tâm khả năng giải thích vì sao dùng upsert, quarantine và transaction hơn số lượng dòng SQL.

Thêm test tự động trong CI với dataset nhỏ: lần đầu tạo mart, lần hai chạy lại, một dòng lỗi đi quarantine và tổng tiền khớp. Đây là bằng chứng project có tính kỹ thuật, không chỉ là ảnh dashboard.

Thiết kế dữ liệu mẫu có thể chứng minh

Dataset portfolio nên đủ nhỏ để commit nhưng chứa các trường hợp khó: một đơn nhiều item, một item discount vượt gross, một order cancelled, một khách mua hai cửa hàng và một ngày không có giao dịch. Mỗi dòng đặc biệt liên kết với một test trong README.

Không tạo dữ liệu ngẫu nhiên hoàn toàn nếu không cố định seed, vì kết quả mong đợi sẽ thay đổi. Dữ liệu deterministic giúp CI so sánh con số chính xác.

SELECT setseed(0.42);
-- Với dữ liệu sinh tự động, ghi seed và version generator vào README.

Stored procedure hay script orchestration

Procedure có thể đóng gói transaction và giảm round trip, nhưng versioning, logging và test đôi khi rõ hơn ở dbt hoặc orchestrator. Project học nên bắt đầu bằng SQL files tường minh rồi mới đóng gói khi boundary ổn định.

Không đặt logic không thể quan sát vào một procedure lớn. Mỗi bước vẫn cần row count, error count và duration.

CREATE OR REPLACE PROCEDURE build_daily_sales(p_source_date date,p_batch_id text)
LANGUAGE plpgsql AS $$
BEGIN
 DELETE FROM mart.daily_store_sales WHERE sale_date=p_source_date;
 -- INSERT tổng hợp đã được kiểm thử đặt tại đây
END; $$;

Checksum để kiểm tra chạy lại

Ngoài row count và sum, checksum trên chuỗi cột đã sắp xếp giúp phát hiện thay đổi chi tiết khi tổng vẫn khớp. Không dùng như bảo mật; đây là tín hiệu so sánh dataset.

Checksum phải ổn định về thứ tự và cách biểu diễn NULL, numeric. Ghi thuật toán trong test.

SELECT md5(string_agg(CONCAT_WS('|',sale_date,store_id,order_count,net_revenue),E'\n' ORDER BY sale_date,store_id)) dataset_checksum
FROM mart.daily_store_sales;

Late-arriving data và cửa sổ nạp lại

Đơn ngày hôm qua có thể hoàn tất hôm nay. Pipeline chỉ xử lý ngày hiện tại sẽ bỏ thay đổi. Chọn lookback ba hoặc bảy ngày và upsert lại các partition gần, hoặc dựa updated_at watermark.

Watermark cần overlap để chống bản ghi đến đúng biên và timezone. Audit lưu min, max updated_at đã xử lý.

SELECT * FROM staging.orders
WHERE updated_at>=:last_watermark-INTERVAL '10 minutes'
  AND updated_at<:current_watermark;

Tiêu chí nghiệm thu project 9/10

Project phải chạy từ database rỗng, tạo schema, nạp sample, sinh quarantine, xây mart và vượt test bằng một lệnh được mô tả. Chạy lần hai không đổi row count hoặc checksum.

README giải thích grain, rule, kiến trúc, trade-off và kết quả. Không cần nhiều công cụ; sự nhất quán và khả năng chứng minh quan trọng hơn việc gắn thêm Airflow hoặc dbt chỉ để tăng danh sách công nghệ.

SELECT status,COUNT(*) FROM etl_runs GROUP BY status;
SELECT batch_id,COUNT(*) quarantine_rows FROM quarantine_order_items GROUP BY batch_id;
SELECT sale_date,SUM(net_revenue) FROM mart.daily_store_sales GROUP BY sale_date ORDER BY sale_date;

7. Chương trình hoàn chỉnh

Chương trình hoàn chỉnh đặt validate, quarantine, aggregate và upsert trong transaction. Advisory lock hoặc orchestration ngăn hai run cùng source_date chạy đồng thời.

README mô tả cách reset database, nạp sample và chạy từng file. Kết quả mẫu giúp nhà tuyển dụng xác minh project.

BEGIN;

CREATE TEMP TABLE valid_items ON COMMIT DROP AS
SELECT oi.*,oi.quantity*oi.unit_price AS gross_revenue,
       oi.quantity*oi.unit_price-oi.discount_amount AS net_revenue
FROM staging.order_items oi
WHERE oi.batch_id=:batch_id
  AND oi.quantity>0 AND oi.unit_price>=0
  AND oi.discount_amount BETWEEN 0 AND oi.quantity*oi.unit_price;

DELETE FROM mart.daily_store_sales WHERE sale_date=:source_date;

INSERT INTO mart.daily_store_sales (sale_date,store_id,order_count,customer_count,units_sold,gross_revenue,discount_amount,net_revenue,batch_id)
SELECT o.order_date::date,o.store_id,COUNT(DISTINCT o.order_id),COUNT(DISTINCT o.customer_id),
       SUM(v.quantity),SUM(v.gross_revenue),SUM(v.discount_amount),SUM(v.net_revenue),:batch_id
FROM staging.orders o JOIN valid_items v ON v.order_id=o.order_id
WHERE o.batch_id=:batch_id AND o.status='completed'
GROUP BY o.order_date::date,o.store_id;

UPDATE etl_runs SET status='success',finished_at=now() WHERE batch_id=:batch_id;
COMMIT;

8. Kiểm tra kết quả

Kiểm tra conservation ở cấp dòng: raw items = valid items + quarantined items. Ở cấp tiền, mart net revenue phải bằng tổng valid completed items.

Chạy cùng batch hai lần trong môi trường test và xác nhận số dòng mart, tổng tiền không đổi. Thay đổi một dòng nguồn rồi chạy batch sửa phải cập nhật đúng partition.

SELECT sale_date,store_id,COUNT(*)
FROM mart.daily_store_sales GROUP BY sale_date,store_id HAVING COUNT(*)>1;

SELECT SUM(net_revenue) mart_total FROM mart.daily_store_sales WHERE sale_date=:source_date;
SELECT SUM(quantity*unit_price-discount_amount) source_total
FROM valid_completed_items WHERE order_date::date=:source_date;

9. Lỗi thường gặp

• TRUNCATE toàn mart mỗi ngày: Tốn kém và tăng phạm vi rủi ro.
• Không có batch_id: Không truy vết hoặc chặn nạp lại.
• Bỏ dòng lỗi im lặng: Tổng sai nhưng không giải thích được.
• Audit success trước COMMIT: Trạng thái không phản ánh dữ liệu.
• Dashboard tự viết logic: KPI không nhất quán giữa báo cáo.

10. Đưa vào thực tế

Orchestrator quản lý retry, timeout và cảnh báo; SQL vẫn phải idempotent. Quyền ghi mart chỉ cấp cho service account pipeline, người dùng BI chỉ SELECT.

Theo dõi freshness, row count, quarantine rate, revenue variance và runtime. Đặt ngưỡng theo lịch sử và mùa vụ thay vì một con số cố định cho mọi ngày.

11. Bài tập thực hành

1. Viết migration đầy đủ cho ba schema.
2. Tạo procedure nhận source_date và batch_id.
3. Thêm refund_amount vào reconciliation.
4. Xây KPI tháng từ mart ngày.
5. Viết test chạy lại hai lần và so sánh checksum.

12. Project mở rộng

Đưa project lên dbt hoặc orchestration bằng Airflow, nhưng giữ nguyên grain, quality rule và reconciliation. Công cụ thay đổi cách vận hành, không thay thế thiết kế dữ liệu.

13. Tổng kết

Project chứng minh năng lực SQL khi có thể tái tạo, chạy lại, kiểm tra và giải thích. Một dashboard đẹp nhưng không có audit chưa phải hệ thống dữ liệu hoàn chỉnh.

Chuỗi 10 bài kết thúc ở đây: từ câu SELECT đầu tiên đến một data mart có kiểm soát. Bước tiếp theo phù hợp là Data Warehouse, dbt và orchestration.

14. Tài liệu tham khảo

• Transactions
• INSERT ON CONFLICT
• Materialized Views

TechData.AI - Leading the Future.
Tham khảo các khoá học theo link: https://techdata.ai/techdata-ai-course/
Hoàng Minh.