BigQuery 6 — Nạp dữ liệu, streaming & ELT

Ở bài Storage & tables chúng ta đã thấy BigQuery lưu dữ liệu ở định dạng cột, tách rời khỏi lớp tính toán. Nhưng trước khi truy vấn được thì dữ liệu phải vào BigQuery đã — hoặc BigQuery phải đọc được nó ở nơi nó đang nằm. Đó chính là chủ đề bài này: các con đường đưa dữ liệu vào kho, từ nạp khối (batch) đến streaming near-real-time, từ external table không nạp gì đến pipeline ELT hoàn chỉnh.

Câu hỏi thực tế đầu tiên không phải "làm sao nạp" mà là "có cần nạp không?". BigQuery cho bạn hai lựa chọn lớn: managed storage (nạp dữ liệu vào, BigQuery quản lý) và external/federated (để dữ liệu ở nguyên chỗ cũ, chỉ query). Sau khi chọn xong, mới đến chuyện chọn khối hay dòng (batch vs streaming). Cả bài xoay quanh hai trục quyết định này.

Nhắc lại quy ước của series: sandbox học tập chạy PostgreSQL, nên các đoạn code BigQuery bên dưới không chạy được ở đây — chúng chỉ mang tính minh hoạ cú pháp và ý tưởng.

Bức tranh tổng thể các con đường nạp

Sơ đồ trên là bản đồ của cả bài. Bên trái là các nguồn, ở giữa là BigQuery, bên phải là lớp biến đổi (dbt) và tiêu thụ (Looker). Ta đi lần lượt từng con đường.

Batch load — nạp khối, xương sống của mọi kho dữ liệu

Batch load là cách kinh điển và rẻ nhất: gom dữ liệu thành file, tạo một load job, BigQuery đọc file và ghi vào bảng managed. Đây nên là lựa chọn mặc định cho phần lớn khối lượng dữ liệu.

Định dạng hỗ trợ: theo tài liệu Google, batch load nhận đúng năm định dạng — CSV, JSON (newline-delimited), Avro, ORC, Parquet. Trong đó Parquet và Avro là hai định dạng đáng ưu tiên: chúng là định dạng nhị phân, tự mang schema, nén tốt, và (đặc biệt Parquet) là columnar nên load nhanh, ít lỗi kiểu dữ liệu hơn CSV/JSON rất nhiều. CSV tuy phổ biến nhưng dễ dính lỗi encoding, dấu phân cách, và không mang kiểu — hãy tránh cho khối lớn.

Nguồn: từ Google Cloud Storage (GCS) — đường phổ biến nhất, hoặc từ local file (tiện cho dữ liệu nhỏ, dev). GCS được ưu tiên vì có thể nạp song song nhiều file và không phụ thuộc máy của bạn.

Chi phí: đây là điểm hấp dẫn nhất — load job miễn phí (bạn không trả tiền cho băng thông hay công tính toán của job nạp khi dùng shared slot pool). Bạn chỉ trả tiền lưu trữ cho dữ liệu sau khi đã nằm trong bảng. So sánh: streaming có phí theo dung lượng. Vì vậy nguyên tắc chi phí số một là: nếu batch được thì đừng streaming.

Schema: hai cách. Auto-detect — BigQuery đoán kiểu cột từ mẫu dữ liệu (tiện cho khám phá nhanh, nhưng đoán sai kiểu là chuyện thường, ví dụ mã bưu chính bị đoán thành INT64 rồi mất số 0 đầu). Khai báo tường minh — bạn cung cấp schema; nên dùng cho production để tránh trôi kiểu. Với Parquet/Avro, schema đã nằm sẵn trong file nên gần như không cần lo.

Write disposition — quyết định ghi đè hay ghi thêm:

• WRITE_TRUNCATE: xoá sạch bảng rồi ghi mới (dùng cho snapshot/full refresh hằng ngày).
• WRITE_APPEND: ghi thêm vào cuối (dùng cho log/incremental).
• WRITE_EMPTY: chỉ ghi nếu bảng đang rỗng, nếu không thì báo lỗi (an toàn, tránh ghi nhầm).

Ví dụ nạp Parquet từ GCS bằng CLI:

# (minh hoạ — BigQuery)
# Nạp file Parquet từ GCS vào bảng, ghi đè toàn bộ (full refresh)
bq load \
  --source_format=PARQUET \
  --replace \
  analytics.orders_raw \
  gs://my-bucket/exports/orders/2026-06-30/*.parquet
