Snowflake 3 — Nạp dữ liệu: Stage, COPY, Snowpipe

Bức tranh tổng thể việc nạp dữ liệu

Trong Snowflake, mọi dữ liệu đi vào bảng đều phải đi qua một stage — một vị trí lưu file trung gian. Dữ liệu ở dạng file (CSV, JSON, Parquet...) được đặt vào stage, sau đó lệnh COPY INTO đọc file, phân tích (parse) theo một file format và ghi vào bảng. Đây là mô hình nền tảng: file → stage → COPY → table.

Có hai con đường chính đưa dữ liệu vào:

• Batch load bằng COPY INTO: bạn (hoặc một job Airflow/dbt) chủ động gọi lệnh nạp một lô file. Thích hợp cho tải theo lịch (hằng giờ, hằng ngày), backfill lịch sử, reload.
• Continuous load bằng Snowpipe: file vừa xuất hiện trong stage là được nạp gần như tức thì (near-real-time), thường kích hoạt tự động qua thông báo sự kiện (event notification) của cloud storage.

Bài này đi từ stage → file format → COPY và xử lý lỗi → PUT/GET → Snowpipe → so sánh và best practice. Về warehouse (compute) dùng cho COPY, xem Snowflake 2 — Virtual Warehouse. Tổng quan kiến trúc ở Snowflake 1 — Tổng quan.

Các loại stage

Stage là con trỏ tới nơi chứa file. Snowflake phân biệt internal stage (file nằm trong vùng lưu trữ do Snowflake quản lý) và external stage (file nằm ở S3/GCS/Azure Blob của bạn, Snowflake chỉ tham chiếu).

User stage và table stage tồn tại mặc định, không cần tạo. Named stage phải khai báo và là lựa chọn tốt nhất cho pipeline sản xuất vì tách bạch, phân quyền được, gắn sẵn file format.

Ví dụ minh hoạ (cú pháp Snowflake, không phải SQL sandbox):

-- Named internal stage, gắn sẵn file format
CREATE STAGE stg_core_txn
  FILE_FORMAT = (TYPE = PARQUET);

-- External stage trỏ vào S3 data lake, dùng storage integration
CREATE STAGE stg_lake_txn
  URL = 's3://ncb-datalake/core/transactions/'
  STORAGE_INTEGRATION = int_s3_datalake
  FILE_FORMAT = (TYPE = PARQUET);

Storage integration là đối tượng bảo mật ánh xạ quyền truy cập cloud (IAM role của AWS, service account của GCP, SAS/managed identity của Azure) — cách khuyến nghị thay vì nhúng credential trực tiếp vào stage. Chi tiết phân quyền xem Snowflake 7 — Security & Governance.

FILE FORMAT — mô tả cấu trúc file

FILE FORMAT cho Snowflake biết cách đọc file: định dạng, ký tự phân tách, cách xử lý header, nén, giá trị null... Tách nó thành đối tượng riêng để tái sử dụng và chuẩn hoá.

-- CSV có header, phân tách bằng dấu phẩy, nén gzip
CREATE FILE FORMAT ff_csv
  TYPE = CSV
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  NULL_IF = ('', 'NULL')
  COMPRESSION = GZIP;

-- JSON dạng NDJSON (mỗi dòng một object), strip mảng ngoài
CREATE FILE FORMAT ff_json
  TYPE = JSON
  STRIP_OUTER_ARRAY = TRUE;

-- Parquet
CREATE FILE FORMAT ff_parquet
  TYPE = PARQUET;

Ba định dạng thường gặp trong ngân hàng:

• CSV: xuất từ hệ thống cũ, extract của core banking. Cần cẩn thận với dấu phân tách trùng trong nội dung, encoding (nên UTF-8), và FIELD_OPTIONALLY_ENCLOSED_BY.
• JSON: log giao dịch, message từ Kafka, API payload. Snowflake nạp cả JSON vào cột kiểu VARIANT rồi truy vấn semi-structured — xem Snowflake 4 — SQL & dữ liệu bán cấu trúc.
• Parquet: định dạng cột nén, chuẩn cho data lake. Nhanh nhất để nạp vào Snowflake vì đã có schema và nén sẵn; nên ưu tiên khi bạn kiểm soát được đầu ra.

