Nguồn NotebookLM: Data Engineering nâng cao: Spark & Lakehouse

Spark là gì? Xử lý dữ liệu bằng nhiều task

Nguồn gốc: Spark là gì? Xử lý dữ liệu bằng nhiều task. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Một file doanh thu vài nghìn dòng có thể xử lý bằng SQL hoặc Python trên laptop. Nhưng khi dữ liệu nằm trong hàng nghìn file và một máy không đủ thời gian hay bộ nhớ, bạn cần chia công việc. Apache Spark là một engine xử lý dữ liệu có thể thực thi các phần việc song song trên nhiều máy.

Spark không phải nơi lưu dữ liệu lâu dài. Nó đọc từ hệ thống lưu trữ, biến đổi dữ liệu rồi ghi kết quả ra ngoài. Chạy Spark cũng không tự biến một phép JOIN sai thành đúng: bạn vẫn phải biết mỗi dòng đại diện cho điều gì.

Một bài toán nhỏ

Quán cà phê có ba đơn đã thanh toán: cửa hàng A nhận 40.000 và 35.000 đồng, cửa hàng B nhận 50.000 đồng. Ta muốn tổng tiền theo cửa hàng. Spark có thể đọc các phần dữ liệu độc lập, tính tổng cục bộ, chuyển các tổng cùng cửa hàng về một nơi rồi cộng lần cuối.

Với ba dòng, chi phí khởi động Spark lớn hơn lợi ích song song. Ví dụ nhỏ giúp nhìn rõ cơ chế; nó không chứng minh Spark nhanh hơn database.

Môi trường thực hành

Bộ bài này dùng Spark/PySpark 3.5.6, Python 3.12, JDK 17, chạy local[2] trên macOS Intel. Đây là phiên bản cố định để tái hiện ví dụ, không phải tuyên bố về phiên bản mới nhất. Linux và Windows/WSL chưa được kiểm thử trong đợt này. local[2] dùng hai luồng thực thi trên một máy, không dựng cluster hai máy.

Trước khi chạy, cài Python 3.12 và JDK 17 từ nhà phân phối phù hợp hệ điều hành. JAVA_HOME phải trỏ tới thư mục JDK chứa bin/java; trên macOS, nếu JDK đã đăng ký với hệ thống, có thể dùng lệnh dưới đây. Nếu giải nén JDK riêng, đặt đường dẫn thực tế thay thế.

java -version
python3.12 --version
export JAVA_HOME=$(/usr/libexec/java_home -v 17)
python3.12 -m venv .venv-spark
source .venv-spark/bin/activate
python -m pip install pyspark==3.5.6
export PYSPARK_PYTHON="$VIRTUAL_ENV/bin/python"
export SPARK_LOCAL_IP=127.0.0.1

Các bài tiếp theo cho code Python hoàn chỉnh. Lưu vào file rồi chạy python ten_file.py trong môi trường vừa tạo. Chỉ cài thêm pandas/Arrow nếu bài thực sự dùng các API đó.

Ví dụ đầu tiên

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").appName("cafe-v1").getOrCreate()
orders = spark.createDataFrame(
    [(101, "A", 40000), (102, "A", 35000), (103, "B", 50000)],
    "order_id long, shop string, amount_dong long",
)
result = orders.groupBy("shop").agg(F.sum("amount_dong").alias("total_dong"))
print([(r.shop, r.total_dong) for r in result.orderBy("shop").collect()])
spark.stop()

Kết quả: [('A', 75000), ('B', 50000)]. Mỗi dòng nguồn là một đơn, order_id duy nhất; tiền là số nguyên đồng. Chỉ dùng collect() vì kết quả có hai dòng, không áp dụng cách này cho bảng lớn.

Khi nào dùng, khi nào cân nhắc công cụ khác?

Spark phù hợp với xử lý file lớn, nhiều bước biến đổi và hệ thống đã có tài nguyên phân tán. Nếu dữ liệu vừa một máy, DuckDB hoặc database thường đơn giản hơn. Nếu cần trả lời từng request trong vài mili giây, Spark batch không phải lựa chọn mặc định.