# --replace tương đương WRITE_TRUNCATE.
# Parquet tự mang schema nên không cần --autodetect hay khai báo cột.

# Với CSV thì cần bật auto-detect (hoặc truyền schema tường minh):
bq load \
  --source_format=CSV --skip_leading_rows=1 --autodetect \
  --noreplace \
  analytics.events_raw \
  gs://my-bucket/events/*.csv
# --noreplace = WRITE_APPEND (ghi thêm).

Cùng logic đó, phiên bản SQL bằng câu lệnh LOAD DATA (chạy như một truy vấn, tiện đưa vào script/scheduled query):

-- (minh hoạ — BigQuery)
LOAD DATA INTO analytics.orders_raw
FROM FILES (
  format = 'PARQUET',
  uris   = ['gs://my-bucket/exports/orders/2026-06-30/*.parquet']
);
-- Mặc định là ghi thêm; dùng LOAD DATA OVERWRITE để thay WRITE_TRUNCATE.

External & federated — query mà không nạp

Đôi khi bạn không muốn dữ liệu vào managed storage: dữ liệu lake khổng lồ hiếm khi truy vấn, hay dữ liệu đang sống ở hệ khác và cần tươi tức thì. BigQuery cho phép query trực tiếp tại nguồn.

External table: bạn định nghĩa một bảng trỏ tới file trên GCS (Parquet/CSV/…). BigQuery không copy dữ liệu; mỗi lần query nó đọc file tại chỗ.

-- (minh hoạ — BigQuery)
CREATE EXTERNAL TABLE analytics.orders_ext
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://data-lake/orders/*.parquet']
);
-- Query như bảng thường, nhưng dữ liệu vẫn nằm trên GCS.
SELECT country, COUNT(*) FROM analytics.orders_ext GROUP BY country;

Federated query: query trực tiếp các nguồn không phải GCS — Cloud SQL (MySQL/Postgres), Google Sheets, Bigtable — bằng hàm EXTERNAL_QUERY, dữ liệu chảy về BigQuery đúng lúc truy vấn.

BigLake là thế hệ nâng cấp của external table: thêm quản trị (fine-grained access, cache metadata) cho dữ liệu trên GCS, xoá dần ranh giới giữa lake và warehouse.

Đánh đổi cần nhớ rõ: external cho bạn độ tươi tuyệt đối (query luôn thấy file mới nhất) và không tốn storage BigQuery, nhưng hiệu năng kém hơn managed table — không có tối ưu columnar nội bộ, không partition metadata tốt bằng, và với Cloud SQL còn giới hạn tải lên hệ nguồn. Quy tắc: external cho dữ liệu lake ít truy vấn hoặc cần tươi; managed cho dữ liệu truy vấn thường xuyên.

Streaming ingest — khi dữ liệu phải tươi từng giây

Batch có độ trễ theo chu kỳ (5 phút, 1 giờ, 1 ngày). Khi nghiệp vụ cần near-real-time — dashboard giao dịch trực tuyến, phát hiện gian lận, giám sát — bạn cần streaming: ghi từng dòng (hoặc lô nhỏ) và dữ liệu queryable gần như tức thì.

Storage Write API — lựa chọn khuyến nghị hiện nay

Storage Write API là dịch vụ nạp hợp nhất, hiệu năng cao dựa trên gRPC, gộp cả streaming và batch vào một API. Đây là con đường Google khuyến nghị cho mọi nhu cầu ghi dữ liệu mới. Điểm mạnh:

• Hiệu năng cao & chi phí thấp hơn so với API cũ (gRPC streaming thay vì HTTP REST).
• Exactly-once qua stream offset: nếu client cung cấp offset, API không bao giờ ghi hai message trùng offset trong cùng stream — nên retry an toàn, không sinh bản trùng.
• Miễn phí tới 2 TiB/tháng, sau đó tính theo dung lượng.
• Hỗ trợ nhiều loại stream cho các nhu cầu khác nhau:
  • Default stream: có sẵn, dữ liệu queryable ngay, đảm bảo at-least-once. Dùng cho luồng liên tục đơn giản.
  • Committed type: ghi xong thấy ngay, hỗ trợ exactly-once khi có offset — cho streaming độ trễ thấp cần tin cậy.
  • Pending type: dữ liệu buffer đến khi commit tường minh, mang tính giao dịch nguyên tử — là lựa chọn thay cho load job cho khối lớn.
  • Buffered type: commit theo dòng, dùng cho connector chuyên biệt (ví dụ Apache Beam).