COPY INTO — batch load

COPY INTO <table> là chủ lực của batch load. Nó đọc file từ stage, parse theo file format, và ghi vào bảng. Snowflake ghi nhớ (metadata) các file đã nạp trong ~64 ngày nên mặc định không nạp lại file trùng — tránh nhân đôi dữ liệu.

-- Nạp toàn bộ file Parquet trong stage vào bảng
COPY INTO core_transactions
  FROM @stg_lake_txn
  FILE_FORMAT = (FORMAT_NAME = ff_parquet)
  ON_ERROR = 'ABORT_STATEMENT';

-- Nạp có ánh xạ cột và biến đổi ngay trong lệnh
COPY INTO core_transactions (txn_id, account_no, amount, ccy, event_ts)
  FROM (
    SELECT $1:id::STRING,
           $1:acct::STRING,
           $1:amt::NUMBER(18,2),
           $1:ccy::STRING,
           TO_TIMESTAMP($1:ts::NUMBER)
    FROM @stg_lake_txn
  )
  FILE_FORMAT = (FORMAT_NAME = ff_parquet);

Điểm quan trọng: COPY cho phép transform nhẹ ngay khi nạp (cast kiểu, chọn cột, hàm đơn giản, SEQ, biểu thức) nhưng không hỗ trợ JOIN hay aggregate. Biến đổi phức tạp để lại cho lớp transform (dbt) — xem dbt — Tổng quan.

ON_ERROR và xử lý lỗi

ON_ERROR quyết định hành vi khi một dòng lỗi (sai kiểu, thiếu cột, parse fail):

Trong ngân hàng, dữ liệu tài chính nên nghiêng về ABORT_STATEMENT hoặc SKIP_FILE_5% — không âm thầm bỏ dòng giao dịch. Nếu buộc dùng CONTINUE, phải giám sát cột lỗi.

VALIDATION_MODE — kiểm thử trước khi nạp

Trước khi nạp thật, chạy khô để phát hiện lỗi mà không ghi dữ liệu:

-- Kiểm tra file, KHÔNG nạp: liệt kê tối đa 100 lỗi
COPY INTO core_transactions
  FROM @stg_lake_txn
  FILE_FORMAT = (FORMAT_NAME = ff_parquet)
  VALIDATION_MODE = 'RETURN_100_ERRORS';

Sau khi COPY thật, dùng VALIDATE(table, job_id => '_last') để lấy chi tiết dòng lỗi của lần nạp gần nhất. Kết hợp với view COPY_HISTORY trong INFORMATION_SCHEMA/ACCOUNT_USAGE để giám sát pipeline.

PUT / GET với internal stage

Với internal stage, file phải được đẩy từ máy client lên stage trước khi COPY. Đó là việc của PUT (upload) và GET (download). Hai lệnh này chỉ chạy qua SnowSQL CLI hoặc driver (không chạy trong Snowsight worksheet).

-- Đẩy file local lên table stage của bảng core_transactions
PUT file:///data/extract/txn_2026_06_30_*.csv @%core_transactions
  AUTO_COMPRESS = TRUE
  PARALLEL = 8;

COPY INTO core_transactions
  FROM @%core_transactions
  FILE_FORMAT = (FORMAT_NAME = ff_csv);

-- Tải file kết quả từ stage về máy
GET @%core_transactions file:///data/download/;

PUT tự nén file (gzip) và mã hoá trước khi truyền. PARALLEL chia file lớn thành nhiều luồng upload. Với external stage (S3/GCS/Azure), bạn không dùng PUT — file đã nằm sẵn ở data lake, chỉ cần trỏ external stage và COPY.

Snowpipe — nạp liên tục, gần real-time

Snowpipe tự động chạy COPY INTO khi file mới xuất hiện trong stage, dựa trên event notification của cloud storage (S3 → SNS/SQS, GCS → Pub/Sub, Azure → Event Grid). Không cần warehouse riêng — Snowpipe dùng compute serverless do Snowflake quản lý, tính phí theo lượng dữ liệu nạp cộng phụ phí quản lý (per-file overhead).