Lỗi JAVA_GATEWAY_EXITED thường cần kiểm tra JDK và JAVA_HOME trước. Nếu Python worker không khởi động, kiểm tra driver và worker dùng cùng phiên bản Python. Không tăng RAM ngẫu nhiên khi thông báo lỗi đang chỉ ra đường dẫn sai.

Đọc tiếp kiến trúc Spark để biết driver giao việc cho ai.

Tài liệu gốc

Hướng dẫn cài PySpark 3.5.6 quy định phụ thuộc Python/Java và cách cài bằng pip. Spark SQL mô tả DataFrame và engine dùng chung cho SQL/API. Đối chiếu ngày 1/10/2026.

DataFrame trong Spark: schema, lọc và tổng hợp

Nguồn gốc: DataFrame trong Spark: schema, lọc và tổng hợp. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Một DataFrame có các cột có tên và kiểu dữ liệu, nhưng không phải toàn bộ bảng đã nằm trong RAM laptop. Nó biểu diễn dữ liệu và kế hoạch xử lý; Spark thực thi khi có action. Nếu quen SQL, bạn có thể dùng DataFrame API và SQL cho cùng bài toán.

Schema là thỏa thuận đầu vào

Mỗi dòng sau là một đơn hàng; order_id là khóa nghiệp vụ. Spark không tự áp đặt uniqueness hay foreign key cho DataFrame. amount_dong là số nguyên đồng, NULL nghĩa là chưa biết tiền, không phải đơn có giá trị bằng 0.

Dùng môi trường trong bài tổng quan.

from tempfile import TemporaryDirectory
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").appName("schema-v1").getOrCreate()
orders = spark.createDataFrame(
    [(101, "A", 40000, "paid"), (102, "A", 35000, "paid"),
     (103, "B", 50000, "paid"), (104, "B", None, "pending")],
    "order_id long, shop string, amount_dong long, status string",
)
paid = orders.filter(F.col("status") == "paid")
summary = paid.groupBy("shop").agg(F.sum("amount_dong").alias("total_dong"))
print([(r.shop, r.total_dong) for r in summary.orderBy("shop").collect()])
print(orders.filter(F.col("amount_dong").isNull()).count())
with TemporaryDirectory() as tmp:
    summary.write.mode("overwrite").parquet(tmp + "/summary")
    print(spark.read.parquet(tmp + "/summary").count())
spark.stop()

Ba kết quả lần lượt là [('A', 75000), ('B', 50000)], 1, 2. Thư mục tạm chỉ chứa output của ví dụ; thao tác overwrite không chạm dữ liệu khác. Parquet output là một thư mục chứa các part file, không đảm bảo một file có tên do bạn chọn.

Những lỗi dễ lọt qua

amount_dong == None không thay thế isNull(). So sánh SQL với NULL không cho kết quả true; filter có thể loại dòng mà bạn tưởng sẽ giữ lại. sum bỏ qua NULL, vì vậy tổng có vẻ hợp lý vẫn có thể thiếu đơn. Theo dõi số dòng thiếu riêng và định nghĩa chính sách trước khi điền bằng 0.

Khi đọc CSV, suy luận schema trên mẫu có thể không phù hợp dữ liệu ngày sau. Chọn schema rõ ràng, kiểm tra các dòng parse lỗi và quy định cách xử lý. DataFrame không có thứ tự mặc định: dùng orderBy nếu kết quả cần thứ tự để kiểm tra; sort toàn bộ bảng có chi phí.

SQL và API

Bạn có thể tạo temporary view bằng orders.createOrReplaceTempView("orders") rồi truy vấn với spark.sql(...). View này thuộc session, không tự tạo bảng bền vững trong một database. Dùng API nào dễ đọc hơn, nhưng giữ cùng schema, logic NULL và kiểm tra output.

Ưu tiên hàm dựng sẵn như sum, when, to_timestamp trước Python UDF. Hàm dựng sẵn cho engine nhiều thông tin hơn để tối ưu; UDF còn có chi phí chuyển dữ liệu và thực thi Python.

Đọc tiếp Catalyst để hiểu vì sao hai cách viết có thể dẫn đến kế hoạch tương tự.

Tài liệu gốc

