การประมวลผลข้อมูลขนาดใหญ่แบบกระจายด้วย Apache Spark

ทฤษฎีและกลไกการทำงาน (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()`

ทฤษฎีและกลไกการทำงาน (How it works): แรมหน่วยความจำ Heap Memory ของ JVM บนโหนด Executor ถูกจัดสรรเพื่อความเสถียร: - **Execution Memory**: ใช้คำนวณการจัดกลุ่มคีย์ การเรียงข้อมูล และการทำ Shuffling หากหน่วยความจำในส่วนนี้มีเนื้อที่ไม่พอ Spark จะยอมเขียนหน่วยความจำส่วนเกินนี้บันทึกลงสู่ฮาร์ดดิสก์แทน (Spill to Disk) - **Storage Memory**: เก็บรักษากลุ่มตารางข้อมูลแคช (`.cache()`) และตัวแปรกลุ่ม Broadcast - **User Memory**: เก็บโครงสร้างข้อมูลปกติหรือตัวแปรโปรแกรมของ Python

# 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` หรือแก้ไขขอบเขตพาร์ทิชันให้เล็กลง

ทฤษฎีและกลไกการทำงาน (How it works): การเชื่อมตารางข้อมูล (JOIN) ในคลัสเตอร์ Spark มี 2 ยุทธวิธีหลัก: - **Shuffle Hash Join**: ใช้เมื่อทั้งสองตารางมีขนาดใหญ่ Spark จะหั่นเรียงข้อมูลและยิงส่งสลับข้อมูลที่มีคีย์เดียวกันข้ามเครือข่ายเน็ตเวิร์กไปไว้ที่ Executor เครื่องเดียวกัน ส่งผลให้เกิดความหน่วงเครือข่ายสูง - **Broadcast Hash Join**: ใช้เมื่อมีตารางหนึ่งขนาดเล็กมาก (เช่น <10MB) ไดรเวอร์จะสำเนาก๊อปปี้ตารางเล็กส่งไปจดจำใน Storage memory ของทุก Executors เพื่อให้ Executors เชื่อมตารางใหญ่ขนานกันในแรมเครื่องตัวเองทันที ข้ามขั้นตอนยิงสลับข้อมูลข้ามเน็ตเวิร์ก

# PySpark code demonstrating Broadcast Hash Join
from pyspark.sql.functions import broadcast

large_transactions_df = spark.read.parquet("s3a://datalake/silver/transactions/")
small_dim_stores_df = spark.read.parquet("s3a://datalake/silver/stores/")

# Forces Broadcast Join on the smaller stores lookup table
enriched_sales_df = large_transactions_df.join(
    broadcast(small_dim_stores_df),
    on="store_id",
    how="inner"
)
                    

Use Case ในชีวิตจริง (Real-world Scenario): ตารางบันทึกการทำธุรกรรม 500GB เชื่อมเข้าตารางรหัสแคมเปญโฆษณา 5MB การเลือกบรูทฟอร์ซ Broadcast Join ช่วยหลีกเลี่ยงขั้นตอน Shuffle ย่นระยะเวลาการรันจ๊อบจากเดิม 1 ชั่วโมงเหลือเพียง 4 นาที

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หากสั่ง Broadcast ตารางที่มีขนาดใหญ่เกินขีดจำกัดแรมสูงสุด (เช่น 2GB) จะทำให้ Driver และ Executors ล้มเหลว OOM ทันที แก้ไขโดยปิดใช้ Broadcast หากพบว่าตาราง Lookup ขยายขนาดเกินความปลอดภัย

ทฤษฎีและกลไกการทำงาน (How it works): ปัญหา **Data Skew** เกิดขึ้นเมื่อข้อมูลส่วนใหญ่ไปกระจุกตัวอยู่ภายใต้คีย์พาร์ทิชันไม่กี่ตัว (เช่น ลูกค้าองค์กรยักษ์ใหญ่สร้างรายการซื้อมากกว่าลูกค้าทั่วไปล้านเท่า) ส่งผลให้ Executor ตัวที่ได้พาร์ทิชันตัวนั้นไปรันต้องแบกภาระงานหนักและล่าช้ากว่าเครื่องอื่น วิธีแก้ไขทำได้โดยการซอยสัดส่วนพาร์ทิชันเฉลี่ยผ่านสูตร `spark.sql.shuffle.partitions` หรือเทคนิคการเติมเกลือสุ่มตัวเลขต่อท้ายคีย์ (Salting) เพื่อช่วยกระจายข้อมูลให้ออกไปรันบนโหนดอื่นเฉลี่ยกัน

# Salting technique to distribute skewed partition keys evenly
from pyspark.sql.functions import col, concat, lit, rand

# Add a random integer salt suffix (0-3) to the skew join key column
skewed_df = skewed_df.withColumn("salt_key", concat(col("customer_id"), lit("_"), (rand() * 4).cast("int")))
# Duplicate lookup tables entries with mapping keys to align joins
                    

Use Case ในชีวิตจริง (Real-world Scenario): ในท่อส่งข้อมูลการสั่งของยักษ์ใหญ่ที่ยอดคลิกซื้อ 80% กระจุกตัวอยู่ที่กรุงเทพฯ วิศวกรข้อมูลทำ Salting บนคีย์ `province` ช่วยลดเวลาคอยงานจ๊อบค้างคิวใน Executor เครื่องสุดท้ายลงได้เกินครึ่ง

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): จำนวนพาร์ทิชันที่มากเกินไปส่งผลลบ (Too many small partitions) ทำให้หน่วยความจำ Metadata บน Driver ทำงานหนักและรันล่วงเวลาช้า แก้ไขโดยให้ปรับจำนวนเฉลี่ยพาร์ทิชันให้ขนาดแต่ละพาร์ทิชันข้อมูลดิบในแรมอยู่ที่ประมาณ 100MB - 200MB