# (minh hoạ — BigQuery) — Storage Write API, default stream, rút gọn
from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types, writer
from google.protobuf import descriptor_pb2

client = bigquery_storage_v1.BigQueryWriteClient()
parent = client.table_path("my-project", "analytics", "events")
stream = f"{parent}/streams/_default"          # default stream có sẵn

# Mỗi dòng được serialize bằng protobuf theo schema bảng (bỏ chi tiết).
proto_rows = types.ProtoRows()
proto_rows.serialized_rows.append(serialize_event({"user_id": 42, "ts": now}))

req = types.AppendRowsRequest()
req.write_stream = stream
req.proto_rows.rows.CopyFrom(proto_rows)
# (với committed/pending stream có thể gán req.offset để exactly-once)

append_stream = writer.AppendRowsStream(client, req)
append_stream.send(req)                         # dữ liệu queryable gần như tức thì

Legacy streaming inserts (tabledata.insertAll)

API cũ hơn: gọi REST tabledata.insertAll, mỗi request đẩy vài dòng JSON. Nó vẫn hoạt động và đơn giản để thử, nhưng đã bị Storage Write API thay thế: có phí theo dung lượng ($0.01/200 MB), giới hạn hạn ngạch chặt hơn, không exactly-once thuận tiện. Với dự án mới, hãy dùng Storage Write API; chỉ giữ insertAll nếu đang chạy hệ cũ.

Streaming buffer

Dù dùng API nào, dòng streaming mới vào trước tiên nằm trong streaming buffer — một vùng đệm — rồi mới được kết vào lưu trữ cột nền. Hệ quả thực tế cần nhớ: dữ liệu trong buffer không thể bị UPDATE/DELETE trong một khoảng thời gian ngắn sau khi ghi (thường tới ~30 phút). Nếu pipeline của bạn cần sửa dòng vừa stream, hãy tính đến độ trễ này.

Công cụ nạp có sẵn của hệ sinh thái Google

Không phải lúc nào cũng tự viết code nạp. Google có sẵn vài dịch vụ:

• BigQuery Data Transfer Service (DTS): nạp định kỳ, tự động từ các nguồn SaaS của Google (Google Ads, YouTube, Campaign Manager…), từ GCS, và từ warehouse khác. Bạn cấu hình lịch, DTS lo phần còn lại — không cần viết code.
• Dataflow (chạy Apache Beam): dịch vụ ETL/stream có quản lý. Dùng khi cần biến đổi dữ liệu trên đường vào (làm sạch, enrich, join stream) chứ không chỉ nạp thô. Beam viết một lần, chạy cả batch lẫn stream.
• Datastream: dịch vụ CDC (Change Data Capture) không-agent, bắt thay đổi từ Oracle/MySQL/PostgreSQL và đẩy sang BigQuery gần real-time — nối tiếp trực tiếp với series Oracle CDC.
• Pub/Sub → BigQuery subscription: đăng ký một topic Pub/Sub ghi thẳng vào bảng BigQuery, không cần Dataflow trung gian — con đường ngắn nhất cho event stream đơn giản.

ELT với dbt — biến đổi sau khi đã nạp

Kho dữ liệu hiện đại đảo trật tự truyền thống: không còn ETL (Transform trước, rồi Load), mà là ELT — Extract, Load dữ liệu thô vào BigQuery trước, rồi Transform bên trong BigQuery bằng SQL. Lý do: BigQuery đủ mạnh và rẻ để tự làm phần transform, nên đừng phí công biến đổi bên ngoài.

Công cụ transform số một cho lớp này là dbt: bạn viết các model bằng SQL SELECT, dbt lo việc CREATE TABLE/CREATE VIEW, quản lý phụ thuộc (DAG), test và tài liệu. Mẫu modern data stack kinh điển:

ingest (batch/stream) → BigQuery (raw) → dbt (staging → mart) → Looker (BI)

Đây chính là nửa phải của sơ đồ đầu bài. dbt biến các bảng thô lộn xộn thành các mart sạch, có kiểm thử, phục vụ trực tiếp cho báo cáo.