Spark SQL getting started 3.5.6, NULL semantics và Parquet là các tài liệu cho schema/API, giá trị thiếu và đọc ghi file. Đối chiếu ngày 1/10/2026.

Structured Streaming: micro-batch, checkpoint và output

Nguồn gốc: Structured Streaming: micro-batch, checkpoint và output. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Mỗi vài giây có file đơn mới. Bạn muốn cập nhật kết quả mà không đọc lại tất cả file đã xử lý. Structured Streaming dùng các phép biến đổi DataFrame trên dữ liệu đến dần; chế độ micro-batch xử lý các lô nhỏ và ghi nhận tiến độ.

Bài này dùng file JSON đến dần, không kết nối Kafka. Ví dụ chỉ kiểm tra tiến độ nguồn và file sink sau restart, không mô phỏng lỗi cluster giữa lúc commit.

Một input, một checkpoint, một output

Dùng môi trường trong bài tổng quan. Mỗi dòng là một đơn, ID duy nhất. Chưa có phép tổng hợp theo thời gian nên không cần watermark. Ba thư mục được tạo trong vùng tạm và xóa sau khi kết thúc.

import json
from pathlib import Path
from tempfile import TemporaryDirectory
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[2]").appName("stream-v1").getOrCreate()
with TemporaryDirectory() as tmp:
    base = Path(tmp)
    source = base / "input"
    source.mkdir()
    output = str(base / "output")
    checkpoint = str(base / "checkpoint")
    def add_file(name, row):
        (source / name).write_text(json.dumps(row) + "\n", encoding="utf-8")
    def run_once():
        stream = spark.readStream.schema("order_id long, amount_dong long").json(str(source))
        query = (stream.writeStream.format("parquet").outputMode("append")
                 .option("path", output).option("checkpointLocation", checkpoint)
                 .trigger(availableNow=True).start())
        query.awaitTermination()
    add_file("batch1.json", {"order_id": 101, "amount_dong": 40000})
    run_once()
    add_file("batch2.json", {"order_id": 102, "amount_dong": 35000})
    run_once()
    run_once()  # Không thêm file: không được ghi lại hai đơn cũ.
    result = spark.read.parquet(output).orderBy("order_id")
    print([(r.order_id, r.amount_dong) for r in result.collect()])
spark.stop()

Kết quả: [(101, 40000), (102, 35000)]. availableNow xử lý input sẵn có rồi kết thúc, tiện kiểm tra hữu hạn; muốn chạy liên tục cần chọn trigger phù hợp. File nguồn thực tế phải được đưa vào hoàn chỉnh, thường bằng rename nguyên tử trên filesystem hỗ trợ; không để query đọc file đang ghi dở.

Checkpoint không phải tùy chọn trang trí

Checkpoint chứa tiến độ và, với truy vấn stateful, trạng thái xử lý. Mỗi query có checkpoint riêng trên storage bền vững mà các lần chạy sau truy cập được. Ví dụ dùng vùng tạm để học nên không bảo vệ dữ liệu qua lần chạy script mới.

Nếu xóa checkpoint, query có thể đọc lại nguồn. Nếu đổi schema hoặc phép xử lý stateful, không mặc định checkpoint cũ tương thích. Lập kế hoạch migration và replay với kiểm tra output; không xóa checkpoint production chỉ để hết lỗi.

Event time, watermark và bảo đảm đầu cuối

Event time là thời điểm sự kiện xảy ra, khác lúc Spark nhận được. Khi tổng hợp cửa sổ, bạn cần định nghĩa timestamp/múi giờ và chính sách dữ liệu đến muộn. Watermark giúp giới hạn state; nó không sửa timestamp sai và không có nghĩa mọi sự kiện muộn đều sẽ được giữ.

Khả năng tránh output trùng còn tùy source có replay được và sink có commit phù hợp. foreachBatch mặc định có thể gọi lại cùng batch; cần dùng batch ID hoặc khóa nghiệp vụ để ghi idempotent. Checkpoint không làm một HTTP API bất kỳ trở thành exactly-once. Khi ID đơn xuất hiện trong hai file khác nhau, ví dụ trên vẫn ghi hai dòng: theo dõi file không phải deduplicate nghiệp vụ.

