Spark 8 — Triển khai & vận hành production

Chạy một job Spark trên laptop bằng spark-submit local[*] thì dễ. Nhưng đưa job đó lên production — nơi nó phải xử lý hàng terabyte mỗi đêm, tự co giãn theo tải, retry khi lỗi, và không được ghi trùng dữ liệu khi chạy lại — mới là phần khó. Bài này khép lại series Spark chuyên sâu bằng câu chuyện vận hành: chọn cách chạy Spark ở đâu, cấu hình spark-submit cho đúng, đọc Spark UI để chẩn đoán sự cố, viết PySpark theo best practice, điều phối bằng Airflow, và xử lý những lỗi ai làm Spark cũng gặp.

Bài giả định bạn đã nắm phần lõi. Nếu chưa, đọc trước Shuffle & partitions, Tối ưu hiệu năng và Delta Lake & Lakehouse — ba bài đó là nền cho hầu hết quyết định vận hành ở đây. Bối cảnh xuyên suốt: một job ETL hằng ngày của nền tảng dữ liệu ngân hàng, đọc log giao dịch thô, làm sạch và ghi vào lakehouse.

Spark chạy ở đâu: các cluster manager

Spark tách biệt ứng dụng (driver + executors) khỏi nơi cấp tài nguyên (cluster manager). Cùng một job có thể chạy trên nhiều nền, chỉ đổi --master.

YARN — "lão làng" trong hệ sinh thái Hadoop

YARN là cluster manager mặc định của mọi cụm Hadoop. Ưu điểm: chín muồi, tích hợp sẵn HDFS, hỗ trợ hàng đợi (queue) đa người dùng và phân bổ tài nguyên công bằng. Ở chế độ cluster, driver chạy trong Application Master trên cụm; ở client, driver chạy tại máy gửi lệnh. Nhược: gắn với Hadoop, khó dùng khi hạ tầng đã chuyển sang container/K8s, và việc đóng gói dependency Python trên các node YARN khá phiền.

Kubernetes — hướng đi hiện đại

Từ Spark 3.1, chế độ chạy trên Kubernetes đã ổn định (GA). Spark coi K8s như một cluster manager: driver là một pod, mỗi executor cũng là một pod riêng. Có hai cách gửi:

• spark-submit trực tiếp: chỉ --master k8s://https://<api-server> là Spark tự tạo driver pod, driver lại tạo các executor pod.
• Spark Operator (Kubeflow): khai báo job bằng CRD SparkApplication (YAML), phù hợp GitOps và quản lý vòng đời tốt hơn.

K8s hỗ trợ dynamic allocation qua cơ chế shuffle tracking (không cần external shuffle service như YARN): executor tự thêm khi nhiều task chờ, tự bớt khi rảnh. Ưu điểm lớn nhất là đóng gói dependency bằng image — mọi thư viện Python/Java nằm gọn trong Docker image, hết cảnh "máy này thiếu thư viện". Nhược: cần đội vững K8s, và I/O shuffle trên storage tạm của pod cần chú ý.

Standalone và các nền managed

Standalone là cluster manager riêng của Spark, nhẹ, dễ dựng cho cụm nhỏ hoặc thử nghiệm, nhưng thiếu tính năng đa người dùng của YARN/K8s.

Các nền managed giúp bỏ qua phần lớn việc vận hành:

• Databricks: tối ưu sâu (Photon engine, Delta Lake gốc, Unity Catalog), autoscale tốt, notebook + job scheduler. Đổi lại chi phí và khóa nền tảng.
• AWS EMR / EMR on EKS: Spark trên hạ tầng AWS, tích hợp S3, spot instance rẻ; linh hoạt hơn Databricks nhưng phải tự lo nhiều hơn.
• AWS Glue: serverless, trả theo DPU, hợp job ETL nhẹ và ít phải quản cụm; kém linh hoạt cho job phức tạp.

Nguyên tắc chọn: đội nhỏ, muốn tập trung vào logic → managed. Đội platform vững, cần kiểm soát chi phí và hạ tầng → K8s hoặc EMR.