-- Pipe auto-ingest: file mới vào stage → tự động COPY
CREATE PIPE pipe_core_txn
  AUTO_INGEST = TRUE
AS
COPY INTO core_transactions
  FROM @stg_lake_txn
  FILE_FORMAT = (FORMAT_NAME = ff_parquet);

Sau khi tạo pipe, lấy notification_channel (SQS ARN với AWS) bằng SHOW PIPES và cấu hình event notification của bucket trỏ vào đó. Từ đó, mỗi lần một file được ghi vào prefix s3://ncb-datalake/core/transactions/, Snowflake nhận thông báo và nạp trong vòng vài chục giây tới vài phút.

Sơ đồ luồng nạp

Snowpipe Streaming — nạp theo dòng, độ trễ thấp nhất

Snowpipe kinh điển vẫn dựa trên file: dữ liệu phải được gom thành file rồi đặt vào stage. Snowpipe Streaming là cơ chế khác — nạp trực tiếp từng dòng (row-level) vào bảng qua API/SDK, không cần tạo file trung gian, độ trễ thấp hơn hẳn và chi phí per-row rẻ hơn với luồng liên tục. Đây là lựa chọn khi nguồn là stream thuần (ví dụ đọc thẳng từ Kafka bằng Snowflake Connector for Kafka ở chế độ streaming).

So sánh: COPY vs Snowpipe vs Snowpipe Streaming

Nguyên tắc chọn: cần backfill lớn hoặc tải theo lịch rõ ràng → COPY. File tới lai rai và cần độ trễ vài phút → Snowpipe. Cần near-real-time thật sự từ stream → Snowpipe Streaming.

Best practice kích cỡ file & phân vùng

• Kích cỡ file lý tưởng: 100–250 MB đã nén. File quá nhỏ (vài KB–MB) làm tăng per-file overhead của Snowpipe và giảm hiệu quả song song của COPY; file quá lớn (>vài GB) khó chia luồng và làm chậm phục hồi khi lỗi.
• Gom file nhỏ trước khi nạp. Nếu Kafka Connect S3 sink sinh ra hàng nghìn file bé, cấu hình flush theo kích thước/thời gian để ra file gần ngưỡng 100–250 MB.
• Nén dữ liệu (gzip cho CSV/JSON; Parquet đã nén sẵn). Giảm chi phí lưu trữ và truyền.
• Phân vùng thư mục theo thời gian: .../transactions/dt=2026-06-30/hour=14/part-0001.parquet. Cho phép COPY nạp có chọn lọc bằng path prefix và loại bỏ file cũ dễ dàng.
• Ưu tiên Parquet khi kiểm soát được nguồn — schema sẵn, nén cột, parse nhanh.
• Tách bảng raw và bảng đã chuẩn hoá: nạp thô vào bảng raw (giữ nguyên, có VARIANT nếu cần), rồi để dbt/SQL làm phần biến đổi. Xem thêm mô hình lakehouse ở Lakehouse.

Nạp từ data lake và liên hệ Kafka/CDC

Với ngân hàng, đa số dữ liệu vào Snowflake không phải upload thủ công mà chảy từ hệ thống lõi qua một chuỗi: CDC → Kafka → data lake → Snowflake.

• CDC (Change Data Capture) bắt thay đổi từ core banking (Oracle) bằng Debezium hoặc GoldenGate, đẩy vào Kafka. Xem Kafka — Kiến trúc.
• Kafka Connect (S3/GCS sink) ghi message thành file Parquet/JSON vào data lake, phân vùng theo thời gian.
• Snowflake nạp từ external stage trỏ vào lake, bằng COPY theo lịch hoặc Snowpipe auto-ingest.
• Nếu cần độ trễ rất thấp, dùng Snowflake Connector for Kafka chế độ Snowpipe Streaming, đọc thẳng topic và nạp row-level — bỏ qua bước file trung gian.