Nếu query đứng lâu, xem lastProgress, input rate, processing rate, thời gian batch và lỗi sink. Kiểm tra storage/checkpoint trước khi tăng executor.

Tài liệu gốc

Structured Streaming guide 3.5.6 mô tả file source, triggers, checkpoint, output modes và bảo đảm theo từng sink. Đối chiếu ngày 1/10/2026.

Hệ thống phân tán là gì?

Nguồn gốc: Hệ thống phân tán là gì?. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Một pipeline đọc nhiều file, chia việc cho vài worker rồi gom kết quả. Nhìn từ bên ngoài đó là một hệ thống; bên trong có nhiều tiến trình trao đổi qua mạng và có thể lỗi độc lập. Đây là điểm xuất phát của hệ thống phân tán.

Không nhất thiết phải thuê cloud. Bạn có thể chạy nhiều tiến trình hoặc container trên máy mình để học. Nhưng chúng cùng phụ thuộc một máy, nên không mô phỏng đầy đủ việc một máy hỏng hay một vùng mất điện.

Ba worker và một file bị đếm hai lần

Giả sử có ba file với tổng tiền lần lượt 40000, 60000 và 50000 đồng. Bộ điều phối giao mỗi file cho một worker. Tổng đúng là 150000.

Worker B tính xong 60000 và gửi kết quả, nhưng phản hồi nhận kết quả bị mất. Bộ điều phối giao lại file B. Nếu cộng cả hai lần trả kết quả, tổng thành 210000. Retry giúp hoàn thành công việc nhưng có thể tạo trùng nếu kết quả không gắn với mã nhiệm vụ.

Một cách thiết kế là lưu một kết quả cho mỗi task_id, chỉ cập nhật trạng thái theo quy tắc rõ ràng. Khi đủ ba mã nhiệm vụ, mới tổng hợp. Đây là ví dụ bằng tay, không phải cam kết một thuật toán tổng quát đã chịu mọi lỗi.

Timeout không cho biết điều gì đã xảy ra

Khi không nhận phản hồi, tiến trình kia có thể đã chết, đang chậm, hoặc đã làm xong nhưng mạng mất phản hồi. Mình chỉ biết không nhận được câu trả lời trong khoảng chờ. Vì thế các quyết định retry, chọn leader và commit phải tính đến sự không chắc chắn này.

Thời gian trên các máy cũng không tự đồng bộ hoàn hảo. Không nên lấy timestamp từ hai máy rồi khẳng định sự kiện nhỏ hơn luôn xảy ra trước về mặt nhân quả. Lamport: Time, Clocks, and the Ordering of Events phân biệt thứ tự nhân quả với đồng hồ vật lý.

Được thêm tài nguyên, phải phối hợp nhiều hơn

Chia việc giúp sử dụng nhiều CPU hoặc storage. Đổi lại, truyền dữ liệu, phối hợp trạng thái, xử lý lỗi và quan sát hệ thống đều khó hơn. Với dữ liệu nhỏ, một tiến trình thường dễ vận hành hơn và có thể nhanh hơn do không tốn phối hợp.

Thử ca worker chết sau khi ghi kết quả nhưng trước khi báo xong: bạn sẽ xác định file đó đã được xử lý dựa trên điều gì? Câu trả lời nên dựa trên trạng thái có thể kiểm tra, không chỉ log “đã bắt đầu”.

Đọc tiếp partitioning, replication và idempotency.

Phân vùng dữ liệu

Nguồn gốc: Phân vùng dữ liệu. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Nếu mọi đơn hàng nằm trên một máy, dung lượng và khả năng xử lý của máy đó có giới hạn. Partitioning chia dữ liệu thành các phần để lưu hoặc xử lý riêng. Phần được đặt trên các node khác nhau thường được gọi là shard.

Partition dữ liệu khác với network partition — tình trạng các nhóm máy không liên lạc được. Đừng nhầm hai ý nghĩa chỉ vì cùng một từ.

Thử chia sáu khách

Một quy tắc học đơn giản là customer_id % 3: khách 1 và 4 vào phần 1, khách 2 và 5 vào phần 2, khách 3 và 6 vào phần 0. Muốn lấy tất cả đơn của khách 4, mình biết phần cần tìm. Muốn cộng doanh thu toàn bộ, phải đọc cả ba phần rồi gom.