CDC vào BigQuery — áp thay đổi bằng MERGE

Với dữ liệu từ OLTP (Oracle/MySQL) chảy vào qua CDC — dù bằng Datastream hay Kafka Connect với Debezium — bạn thường nhận được một dòng thay đổi (insert/update/delete). Mẫu chuẩn để áp chúng vào bảng đích là dùng bảng staging chứa thay đổi rồi chạy MERGE (xem SQL features/MERGE):

-- (minh hoạ — BigQuery) — áp CDC vào bảng đích bằng MERGE
MERGE analytics.customers AS t
USING analytics.customers_cdc AS s          -- staging: bản ghi thay đổi mới nhất
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED AND s.op IN ('U','I') THEN UPDATE
  SET name = s.name, email = s.email, updated_at = s.ts
WHEN NOT MATCHED AND s.op <> 'D' THEN
  INSERT (customer_id, name, email, updated_at)
  VALUES (s.customer_id, s.name, s.email, s.ts);

Trong thực tế bạn stream các bản ghi CDC thô vào một bảng append-only, dùng window function lấy bản mới nhất mỗi khoá (xem SQL features), rồi MERGE định kỳ vào bảng "current state".

Use case thực tế: dựng pipeline ELT bán lẻ

Giả sử nhóm phân tích cần một mart doanh thu theo ngày/quốc gia, dữ liệu tươi tới sáng hôm sau là đủ.

1. Load (EL): hằng đêm, hệ nguồn export đơn hàng ra GCS dạng Parquet theo thư mục ngày (gs://.../orders/2026-06-30/). Một scheduled job chạy bq load --source_format=PARQUET --noreplace analytics.orders_raw ... để ghi thêm phần mới. Chọn Parquet + batch vì khối lớn, tươi hằng ngày là đủ — rẻ nhất và tin cậy nhất.
2. Sự kiện tươi: giỏ hàng và click stream cần tươi hơn → app đẩy qua Storage Write API (default stream) vào analytics.events, dashboard vận hành đọc gần real-time.
3. Transform (T): dbt chạy sau khi load xong: model stg_orders làm sạch orders_raw, model fct_daily_revenue group theo DATE(order_time) và country, ghi thành bảng partition theo ngày + cluster theo country (xem Partitioning & clustering) để rẻ khi truy vấn.
4. Serve: Looker trỏ vào fct_daily_revenue — nhà phân tích không bao giờ chạm bảng thô.
5. Dữ liệu lake ít dùng: log server 3 năm để trên GCS, khai báo external table, chỉ query khi điều tra sự cố — không tốn storage BigQuery.

Kết quả: hoá đơn thấp (batch miễn phí băng thông, mart nhỏ gọn), dữ liệu chính tươi hằng ngày, event tươi từng giây ở đúng nơi cần, và một điểm điều tra lake khi cần.

Ghi nhớ

• Có cần nạp không? Managed cho dữ liệu truy vấn thường xuyên; external/federated (GCS/Cloud SQL/Sheets/Bigtable, BigLake) cho lake ít dùng hoặc cần tươi — đổi hiệu năng lấy độ tươi & tiết kiệm storage.
• Batch load là mặc định: năm định dạng (CSV/JSON/Avro/ORC/Parquet), ưu tiên Parquet/Avro cho khối lớn; load job miễn phí băng thông, chỉ tính storage; nắm rõ WRITE_TRUNCATE / WRITE_APPEND / WRITE_EMPTY.
• Streaming khi cần tươi: dùng Storage Write API (hiệu năng cao, exactly-once qua offset, 2 TiB/tháng miễn phí, cả batch lẫn stream) thay cho legacy insertAll (cũ, có phí, hạn chế). Nhớ streaming buffer không cho UPDATE/DELETE ngay.
• Kiểm soát chi phí: nếu batch được thì đừng streaming; streaming có phí theo dung lượng.
• Công cụ: DTS (định kỳ từ SaaS/GCS), Dataflow/Beam (ETL/stream có biến đổi), Datastream (CDC), Pub/Sub → BQ (event đơn giản).
• ELT là chuẩn hiện đại: load thô → dbt transform → Looker; áp CDC (Datastream / Kafka Connect + Debezium) bằng MERGE.

Xem thêm: Tổng quan BigQuery · Storage & tables · SQL features & MERGE · Partitioning & clustering.