Điểm cần nhớ về CDC: dữ liệu tới dưới dạng chuỗi thay đổi (insert/update/delete). Nạp thô toàn bộ event vào bảng raw, rồi dựng bảng "hiện trạng" (current state) ở lớp transform bằng logic dedupe theo khoá + timestamp mới nhất (thường qua dbt incremental hoặc MERGE). COPY/Snowpipe chỉ lo phần đưa event vào, không lo phần hoà giải trạng thái.

Use case thực tế

Bối cảnh: NCB cần đưa nhật ký giao dịch thẻ từ core banking vào Snowflake để đội rủi ro và AML phân tích. Khối lượng ~8 triệu giao dịch/ngày, yêu cầu độ trễ dưới 5 phút cho cảnh báo gian lận.

Kiến trúc và cấu hình:

1. CDC: Debezium bắt thay đổi bảng CARD_TXN trên Oracle, đẩy vào topic Kafka core.card_txn.
2. Sink to lake: Kafka Connect S3 sink ghi Parquet vào s3://ncb-datalake/core/card_txn/dt=YYYY-MM-DD/hour=HH/, flush mỗi khi đạt ~150 MB hoặc mỗi 60 giây — cân bằng giữa độ trễ và kích cỡ file.
3. External stage + Snowpipe: tạo stg_card_txn trỏ prefix trên với storage integration; tạo pipe AUTO_INGEST = TRUE; cấu hình S3 event notification (SQS) trỏ vào notification channel của pipe.
4. Kết quả: file rơi vào lake được Snowpipe nạp vào bảng raw raw.card_txn trong ~40–90 giây. Độ trễ tổng end-to-end (Oracle commit → Snowflake) dưới 3 phút, đạt SLA.
5. Xử lý lỗi: pipe dùng ON_ERROR = SKIP_FILE_5%; job giám sát đọc COPY_HISTORY mỗi 5 phút, cảnh báo nếu có file bị skip hoặc pipe tồn đọng (SYSTEM$PIPE_STATUS).
6. Transform: dbt incremental dedupe theo (txn_id, op_ts) để dựng bảng mart.card_txn_current, phục vụ mô hình phát hiện gian lận.

Chi phí: dùng Snowpipe serverless thay vì warehouse chạy 24/7 giúp giảm đáng kể compute — chỉ trả theo lượng dữ liệu nạp thực tế. Backfill 90 ngày lịch sử ban đầu thì dùng COPY INTO với một warehouse size L chạy trong vài giờ rồi tắt, thay vì để Snowpipe xử lý per-file cho hàng trăm nghìn file cũ.

Ghi nhớ

• Mọi dữ liệu vào Snowflake đi qua stage: file → stage → COPY INTO → table. External stage trỏ S3/GCS/Azure, internal stage do Snowflake quản lý (cần PUT để upload).
• Bốn loại stage: user (@~), table (@%bang), named internal, external. Sản xuất nên dùng named/external + storage integration để phân quyền và tái dùng file format.
• FILE FORMAT mô tả cách parse (CSV/JSON/Parquet). Ưu tiên Parquet khi kiểm soát nguồn.
• COPY INTO là batch load, cho transform nhẹ (cast/chọn cột), không JOIN/aggregate. ON_ERROR chọn ABORT_STATEMENT/SKIP_FILE_n% cho dữ liệu tài chính; dùng VALIDATION_MODE để chạy khô.
• Snowflake nhớ metadata file ~64 ngày → không nạp lại file trùng mặc định.
• Snowpipe = auto-ingest theo event, serverless, độ trễ vài chục giây–phút, hợp file nhỏ giọt. Snowpipe Streaming = nạp row-level qua SDK, độ trễ giây, hợp stream/CDC.
• Kích cỡ file lý tưởng 100–250 MB đã nén; gom file nhỏ, nén, phân vùng thư mục theo thời gian.
• Đường dữ liệu core banking điển hình: CDC → Kafka → data lake → Snowflake (COPY/Snowpipe); phần hoà giải trạng thái CDC để cho lớp transform (dbt) lo.