Bạn có thể tự kiểm tra bằng Python 3, không cần thư viện ngoài:

parts = {0: [], 1: [], 2: []}
for customer_id in range(1, 7):
    parts[customer_id % 3].append(customer_id)
print(parts)

Kết quả: {0: [3, 6], 1: [1, 4], 2: [2, 5]}.

Khóa đều chưa chắc tải đều

Sáu khách chia đều, nhưng nếu khách 4 có một triệu đơn còn người khác chỉ có vài đơn, phần 1 vẫn quá tải. Chọn khóa cần nhìn phân bố dữ liệu và truy vấn thật. Chia theo ngày tiện cho lọc thời gian, nhưng ngày hôm nay có thể nhận hầu hết các lượt ghi.

Chia theo khoảng giữ các khóa gần nhau cùng phần, giúp truy vấn khoảng. Chia bằng hash thường phân tán khóa hơn, nhưng truy vấn khoảng có thể phải tìm nhiều phần. Một phép JOIN giữa hai bộ dữ liệu được chia theo khóa khác nhau có thể cần chuyển rất nhiều dòng qua mạng.

Thêm phần không chỉ sửa một con số

Với quy tắc modulo, chuyển từ 3 sang 4 làm nhiều khóa đổi nơi. Nếu reader dùng quy tắc mới trước khi dữ liệu đã chuyển, nó có thể tìm sai chỗ. Hệ thống thật cần cơ chế định tuyến, di chuyển và kiểm soát trong giai đoạn chuyển tiếp. Consistent hashing là một hướng để giảm lượng di chuyển, không loại bỏ mọi việc vận hành.

Bài tập: tính phần của khách 4 khi đổi từ 3 sang 4. Kết quả đổi từ 1 sang 0. Dữ liệu cũ cần được tìm ở đâu trong lúc chưa chuyển xong?

Partitioning không tạo bản sao để chịu mất dữ liệu; đó là vai trò khác của replication. Tham khảo thiết kế partition trong Dynamo paper.

Lưu raw data ở quy mô lớn với Data Lake

Nguồn gốc: Lưu raw data ở quy mô lớn với Data Lake. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Công ty muốn giữ log ứng dụng, JSON từ đối tác, ảnh biên nhận và lịch sử order nhiều năm. Đưa mọi thứ vào bảng warehouse có thể đắt hoặc không phù hợp.

Data Lake (hồ dữ liệu) lưu data ở quy mô lớn, thường trên object storage, với nhiều định dạng và mức xử lý khác nhau.

    flowchart TD
 A["Database, API, logs, files"] --> B["Data Lake<br/>Object storage"]
 B --> C["Raw"]
 B --> D["Cleaned"]
 B --> E["Curated"]
  

Một layout đơn giản:

/orders/raw/ingest_date=2026-09-08/part-001.json
/orders/clean/order_date=2026-09-08/part-001.parquet

Data Lake khác Data Warehouse ở đâu?

Ranh giới ngày càng mờ vì warehouse đọc object storage và lake có table format. Điều quan trọng là khả năng hệ thống thật, không phải nhãn marketing.

Hồ data có thể thành nơi khó dùng

Nếu file không có owner, schema, partition convention hoặc retention, người dùng không biết data nào tin được. Small files làm query chậm; quyền bucket quá rộng làm lộ data; xoá nhầm path có thể phá nhiều bảng.

Data Lake phù hợp khi cần giữ raw data lớn, nhiều định dạng, chia sẻ cho nhiều engine hoặc tách compute khỏi storage. Một team chỉ có các bảng phân tích có cấu trúc có thể bắt đầu bằng Data Warehouse đơn giản hơn.

Khi Data Lake cần cách quản lý như Warehouse

Nguồn gốc: Khi Data Lake cần cách quản lý như Warehouse. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Data Lake lưu Parquet rẻ và linh hoạt, nhưng nhiều job cùng sửa file dễ làm người đọc thấy trạng thái dở dang. Team cũng cần Update, Delete, schema evolution và lịch sử snapshot.