spark-submit và cấu hình

spark-submit là cửa ngõ đưa job vào cụm. Vài tham số cốt lõi:

• --master: nơi chạy (yarn, k8s://..., local[*], spark://host:7077).
• --deploy-mode: cluster (driver chạy trong cụm — chuẩn cho production) hay client (driver ở máy gửi — hợp debug/notebook).
• --executor-memory, --executor-cores: RAM và số core mỗi executor.
• --num-executors: số executor cố định (khi không bật dynamic allocation).
• --driver-memory: RAM cho driver (quan trọng khi có collect() hoặc broadcast).
• --conf: mọi cấu hình khác, ví dụ --conf spark.sql.shuffle.partitions=200.

Ví dụ gửi job PySpark lên K8s:

# (minh hoạ) — spark-submit job ETL lên Kubernetes
spark-submit \
  --master k8s://https://k8s-api.internal:6443 \
  --deploy-mode cluster \
  --name etl-transactions-daily \
  --conf spark.kubernetes.container.image=registry.internal/spark-etl:2026.06.30 \
  --conf spark.kubernetes.namespace=data-eng \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-sa \
  --conf spark.executor.instances=10 \
  --conf spark.executor.memory=8g \
  --conf spark.executor.cores=4 \
  --conf spark.driver.memory=4g \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.shuffleTracking.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=2 \
  --conf spark.dynamicAllocation.maxExecutors=20 \
  --conf spark.sql.shuffle.partitions=400 \
  local:///opt/app/jobs/etl_transactions.py \
  --run-date 2026-06-30

Đóng gói (packaging). Job Scala/Java đóng thành một jar (thường là "fat jar" đã gộp dependency bằng shade/assembly). Job PySpark có mấy lựa chọn:

• --py-files: đính kèm .zip/.egg/.py chứa module riêng của bạn.
• --files: đính kèm file cấu hình.
• --packages: kéo dependency từ Maven (ví dụ connector Delta, Kafka) — cẩn thận va chạm phiên bản.
• Với K8s: cách sạch nhất là bake mọi thứ vào image, dùng đường dẫn local:// trỏ tới file đã nằm sẵn trong image.

Dynamic allocation. Thay vì cố định --num-executors, bật spark.dynamicAllocation.enabled=true để Spark tự co giãn số executor theo số task đang chờ. Trên K8s nhớ bật shuffleTracking.enabled=true để Spark biết executor nào còn giữ dữ liệu shuffle mà chưa thu hồi. Đây là chìa khóa giảm chi phí: job dùng nhiều executor lúc cao điểm, nhả ra khi rảnh.

Kiến trúc chạy trên Kubernetes

Airflow phát lệnh spark-submit; K8s API server tạo driver pod; driver tự yêu cầu tạo các executor pod; toàn bộ đọc/ghi vào lakehouse trên object storage; log và metric đẩy ra hệ thống giám sát. Khi job kết thúc, executor pod bị dọn, driver pod chuyển Completed.

Spark UI và History Server — công cụ chẩn đoán

Khi job chậm hay lỗi, Spark UI là nơi đầu tiên nhìn vào. Driver phục vụ UI ở cổng 4040 lúc job đang chạy. Các tab quan trọng:

• Jobs: mỗi action (như write, count) tạo một job. Xem job nào chiếm nhiều thời gian.
• Stages: mỗi job chia thành stage theo ranh giới shuffle. Đây là tab quan trọng nhất. Nhìn Summary Metrics của task trong một stage: nếu max lớn hơn median nhiều lần → skew (một số task xử lý nhiều dữ liệu hơn hẳn). Cột Shuffle Read/Write cho biết lượng dữ liệu di chuyển; Spill (memory/disk) khác 0 nghĩa là dữ liệu tràn ra đĩa vì thiếu RAM.
• Tasks: chi tiết từng task. Task chạy lâu bất thường thường do skew hoặc GC. Cột GC Time cao báo hiệu áp lực bộ nhớ.
• SQL / DataFrame: xem physical plan thực tế. Kiểm tra Exchange (shuffle), BroadcastHashJoin vs SortMergeJoin, và xem AQE có gộp partition hay xử lý skew join không (nhìn nhãn AQEShuffleRead, coalesced).
• Storage: các RDD/DataFrame đã cache, dung lượng và tỷ lệ nằm trong RAM.
• Executors: bảng tổng hợp per-executor — số task đã chạy, thời gian GC, shuffle, dung lượng bộ nhớ. Một executor GC time chiếm phần lớn wall time là dấu hiệu cấu hình bộ nhớ chưa ổn.

Cách đọc để nối với tối ưu và shuffle: thấy shuffle read lớn → xem lại join/aggregation, cân nhắc broadcast join; thấy skew (max ≫ median) → salting hoặc bật AQE skew join; thấy spill → tăng RAM executor hoặc giảm dữ liệu mỗi partition; thấy GC cao → giảm executor.cores hoặc tăng RAM.

Spark History Server. UI cổng 4040 biến mất khi job kết thúc. Để xem lại job đã chạy, bật ghi event log (spark.eventLog.enabled=true, spark.eventLog.dir=s3://.../spark-events) và dựng History Server đọc thư mục đó. Không có nó, bạn gần như "mù" khi điều tra một job đêm qua bị chậm.

PySpark best practices

Nhiều sự cố production không đến từ Spark mà từ cách viết code. Vài quy tắc đắt giá:

• Ưu tiên DataFrame API và hàm dựng sẵn hơn UDF. Các hàm trong pyspark.sql.functions chạy trong JVM, được Catalyst tối ưu. UDF Python phải serialize dữ liệu qua lại giữa JVM và Python — chậm và tốn RAM. Nếu buộc phải dùng, ưu tiên pandas UDF (vectorized, dùng Arrow) thay vì UDF thường.
• Tránh collect() trên dữ liệu lớn. collect() kéo toàn bộ về driver — nguyên nhân OOM driver kinh điển. Cần xem thì show(20) hoặc take(n); cần lấy ra ngoài thì ghi xuống storage.
• Tránh sinh nhiều file nhỏ. Ghi ra hàng nghìn file bé làm chậm mọi lượt đọc sau và gây áp lực lên object storage. Trước khi ghi, coalesce()/repartition() về số file hợp lý; cân nhắc OPTIMIZE/compaction nếu dùng Delta (xem Delta Lakehouse).
• Đọc/ghi Parquet có partition. Dùng định dạng cột (Parquet/Delta) thay vì CSV/JSON cho dữ liệu lớn; partition theo cột lọc thường dùng (ví dụ event_date) để tận dụng partition pruning. Đừng partition theo cột có cardinality quá cao (như user_id) — sẽ tạo cả rừng thư mục nhỏ.
• Output idempotent. Job chạy lại (retry, backfill) không được nhân đôi dữ liệu. Cách phổ biến: ghi đè đúng partition của ngày đang xử lý bằng replaceWhere (Delta) hoặc partitionOverwriteMode=dynamic, hoặc MERGE theo khóa. Idempotency là tài sản quý nhất của một pipeline production.
• Tách logic để test được. Đưa phần biến đổi vào các hàm thuần nhận DataFrame trả DataFrame, tách khỏi phần đọc/ghi. Nhờ vậy có thể viết unit test với DataFrame nhỏ, không cần cả cụm.

# (minh hoạ) — biến đổi tách rời, dễ test và không có UDF thừa
from pyspark.sql import DataFrame, functions as F

def clean_transactions(df: DataFrame) -> DataFrame:
    return (
        df.filter(F.col("amount") > 0)
          .withColumn("event_date", F.to_date("event_ts"))
          .withColumn("channel", F.upper(F.col("channel")))
          .dropDuplicates(["txn_id"])
    )

# đọc/ghi tách khỏi logic; ghi đè đúng partition -> idempotent
def run(spark, run_date: str):
    raw = spark.read.parquet(f"s3://raw/transactions/dt={run_date}")
    out = clean_transactions(raw)
    (out.write
        .format("delta")
        .mode("overwrite")
        .option("replaceWhere", f"event_date = '{run_date}'")
        .partitionBy("event_date")
        .save("s3://lake/transactions"))

Điều phối bằng Airflow

Job Spark hiếm khi chạy đơn lẻ — nó nằm trong một pipeline có upstream/downstream, cần lịch, retry và backfill. Airflow là công cụ điều phối phổ biến nhất (xem Airflow production ops). Hai operator hay dùng:

• SparkSubmitOperator: gói lệnh spark-submit, lấy cấu hình cụm từ một Connection. Hợp khi Airflow chạy cạnh cụm YARN/Standalone.
• KubernetesPodOperator: chạy spark-submit trong một pod trên K8s. Cách này được ưa chuộng khi cả Airflow lẫn Spark đều ở trên K8s — mỗi run là một pod độc lập, cách ly dependency, dễ dọn.

# (minh hoạ) — Airflow DAG gọi Spark job trên K8s mỗi ngày
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
import pendulum

with DAG(
    dag_id="etl_transactions_daily",
    schedule="0 2 * * *",
    start_date=pendulum.datetime(2026, 6, 1, tz="Asia/Ho_Chi_Minh"),
    catchup=True,               # cho phép backfill khoảng ngày còn thiếu
    default_args={"retries": 2, "retry_delay": pendulum.duration(minutes=10)},
    max_active_runs=1,          # tránh hai run cùng ghi một partition
) as dag:

    spark_etl = KubernetesPodOperator(
        task_id="spark_etl",
        name="spark-etl-{{ ds_nodash }}",
        namespace="data-eng",
        image="registry.internal/spark-etl:2026.06.30",
        cmds=["/opt/spark/bin/spark-submit"],
        arguments=[
            "--master", "k8s://https://k8s-api.internal:6443",
            "--deploy-mode", "cluster",
            "--conf", "spark.executor.instances=10",
            "local:///opt/app/jobs/etl_transactions.py",
            "--run-date", "{{ ds }}",   # ngày logic của run -> idempotent theo ngày
        ],
    )

Điểm quan trọng: truyền ngày logic ({{ ds }}) vào job thay vì today(), để backfill và retry luôn xử lý đúng partition. Đặt retries để chịu lỗi tạm thời, max_active_runs=1 để hai run không ghi đè lẫn nhau. Kết hợp với output idempotent ở trên, một task fail có thể retry an toàn.

Chất lượng và độ tin cậy

• Checkpoint / idempotency: với Structured Streaming, checkpoint đảm bảo exactly-once khi khởi động lại; với batch, dựa vào ghi đè partition hoặc MERGE như đã nói.
• Kiểm tra dữ liệu: sau khi ghi, chạy các assertion cơ bản (số dòng > 0, không null ở khóa, khoảng giá trị hợp lý). Có thể thành một task riêng trong DAG, fail sớm còn hơn phát tán dữ liệu bẩn xuống dưới.
• Logging & metrics: gom log driver/executor về một nơi (S3/ELK); bật event log cho History Server; theo dõi thời gian chạy job, lượng shuffle, số dòng in/out để phát hiện lệch bất thường.
• Chi phí: dùng spot/preemptible cho executor (chịu được mất node nhờ retry), giữ driver trên node ổn định; bật autoscale/dynamic allocation để không trả tiền cho executor rảnh; dọn file nhỏ để giảm chi phí đọc.

Lỗi thường gặp và cách xử lý

• OOM executor (Container killed... exceeded memory limits): dữ liệu mỗi partition quá lớn hoặc skew. Chữa: tăng executor.memory, tăng số partition (spark.sql.shuffle.partitions), xử lý skew, hoặc giảm executor.cores (ít task song song hơn trên cùng RAM).
• OOM driver: gần như luôn do collect()/toPandas() dữ liệu lớn, hoặc broadcast một bảng quá to. Chữa: bỏ collect, hạ spark.sql.autoBroadcastJoinThreshold, tăng driver.memory nếu thực sự cần.
• Skew: vài task chạy mãi không xong trong khi phần còn lại đã xong (nhìn max ≫ median ở tab Stages). Chữa: bật AQE skew join (spark.sql.adaptive.skewJoin.enabled), salting khóa, hoặc tách riêng key nóng. Xem shuffle.
• Small files: quá nhiều file bé làm chậm đọc. Chữa: coalesce/repartition trước khi ghi, OPTIMIZE/compaction định kỳ với Delta.
• Task chậm (straggler): một task lê thê. Nguyên nhân thường là skew, node yếu, hoặc GC. Bật speculation (spark.speculation) để chạy bản sao task chậm, và xem GC time ở tab Executors.
• Dependency conflict: va chạm phiên bản thư viện (jar hoặc Python) giữa máy gửi và cụm. Chữa triệt để bằng bake vào image (K8s) hoặc quản chặt --packages/--py-files; tránh phụ thuộc vào môi trường có sẵn trên node.

Use case thực tế

Đưa job ETL hằng ngày lên K8s, điều phối bằng Airflow. Đội data ngân hàng cần làm sạch log giao dịch mỗi đêm. Trước đây job chạy client mode trên một máy edge, hay OOM và không rollback được. Đội chuyển sang: đóng gói toàn bộ (code + PySpark + connector Delta) vào một Docker image tag theo commit; job chạy cluster mode trên K8s với dynamic allocation (2–20 executor, executor dùng spot, driver trên node ổn định). Airflow trên cùng cụm dùng KubernetesPodOperator phát spark-submit, lịch 2 giờ sáng, retries=2, max_active_runs=1, truyền {{ ds }} làm run-date. Job ghi Delta bằng replaceWhere theo event_date nên retry/backfill không nhân đôi dữ liệu. Sau ghi có một task kiểm tra dữ liệu (số dòng, null khóa) — fail thì cả run fail và cảnh báo bắn đi. Event log đẩy lên S3 cho History Server để điều tra job chậm. Kết quả: chi phí giảm nhờ spot + autoscale, rollback bằng deploy image cũ, và một đêm lỗi mạng chỉ khiến task retry rồi tự khỏi thay vì gọi người dậy lúc 3 giờ sáng.

Ghi nhớ

• Nơi chạy: YARN (hệ Hadoop chín muồi), Kubernetes (hiện đại, đóng gói bằng image, dynamic allocation qua shuffle tracking), Standalone (nhỏ gọn); Databricks/EMR/Glue khi muốn managed. Đội nhỏ → managed; đội vững hạ tầng → K8s/EMR.
• spark-submit: nắm --master, --deploy-mode (cluster cho prod), --executor-memory/-cores, --num-executors, --conf; đóng gói jar hoặc --py-files, tốt nhất là bake vào image trên K8s; ưu tiên dynamic allocation thay vì số executor cố định.
• Spark UI: đọc Stages (skew qua max ≫ median, shuffle, spill), SQL tab (physical plan, AQE), Executors (GC). Bật event log + History Server để điều tra job đã chạy.
• PySpark: DataFrame/hàm dựng sẵn hơn UDF, không collect() lớn, tránh file nhỏ, Parquet/Delta có partition hợp lý, output idempotent, tách logic để test.
• Airflow điều phối: SparkSubmitOperator hoặc KubernetesPodOperator; truyền ngày logic, đặt retries, max_active_runs=1, dùng catchup cho backfill.
• Tin cậy: idempotency/checkpoint, kiểm tra dữ liệu sau ghi, gom log/metrics, tiết kiệm chi phí bằng spot + autoscale.
• Lỗi quen mặt: OOM driver (bỏ collect/broadcast to), OOM executor (RAM/partition/skew), skew (AQE/salting), small files (compaction), straggler (speculation/GC), dependency conflict (bake image).

Đọc lại Tối ưu hiệu năng và Shuffle & partitions khi chẩn đoán từ Spark UI, và Delta Lake & Lakehouse cho phần ghi idempotent — đó là bộ ba đi kèm mọi quyết định vận hành trong bài này.