การประมวลผลข้อมูลขนาดใหญ่แบบกระจายด้วย Apache Spark
5.1 การประมวลผลระดับ Big Data กระจายเครื่องด้วย Apache Spark
เมื่อปริมาณข้อมูลมีขนาดใหญ่เกินกว่าจะประมวลผลในเซิร์ฟเวอร์เครื่องเดียวได้ Apache Spark จะทำหน้าที่กระจายภาระงานไปยังโหนดประมวลผลย่อยในระบบคลัสเตอร์
- Step 1: Data Ingestion & Event Processing
- Step 2: Distributed Computation & State Management
- Step 3: Orchestration & Resource Allocation
1. สถาปัตยกรรม Driver, Cluster Manager, และ Executor (Shared-Nothing JVM)
Learning Progression
- [BASIC] ปูพื้นฐานภาษาเข้าใจง่าย - เข้าใจคอนเซปต์ภาพรวมและการแก้ปัญหาเบื้องต้น
- [INTERMEDIATE] โค้ด/คอนฟิกไวยากรณ์จริง - การเขียนโค้ดเพื่อใช้งานจริงในระบบ
- [PROFESSIONAL] Under-the-hood & Performance/FinOps - กลไกเบื้องลึกและการรีดประสิทธิภาพ
Real-World Enterprise Scenario
เคสระบบการเงิน/Big Tech: การรองรับ Transaction จำนวนมหาศาลต่อวินาทีพร้อมประกัน Data Integrity สูงสุด โดยใช้สถาปัตยกรรมที่ยืดหยุ่นและการมอนิเตอร์ระดับสูง
ทฤษฎีและกลไกการทำงาน (How it works): Apache Spark ทำงานแบบกระจายศูนย์โดยใช้สถาปัตยกรรมแบบ **Shared-Nothing**: - **Driver Program**: ทำหน้าที่แปลรหัสโค้ดของผู้พัฒนาสร้างเป็นแผนผังขั้นตอนคิวรี (Logical/Physical Plans) และแบ่งงานออกเป็น Task ย่อย - **Cluster Manager (เช่น YARN, Kubernetes, Standalone)**: บริหารจัดการและจัดสรรเครื่องคอมพิวเตอร์ในเครือข่าย - **Executors**: โพรเซส Java Virtual Machine (JVM) ย่อยที่รันอยู่บน Workers คอยดึงคำสั่ง Task มาประมวลผลในแรมของตนเองโดยไม่แชร์หน่วยความจำร่วมกันข้ามเครื่อง
# Spark Session configurations in PySpark pointing to a cluster manager
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("PipeCraftBigDataJob") \
.master("spark://spark-master:7077") \
.config("spark.executor.instances", "4") \
.config("spark.executor.cores", "2") \
.getOrCreate()
Use Case ในชีวิตจริง (Real-world Scenario): บริษัทยักษ์ใหญ่ด้านโทรคมนาคมคำนวณสถิติยอดใช้เน็ตเวิร์กของซิมมือถือระดับพันล้านรายการต่อคืน ไดรเวอร์ของ Spark จะหั่นแบ่งภาระงานแยกออกเป็น 2,000 ส่วนและแจกจ่ายให้ Executor 50 เครื่องรันคู่ขนานกันจนสำเร็จใน 10 นาที
ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ปัญหา **Driver Out of Memory (OOM)** เมื่อวิศวกรเรียกคำสั่งรวบรวมข้อมูลดิบขนาดใหญ่เข้าสู่เครื่องแม่ด้วยคำสั่ง `.collect()` แก้ไขโดยห้ามใช้คำสั่งดึงข้อมูลยักษ์เข้า Driver และให้ใช้วิธีเซฟผลลัพธ์ผ่านคำสั่งเขียนไฟล์แบบกระจายโดยตรง `.write.parquet()`
2. รายละเอียดการจัดการแรมหน่วยความจำภายใน Spark Executor JVM (Spark Memory Pools)
ทฤษฎีและกลไกการทำงาน (How it works): แรมหน่วยความจำ Heap Memory ของ JVM บนโหนด Executor ถูกแบ่งเป็น 2 ส่วนหลักแบบ Unified Memory: - **Execution Memory**: ใช้คำนวณการจัดกลุ่มคีย์ การเรียงข้อมูล และการทำ Shuffling หากหน่วยความจำในส่วนนี้มีเนื้อที่ไม่พอ Spark จะยอมเขียนหน่วยความจำส่วนเกินนี้บันทึกลงสู่ฮาร์ดดิสก์แทน (Spill to Disk) - **Storage Memory**: เก็บรักษากลุ่มตารางข้อมูลแคช (`.cache()`) และตัวแปรกลุ่ม Broadcast หากว่าง Execution memory สามารถยืมไปใช้ได้และสามารถเตะข้อมูลแคชออกได้
# Spark configuration settings defining memory allocations
spark.conf.set("spark.memory.fraction", "0.6") # Storage + Execution memory fraction of JVM heap
spark.conf.set("spark.memory.storageFraction", "0.5") # Dedicated storage within memory fraction
Use Case ในชีวิตจริง (Real-world Scenario): ในจ๊อบแปลงประวัติธนาคารที่ต้องการเข้าถึงตารางซ้ำๆ การทำแคชตารางเงินฝากบน Storage Memory ช่วยข้ามความหน่วงเวลาอ่านไฟล์ Parquet ซ้ำๆ ทำให้ขั้นตอนคำนวณขั้นถัดไปรันได้เร็วขึ้นเป็น 5 เท่า
ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ข้อผิดพลาด **Executor OOM (Out of Memory)** เนื่องจากแรมล้นพังถล่ม แก้ไขโดยวิเคราะห์โครงสร้างพาร์ทิชันผ่าน UI และปรับแต่งแรมจัดสรรเพิ่มโควต้า `spark.executor.memory` หรือแก้ไขขอบเขตพาร์ทิชันให้เล็กลง
3. Catalyst Optimizer & Tungsten Engine
ทฤษฎีและกลไกการทำงาน (How it works): - **Catalyst Optimizer**: รับ Query จากผู้ใช้มาสร้าง Logical Plan และ Optimize (เช่น ดัน Filter ไปทำก่อน, เลือกประเภท Join ที่ดีที่สุด) จนได้ Physical Plan ที่มีประสิทธิภาพสูงสุด - **Tungsten Engine**: จัดการเรื่อง Memory Management โดยตรงระดับไบนารี หลีกเลี่ยง Java Garbage Collection Overhead และใช้ Code Generation เพื่อลด CPU cycles
# Explain plan output showing optimized physical plan generated by Catalyst
df.filter(df.age > 25).join(df2, "id").explain("cost")
Use Case ในชีวิตจริง (Real-world Scenario): หากเราเขียนโค้ด Filter ข้อมูลหลังจาก Join ตารางกัน Catalyst จะฉลาดพอที่จะผลัก (Push-down) การ Filter ไปทำก่อน Join เพื่อลดปริมาณข้อมูลที่ต้องประมวลผล ทำให้ Query เร็วขึ้น
ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ผู้พัฒนาอาจพยายามเขียนโค้ดสลับซับซ้อนด้วย UDF (User Defined Functions) แบบดั้งเดิมใน Python ซึ่ง Catalyst ไม่สามารถมองเห็นข้างในเพื่อ Optimize ได้ ควรเปลี่ยนมาใช้ Spark SQL Built-in Functions เสมอ
4. Spark Adaptive Query Execution (AQE)
ทฤษฎีและกลไกการทำงาน (How it works): AQE เป็นฟีเจอร์ที่ปรับเปลี่ยนแผนการรันงาน (Query Plan) แบบไดนามิกขณะรันไทม์ (Dynamic Planning) โดยพิจารณาจากสถิติของข้อมูลจริงที่รันไปแล้วบางส่วน ฟีเจอร์หลักคือ 1) แปลง Sort-Merge Join เป็น Broadcast Join อัตโนมัติ 2) จัดสรรขนาดของ Shuffle Partitions อัตโนมัติ (Coalescing) 3) จัดการข้อมูลกระจุกตัว (Skew Join Optimization) อัตโนมัติ
# Enabling Adaptive Query Execution in PySpark (Enabled by default in Spark 3.2+)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
Use Case ในชีวิตจริง (Real-world Scenario): เมื่อรัน ETL ดึงข้อมูลข้ามตารางที่มี Data Skew หนักๆ AQE จะตรวจพบพาร์ทิชันที่ใหญ่ผิดปกติและทำการแตกย่อยพาร์ทิชันนั้นออกให้สมดุลโดยอัตโนมัติ ทำให้งานที่เคยค้างที่ Executor สุดท้ายสามารถเสร็จได้เร็วขึ้น
ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หากปิด AQE ในเวอร์ชันเก่า จะต้องมานั่งจูน `spark.sql.shuffle.partitions` เองและเสี่ยงเจอปัญหา Data Skew ดังนั้นควรอัปเกรดและเปิดใช้งาน AQE เสมอ
5. Shuffle Mechanics (Sort-Based vs Push-Based)
ทฤษฎีและกลไกการทำงาน (How it works): - **Sort-Based Shuffle**: Map tasks เขียนข้อมูลลง Local disk ของตัวเอง จากนั้น Reduce tasks จะดึง (Pull) ข้อมูลข้ามเน็ตเวิร์กไป - **Push-Based Shuffle (ใหม่)**: Map tasks จะทำการ "ผลัก" (Push) บล็อกของข้อมูลไปรวมกันบน Shuffle services ปลายทาง ทำให้ Reduce tasks สามารถอ่านข้อมูลที่เป็นก้อนใหญ่ๆ เรียงตามลำดับได้รวดเร็วขึ้น ลดการกระจายตัวของ Disk I/O
# Enabling push-based shuffle in Spark on YARN
spark.conf.set("spark.shuffle.push.enabled", "true")
Use Case ในชีวิตจริง (Real-world Scenario): ในคลัสเตอร์ที่มีแบนด์วิดท์จำกัดหรือมีดิสก์แบบจานหมุน (HDD) การใช้ Push-Based Shuffle ในงานที่มี Shuffle ขนาดระดับเพตะไบต์ จะช่วยลดเวลา Disk Seek Time และเพิ่มความเสถียรของจ๊อบอย่างมีนัยสำคัญ
ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หากใช้ Sort-Based แล้ว Map task เยอะมาก จะทำให้เกิดปัญหา "Too Many Open Files" และ Network Timeout ให้เพิ่มค่า Timeout หรือพิจารณาอัปเกรดมาใช้ Push-Based Shuffle หากคลัสเตอร์รองรับ
Weekend Sandbox Challenge: Build Broadcast Hash Join in PySpark
โจทย์ปฏิบัติการ: จงเขียนสคริปต์ PySpark เพื่อทำการเชื่อมตารางธุรกรรมขนาดใหญ่ (Transactions) เข้ากับตาราง Lookup ขนาดเล็ก โดยสังเกตการทำงานของ Adaptive Query Execution (AQE) และทดลองรันคำสั่ง Explain Plan
# local_broadcast_join.py
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").appName("AQEDemo") \
.config("spark.sql.adaptive.enabled", "true").getOrCreate()
# Create mock dataframes
large_df = spark.createDataFrame([(1, "prod_1", 100.0), (2, "prod_2", 250.0)], ["id", "product_id", "price"])
small_df = spark.createDataFrame([("prod_1", "Electronics"), ("prod_2", "Home")], ["product_id", "category"])
# Join and trigger action to let AQE optimize dynamically
result = large_df.join(small_df, "product_id")
result.explain("cost")
result.show()
Senior Technical Interview Q&A
Q1: Spark Adaptive Query Execution (AQE) ช่วยแก้ปัญหาอะไรใน Data Pipeline ได้บ้าง?
A1: AQE ช่วยแก้ปัญหาสามส่วนหลัก: 1) ลดปัญหา Data Skew ระหว่างทำ Join อัตโนมัติ โดยแบ่งพาร์ทิชันที่ใหญ่เกินไปออกเป็นส่วนเล็กๆ 2) รวม Shuffle Partitions อัตโนมัติ ป้องกันปัญหา Small Files 3) เปลี่ยนกลยุทธ์การ Join เป็น Broadcast Join ไดนามิกเมื่อพบว่าตารางขนาดเล็กลงมากหลังการ Filter
Q2: Catalyst Optimizer แตกต่างจาก Tungsten Engine อย่างไรใน Apache Spark?
A2: Catalyst Optimizer ทำหน้าที่เป็นสมองในการวางแผน (Logical to Physical Plan) เพื่อลดปริมาณการคำนวณ เช่น Push-down Predicates ส่วน Tungsten Engine ทำหน้าที่บริหารจัดการหน่วยความจำดิบ (Off-heap memory management) และแปลงคำสั่งเป็นระดับไบนารี (Whole-Stage Code Generation) เพื่อรีดประสิทธิภาพฮาร์ดแวร์สูงสุดในขณะที่คำนวณครับ
ข้อดี / จุดเด่น & ข้อเสีย / ข้อควรระวัง
ข้อดี / จุดเด่น
- ประมวลผลระดับ Big Data สเกลขยายเครื่องคอมพิวเตอร์แบบแนวนอนได้ไร้ขีดจำกัด
- ประมวลผลในแรม (In-memory computing) ทำงานได้รวดเร็วกว่า Hadoop แบบเดิมเกือบ 100 เท่า
ข้อเสีย / ข้อควรระวัง
- เกิดปัญหาแรมหมดบ่อยครั้ง (Out-of-memory) เมื่อข้อมูลในแต่ละ Partition ไม่สมดุลกัน
- ต้องการงบประมาณสูงและควบคุมดูแลเครื่องในคลัสเตอร์ค่อนข้างลำบาก
ปฏิบัติการจริง (Lab Practice)
แล็บ: สร้างและรัน Spark Job ในการประมวลผลข้อมูลระดับล้านแถว
วิธีรันแล็บปฏิบัติการบนเครื่องจริง (Local Terminal Execution Blueprint)
เนื่องจากปฏิบัติการ Data Engineering / DevOps ระดับสูงต้องรันบน Environment จริง ขั้นตอนด้านล่างนี้คือคำสั่งสำหรับนำไปรันบน Terminal / Docker ในเครื่องของคุณ:
mkdir -p pipecraft-lab && cd pipecraft-lab python3 -m venv venv && source venv/bin/activate pip install --upgrade pip pandas polars pytest requests
# Run execution pipeline test
python3 -c "
import polars as pl
print(' [PipeCraft Lab] Running local pipeline engine...')
df = pl.DataFrame({'id': [1, 2, 3], 'status': ['SUCCESS', 'SUCCESS', 'AUDITED']})
print(df)
"
# Verify clean execution exit status echo " PipeCraft Local Lab Execution Completed Successfully!"
💡 คำแนะนำ & ทริกเด็ด
ก่อนจะขยับไปใช้งาน Spark ให้เช็คขนาดข้อมูลดิบเสมอ หากมีขนาดต่ำกว่า 100GB แนะนำให้ใช้ DuckDB รันในโหมด Single-node เครื่องเดี่ยว ซึ่งจะทำงานสำเร็จเร็วกว่าเนื่องจากไม่มีเวลาหน่วงของการเปิดคลัสเตอร์และการ Shuffling