Lakehouse là cách xây khả năng quản lý table kiểu Data Warehouse trên storage của Data Lake, thường bằng object storage, table format và một hay nhiều compute engine.

    flowchart TD
 A["Object Storage<br/>Parquet files"] --> B["Table Format<br/>Metadata và snapshots"]
 B --> C["Compute Engine"]
 C --> D["SQL, BI, ML"]
  

Table không chỉ là một thư mục file

Table format theo dõi file nào thuộc snapshot hiện tại. Writer tạo file mới rồi commit metadata; reader thấy snapshot trước hoặc sau commit, không thấy nửa chừng.

Lakehouse không phải một sản phẩm duy nhất. Iceberg, Delta Lake và Hudi là table format với thiết kế khác nhau; Spark, Flink hoặc query engine là compute.

Khi nào cần?

Lakehouse hữu ích khi data lớn nằm trên object storage, nhiều workload cần dùng chung table và team cần transaction hoặc snapshot. Nếu warehouse hiện tại đáp ứng chi phí, performance và governance, chuyển kiến trúc chỉ để có tên mới hiếm khi tạo giá trị.

Sự linh hoạt cũng mang thêm việc vận hành: compact small files, dọn snapshot cũ, chọn partition, kiểm soát catalog và compatibility giữa engine. Hãy đánh giá những trách nhiệm này cùng lợi ích storage mở.

File Format: chọn giữa CSV, JSON và Parquet

Nguồn gốc: File Format: chọn giữa CSV, JSON và Parquet. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Cùng một triệu order có thể được lưu bằng CSV, JSON hoặc Parquet. Nội dung business giống nhau, nhưng kích thước và cách engine đọc rất khác.

File Format (định dạng file) quy định data được mã hoá và tổ chức trong một file.

Row và column

Report chỉ cần restaurant_id và amount trong bảng 50 field. CSV thường phải đọc cả row. Parquet tổ chức theo column nên engine có thể bỏ qua các column không dùng.

    flowchart TD
 A["Query cần 2/50 columns"] --> B{"Format"}
 B -->|CSV hoặc JSON| C["Đọc gần như toàn file"]
 B -->|Parquet| D["Đọc column cần thiết"]
  

Parquet còn giữ type và statistics theo nhóm row, giúp bỏ qua phần không phù hợp filter. Đây là lý do nó thường dùng cho analytical storage.

Format không sửa data quality

CSV có thể chứa dấu phẩy trong text, encoding khác nhau hoặc ngày không thống nhất. JSON linh hoạt nên hai row có thể mang shape khác nhau. Parquet có schema nhưng nhiều file trong cùng dataset vẫn có thể khác schema.

CSV phù hợp để trao đổi nhỏ và tương thích rộng. JSON phù hợp event/API có cấu trúc lồng. Parquet phù hợp scan phân tích lớn. Có thể nhận JSON ở ingestion rồi chuyển sang Parquet sau khi làm sạch; không cần ép một format cho mọi giai đoạn.

Table Format: chọn giữa Iceberg, Delta Lake và Hudi

Nguồn gốc: Table Format: chọn giữa Iceberg, Delta Lake và Hudi. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Parquet biết cách lưu column trong một file, nhưng không biết 8.000 file nào tạo thành version hiện tại của bảng orders. Nó cũng không tự điều phối hai writer đang commit cùng lúc.

Table Format (định dạng quản lý table) đặt một lớp metadata trên các data file để theo dõi snapshot, schema, partition và thay đổi của table.

    flowchart TD
 A["Catalog<br/>Tìm table"] --> B["Table metadata<br/>Snapshot hiện tại"]
 B --> C["Manifest<br/>Danh sách data files"]
 C --> D["Parquet files<br/>Data thật"]
  

File format và table format làm hai việc

Iceberg, Delta Lake và Hudi đều giải quyết nhóm vấn đề này nhưng khác protocol, catalog integration, engine support và cách tối ưu update.

Cách chọn

Đừng chọn từ bảng feature chung chung. Hãy kiểm tra:

1. Compute engine đang dùng hỗ trợ read/write đến mức nào?
2. Catalog và cloud environment tích hợp ra sao?
3. Workload chủ yếu append, update từ CDC hay streaming upsert?
4. Team có vận hành compaction và cleanup được không?
5. Tool khác có đọc đúng version và feature đang bật không?

Table format đem transaction, schema evolution và time travel đến object storage, nhưng không tự tạo Data Quality, governance hay data model tốt. Một thư mục Parquet append-only nhỏ có thể chưa cần nó. Khi nhiều writer, update/delete và lịch sử snapshot trở thành yêu cầu thật, lớp metadata này bắt đầu có giá trị.

Schema Evolution mà không làm hỏng downstream

Nguồn gốc: Schema Evolution mà không làm hỏng downstream. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Application cần thêm delivery_method vào order. Một tháng sau, team muốn đổi total_amount từ số nguyên sang object gồm amount và currency. Hai thay đổi có mức rủi ro khác nhau.

Schema Evolution (quá trình schema thay đổi theo thời gian) là cách thêm, bỏ hoặc sửa cấu trúc data trong khi quản lý compatibility với consumer.

Thay đổi theo nhiều bước

Để đổi amount thành amount_cents, team có thể:

    flowchart TD
 A["1. Thêm amount_cents"] --> B["2. Ghi cả hai field"]
 B --> C["3. Chuyển consumer"]
 C --> D["4. Theo dõi field cũ"]
 D --> E["5. Xoá amount"]
  

Trong giai đoạn chuyển tiếp:

Data Contract ghi thời hạn và owner. Lineage tìm consumer. Quality check xác nhận hai field tương đương.

Data lịch sử vẫn tồn tại

Schema mới không tự sửa file Parquet cũ. Reader có thể phải hiểu cả version cũ và mới, hoặc team Backfill lịch sử sang schema mới. Cần test với data từ nhiều thời điểm, không chỉ sample mới nhất.

Schema registry và table format có thể kiểm tra một số compatibility rule, nhưng không biết thay đổi ý nghĩa business. Đổi status=done thành status=delivered vẫn cùng type string và vẫn có thể phá metric.

Schema sẽ thay đổi vì sản phẩm thay đổi. Mục tiêu là cho producer tiến lên mà consumer có thời gian phát hiện, chuyển đổi và xác nhận kết quả.

Idempotency: chạy lại pipeline mà không tạo duplicate

Nguồn gốc: Idempotency: chạy lại pipeline mà không tạo duplicate. Nội dung dưới đây là bản sao bài Wiki tại lúc website được build.

Pipeline đã ghi xong 900 order thì mất kết nối. Bạn không biết 100 order cuối đã được ghi hay chưa, nên chạy lại cả lô. Nếu job dùng INSERT, bảng có thể xuất hiện duplicate.

Idempotency (khả năng chạy lại mà không tạo thêm kết quả ngoài ý muốn) nghĩa là xử lý cùng input một hay nhiều lần vẫn đưa hệ thống đến cùng trạng thái.

    flowchart TD
 A["Input<br/>ORD-1042"] --> B["Chạy lần 1"] --> C["Một row ORD-1042"]
 A --> D["Chạy lại"] --> E["Vẫn một row ORD-1042"]
  

Dùng business key để nhận diện row

order_id cho biết hai lần nhận được có phải cùng một order hay không.

Thay vì luôn thêm row, pipeline có thể Upsert: thêm khi key chưa tồn tại, cập nhật khi key đã có.

INSERT INTO orders (order_id, status, amount)
VALUES ('ORD-1042', 'paid', 120000)
ON CONFLICT (order_id) DO UPDATE
SET status = EXCLUDED.status,
    amount = EXCLUDED.amount;

Với bảng tổng hợp theo ngày, một cách khác là tính lại toàn bộ partition rồi thay thế nguyên khối. Xoá rồi ghi lại cần transaction hoặc vùng tạm; nếu job dừng sau bước xoá, người dùng sẽ thấy data trống.

Không phải lúc nào key cũng rõ

Một payment retry có thể tạo nhiều transaction hợp lệ cho cùng order. Dùng order_id làm key sẽ ghi đè nhầm. Key phải phản ánh grain: nếu mỗi row là một lần thanh toán, payment_id mới phù hợp.

Idempotency cần được test bằng cách cố tình chạy cùng lô hai lần và so cả số row lẫn metric. Đây là nền tảng để Retry và Backfill an toàn.