Spark SQL & DataFrame

1เตรียมข้อมูลตัวอย่าง (Orders Dataset)

%%writefile /home/jovyan/data/orders.csv
order_id,customer,product,category,quantity,price,order_date
1,Alice,Laptop,Electronics,1,32000,2026-01-05
2,Bob,Mouse,Electronics,2,450,2026-01-05
3,Alice,Desk,Furniture,1,4500,2026-01-06
4,Charlie,Laptop,Electronics,1,31500,2026-01-07
5,Bob,Chair,Furniture,2,1200,2026-01-08
6,Diana,Monitor,Electronics,2,6000,2026-01-08
7,Alice,Mouse,Electronics,3,450,2026-01-09
8,Charlie,Desk,Furniture,1,4500,2026-01-10

2โหลดข้อมูลพร้อมกำหนด Schema เอง

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType, DateType

spark = SparkSession.builder \
    .appName("Lab2-SparkSQL") \
    .master("spark://spark-master:7077") \
    .getOrCreate()

schema = StructType([
    StructField("order_id", IntegerType(), False),
    StructField("customer", StringType(), False),
    StructField("product", StringType(), False),
    StructField("category", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("price", DoubleType(), False),
    StructField("order_date", DateType(), False),
])

df = spark.read.format("csv") \
    .option("header", "true") \
    .schema(schema) \
    .load("/home/jovyan/data/orders.csv")

df.printSchema()
df.show()

3Query ด้วย DataFrame API

# ยอดขายรวมต่อลูกค้า (quantity * price)
from pyspark.sql.functions import col, sum as spark_sum, round as spark_round

df.withColumn("total", col("quantity") * col("price")) \
  .groupBy("customer") \
  .agg(spark_round(spark_sum("total"), 2).alias("total_spent")) \
  .orderBy(col("total_spent").desc()) \
  .show()

4Query แบบเดียวกัน แต่เขียนเป็น SQL

df.createOrReplaceTempView("orders")

spark.sql("""
    SELECT customer, ROUND(SUM(quantity * price), 2) AS total_spent
    FROM orders
    GROUP BY customer
    ORDER BY total_spent DESC
""").show()

+--------+-----------+
|customer|total_spent|
+--------+-----------+
|   Alice|    38050.0|
|   Diana|    12000.0|
| Charlie|    36000.0|
|     Bob|     3300.0|
+--------+-----------+

5พิสูจน์ว่าทั้งสองวิธีให้แผนการทำงานเดียวกัน

plan_dataframe = df.withColumn("total", col("quantity") * col("price")) \
    .groupBy("customer").agg(spark_sum("total"))._jdf.queryExecution().optimizedPlan()

plan_sql = spark.sql("""
    SELECT customer, SUM(quantity * price) FROM orders GROUP BY customer
""")._jdf.queryExecution().optimizedPlan()

print("แผนเหมือนกันหรือไม่:", str(plan_dataframe) == str(plan_sql))
# หรือใช้วิธีง่ายกว่า: เรียก .explain() แล้วเทียบด้วยตาก็ได้
df.groupBy("customer").agg(spark_sum(col("quantity") * col("price"))).explain()

6เขียนผลลัพธ์เป็น Parquet

summary = df.withColumn("total", col("quantity") * col("price")) \
    .groupBy("category").agg(spark_sum("total").alias("category_total"))

summary.write.mode("overwrite").parquet("/home/jovyan/data/category_summary.parquet")

# อ่านกลับมาตรวจสอบ
spark.read.parquet("/home/jovyan/data/category_summary.parquet").show()

7อ่าน Explain ให้เป็น ไม่ใช่เพียงเปิดดู

query = df.filter(col("category") == "Electronics") \
    .select("customer", "quantity", "price") \
    .withColumn("total", col("quantity") * col("price")) \
    .groupBy("customer").agg(spark_sum("total").alias("revenue"))

query.explain("extended")
query.explain("formatted")

วงคำว่า Exchange ทุกจุด เพราะมักหมายถึง Shuffle แล้วอธิบายว่าทำไมข้อมูลต้องกระจายใหม่ Plan ที่มี Exchange ไม่ได้ผิด แต่ควรรู้ว่าจ่ายเพื่อ operation ใด

8Column Pruning และ Predicate Pushdown

สร้าง Parquet ที่มีหลายคอลัมน์แล้วอ่านเฉพาะสองคอลัมน์พร้อม Filter ตรวจ ReadSchema และ PushedFilters

orders = spark.read.parquet("/home/jovyan/data/orders.parquet")
small = orders.select("customer", "price") \
    .filter(col("price") > 1000)
small.explain("formatted")

ถ้า Source รองรับ Spark อ่านเฉพาะ Column และข้าม Data Group บางส่วนได้ หลักคือ “เอา Code ไปหา Data และอ่านเท่าที่ต้องใช้” ไม่ใช่โหลดทุกอย่างแล้วค่อยทิ้งใน Python

9Built-in Function กับ Python UDF

from pyspark.sql.functions import lower, trim, udf
from pyspark.sql.types import StringType

builtin = df.withColumn("customer_clean", lower(trim(col("customer"))))

@udf(StringType())
def normalize_name(x):
    return x.strip().lower() if x else None

python_udf = df.withColumn("customer_clean", normalize_name(col("customer")))

builtin.explain("formatted")
python_udf.explain("formatted")

Built-in Expression อยู่ในภาษาที่ Catalyst เข้าใจ ส่วน Python UDF อาจสร้าง Boundary และเป็นกล่องดำ ใช้ UDF เมื่อจำเป็น ไม่ใช่เพราะเขียน Python ถนัดกว่า แล้ววัดผลด้วยข้อมูลที่ใหญ่พอ

10Join Strategy และ Broadcast

customers = spark.createDataFrame([
    ("Alice", "Gold"), ("Bob", "Silver"), ("Carol", "Gold")
], ["customer", "segment"])

joined = df.join(customers, "customer", "left")
joined.explain("formatted")

from pyspark.sql.functions import broadcast
broadcast_join = df.join(broadcast(customers), "customer", "left")
broadcast_join.explain("formatted")

ดูว่า Physical Plan ใช้ BroadcastHashJoin หรือไม่ ตารางเล็กถูกส่งไป Executor เพื่อลด Shuffle ฝั่งใหญ่ แต่ถ้าตาราง “เล็ก” โตขึ้นใน Production การบังคับ Broadcast อาจกิน Memory จึงต้องตรวจขนาดและสถิติ

11NULL และ Data Quality

from pyspark.sql.functions import count, when, isnan

df.select([
    count(when(col(c).isNull(), c)).alias(c)
    for c in df.columns
]).show()

df.filter(col("price").isNull()).show()
df.selectExpr("customer", "coalesce(price, 0) AS price_or_zero").show()

NULL ไม่เท่ากับศูนย์ และ price = NULL ไม่ใช่วิธีกรองที่ถูก ต้องใช้ IS NULL การเติมศูนย์ควรมีเหตุผลทางธุรกิจ ไม่ใช่เพียงทำให้โมเดลรับข้อมูลได้

12Partitioned Parquet และ Small Files

df.write.mode("overwrite") \
  .partitionBy("category") \
  .parquet("/home/jovyan/data/orders_by_category")

filtered = spark.read.parquet("/home/jovyan/data/orders_by_category") \
  .filter(col("category") == "Electronics")
filtered.explain("formatted")

ตรวจ PartitionFilters แต่ระวัง Partition ด้วยคอลัมน์ Cardinality สูง เช่น order_id จะสร้าง Directory/Small File จำนวนมาก ทดลอง repartition(2) ก่อนเขียนและนับไฟล์เพื่อเห็นความสัมพันธ์ระหว่าง Execution Partition กับ Output File

13Window Function

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, sum as spark_sum

w_rank = Window.partitionBy("category").orderBy(col("price").desc())
w_total = Window.partitionBy("customer").orderBy("order_id") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

df.withColumn("rank_in_category", row_number().over(w_rank)) \
  .withColumn("running_spend", spark_sum(col("quantity") * col("price")).over(w_total)) \
  .show()

Window ไม่ยุบแถวเหมือน GroupBy แต่ต้อง Partition/Sort ตามกรอบ จึงอาจ Shuffle แพง ตรวจ Plan และอธิบายว่าทำไม Window สองชุดที่ใช้ Key/Order ต่างกันอาจสร้าง Sort หลายครั้ง

14Mini Benchmark และสิ่งที่ต้องส่ง

สร้างข้อมูลอย่างน้อยหนึ่งล้านแถว เปรียบเทียบ CSV กับ Parquet, Built-in กับ UDF และ Join ปกติกับ Broadcast รันซ้ำอย่างน้อยสามครั้ง แยกเวลารอบแรก/รอบถัดไป และเก็บ Plan

• Notebook พร้อม Schema ที่กำหนดเอง
• DataFrame กับ SQL ที่ตอบคำถามเดียวกัน พร้อม Plan Comparison
• ภาพ Column Pruning/Pushdown
• Join Plan สองแบบ
• Data Quality Report และ Quarantine ตัวอย่าง
• ตาราง Benchmark พร้อมข้อจำกัด

สรุปและขั้นตอนถัดไป: DataFrame ทำให้ระบบเห็น Schema และ Algebra แต่ข้อมูลที่สะอาดยังไม่กลายเป็นโมเดล Lab 3 จะนำ Transformation กับ Estimator มาประกอบเป็น ML Pipeline โดยเน้น Data Leakage, Metric และ Reproducibility มากกว่าการได้ Accuracy ตัวเลขหนึ่งค่า

20วิธีทำแล็บให้ได้มากกว่าคำว่า “รันผ่าน”

แล็บนี้ไม่ได้วัดว่าเราพิมพ์คำสั่งตามตัวอย่างได้ครบหรือไม่ เพราะการรันผ่านอาจเกิดจากความบังเอิญ ค่า Default หรือข้อมูลที่เล็กเกินไปจนมองไม่เห็นปัญหา เป้าหมายคือฝึกตั้งสมมติฐาน เปลี่ยนตัวแปรครั้งละอย่าง เก็บหลักฐาน และอธิบายผลด้วยแนวคิดของระบบกระจาย ถ้าผลไม่ตรงกับที่คาด นั่นไม่ใช่ความล้มเหลว แต่เป็นจุดเริ่มต้นของคำถามที่ดี

1. คาดก่อนรัน: เขียนหนึ่งหรือสองประโยคว่าคิดว่าจะเกิดอะไรและเพราะอะไร
2. กำหนดสิ่งที่คุม: ใช้ข้อมูล โค้ด และทรัพยากรเดิม แล้วเปลี่ยนเฉพาะตัวแปรที่กำลังศึกษา
3. เก็บหลักฐาน: บันทึกเวลา แผนการทำงาน Metric, Log หรือภาพ Spark UI ไม่อาศัยความรู้สึกว่า “เร็วขึ้น”
4. อธิบายกลไก: เชื่อมผลกับ Partition, Shuffle, State, Memory, Network หรือ Scheduling
5. บอกข้อจำกัด: ผลจาก Laptop และข้อมูลจำลองอาจไม่เหมือน Production ให้ระบุสิ่งที่ยังสรุปไม่ได้

21สมุดบันทึกการทดลอง

อย่าจดเฉพาะผลลัพธ์สุดท้าย ควรจด Environment ด้วย เช่น Spark version, Python version, จำนวน Core, Memory, จำนวน Partition, Seed และขนาดข้อมูล เปรียบเหมือนใบเสร็จและสูตรอาหารที่ทำให้คนอื่นย้อนทำจานเดิมได้ หากไม่มีข้อมูลเหล่านี้ ผลที่ดูน่าเชื่อถืออาจทำซ้ำไม่ได้

22ชุดการทดลองหลัก: Spark SQL และ DataFrame

23บันไดแก้ปัญหาเมื่อผลไม่เป็นอย่างที่คิด

เวลางานล้ม อย่าเริ่มด้วยการลบทุกอย่างแล้วติดตั้งใหม่ เพราะเราอาจทำลายหลักฐานที่บอกสาเหตุ ให้ไล่จากชั้นนอกเข้าสู่ชั้นในเหมือนช่างไฟที่ตรวจตั้งแต่ปลั๊ก สายไฟ ฟิวส์ ไปจนถึงเครื่องใช้ไฟฟ้า

1. ยืนยันอาการ: คัดลอก Error แรก ไม่ใช่เฉพาะบรรทัดสุดท้าย และบันทึกคำสั่งที่ทำให้เกิดซ้ำ
2. ลดปัญหา: ใช้ข้อมูลเล็กที่สุดและโค้ดสั้นที่สุดที่ยังทำให้ Error เกิด
3. ตรวจสัญญาข้อมูล: ดู Schema, Type, Null, Key และ Path ก่อนสงสัย Cluster
4. ตรวจ Plan และ UI: ดูว่างานเริ่มจริงหรือค้างอยู่ก่อนสร้าง Job; แยกปัญหา Driver, Executor และ Storage
5. ตรวจทรัพยากร: ดู Memory, Disk, Port, Permission และ Container status
6. เปลี่ยนทีละจุด: เมื่อแก้แล้ว ให้ย้อนอธิบายว่าการเปลี่ยนนั้นจัดการสาเหตุใด

24คำถามชวนคิดหลังทำแล็บ

1. ผลที่เห็นเกิดจาก Algorithm หรือเกิดจากการจัดวางข้อมูลและข้อจำกัดของเครื่อง?
2. ถ้าข้อมูลใหญ่ขึ้นหนึ่งร้อยเท่า ขั้นตอนไหนจะพังก่อน และเรามีหลักฐานอะไร?
3. ถ้าเครื่องหนึ่งหายไประหว่างงาน คำตอบจะยังถูกต้องหรือเพียงแค่งานยังรันต่อได้?
4. ค่า Default ใดช่วยเราอยู่ และค่าใดอาจกลายเป็นกับดักเมื่อขึ้น Production?
5. ถ้าต้องอธิบายผลให้เพื่อนที่ไม่ได้เห็นหน้าจอ เรามีหลักฐานครบพอหรือยัง?

25สิ่งที่ต้องส่ง

• Notebook หรือ Source code ที่รันจากต้นจนจบได้ โดยไม่พึ่งลำดับคำสั่งที่ซ่อนอยู่
• README ระบุวิธีเริ่มระบบ Version, Dependency และคำสั่งที่ใช้
• สมุดบันทึกอย่างน้อยสามการทดลอง มีสมมติฐาน ตัวแปร หลักฐาน และข้อสรุป
• ภาพหรือ Export จาก Spark UI/Query Plan ที่วงส่วนสำคัญและมีคำอธิบาย
• กรณีผิดพลาดอย่างน้อยหนึ่งกรณี พร้อมการวิเคราะห์ ไม่ตัดทิ้งเพราะ “ทำไม่สำเร็จ”
• ข้อจำกัดของผลและสิ่งที่จะทดลองต่อ หากมีข้อมูลหรือเครื่องมากขึ้น

26เกณฑ์ตรวจงานแบบสั้น

27สรุปและขั้นตอนถัดไป

หัวใจของแล็บนี้ไม่ใช่จำคำสั่ง แต่คือการมองเห็นว่า Spark SQL และ DataFrame ตัดสินใจอย่างไรเมื่อข้อมูล งาน และทรัพยากรถูกกระจายออกจากกัน เมื่อทำเสร็จแล้วควรตอบได้สามเรื่อง: ระบบทำอะไร หลักฐานใดแสดงว่ามันทำเช่นนั้น และเงื่อนไขใดทำให้พฤติกรรมเปลี่ยน

ก่อนเปิดแล็บถัดไป ลองย่อสิ่งที่เรียนรู้ให้เหลือหนึ่งภาพ หนึ่งตาราง และสามประโยค หากย่อไม่ได้ อาจยังมีส่วนที่เราเห็นผลแต่ยังไม่เข้าใจกลไก ให้กลับไปเลือกการทดลองที่เล็กลง แล้วค่อยต่อภาพกลับขึ้นมาใหม่

28สถานการณ์จำลอง: จากข้อมูลก้อนเล็กไปสู่งานที่เริ่มมีปัญหา

ให้เริ่มด้วยข้อมูลก้อนเล็กที่ตรวจคำตอบด้วยมือได้ จากนั้นขยายเป็นสามระดับ โดยไม่เปลี่ยนความหมายของโจทย์ ระดับแรกใช้ตรวจความถูกต้อง ระดับที่สองใช้เปิดให้เห็น Parallelism และระดับที่สามใช้เปิดให้เห็นคอขวด วิธีนี้ช่วยแยกคำถามสองข้อที่มักปนกัน คือ “คำตอบถูกหรือไม่” กับ “ระบบรองรับขนาดงานได้หรือไม่” ระบบที่เร็วแต่ตอบผิดไม่มีประโยชน์ และระบบที่ตอบถูกเฉพาะข้อมูลสิบแถวก็ยังไม่ใช่คำตอบของงานขนาดใหญ่

29อ่าน Spark UI ให้เป็นเรื่องราว ไม่ใช่เพียงหน้าจอสีสวย

เริ่มจาก Job ว่าถูกสร้างเพราะ Action ใด แล้วลงไปที่ Stage เพื่อดูว่า Shuffle แบ่งเส้นทางตรงไหน จากนั้นจึงดู Task ว่ากระจายตัวสม่ำเสมอหรือมีตัวใดลากยาว สุดท้ายดู Executor ว่างานไปอยู่เครื่องใด ใช้ Memory เท่าไร และมี Failure หรือ Retry หรือไม่ ลำดับนี้เหมือนอ่านแผนที่จากประเทศ จังหวัด ถนน แล้วจึงถึงบ้าน หากกระโดดไปดู Task หนึ่งตัวทันที เราอาจเห็นอาการแต่ไม่เห็นบริบท

การจับภาพ UI ควรใส่ลูกศรหรือคำอธิบายว่ากำลังใช้ภาพนั้นสนับสนุนข้อสรุปใด ภาพทั้งหน้าที่ไม่มีคำอธิบายเหมือนแนบผลตรวจสุขภาพโดยไม่บอกว่าค่าไหนผิดปกติ หลักฐานที่ดีไม่จำเป็นต้องเยอะ แต่ต้องเชื่อมกับคำถามได้ตรงจุด

30แยกความถูกต้อง ความเร็ว และความทนทานออกจากกัน

สามเรื่องนี้สัมพันธ์กันแต่ทดแทนกันไม่ได้ งานอาจเร็วขึ้นเพราะเผลอทิ้งข้อมูลบางส่วน งานอาจทนต่อ Worker หายแต่สร้างข้อมูลซ้ำ หรือคำตอบอาจถูกแต่ใช้ทรัพยากรมากเกินจำเป็น ทุกครั้งที่ปรับระบบจึงควรตรวจทั้งสามแกน ไม่ประกาศชัยชนะจากเวลาเพียงตัวเดียว

1. ความถูกต้อง: สร้างข้อมูลจิ๋วที่รู้คำตอบล่วงหน้า ใส่ Null, Duplicate, Key ที่ไม่มีคู่ และค่าขอบเขต แล้วตรวจผลทีละกรณี
2. ความเร็ว: วัดหลายรอบ แยก Warm-up ออกจากรอบจริง ระบุขนาดข้อมูลและทรัพยากร และรายงานทั้งค่ากลางกับช่วง ไม่เลือกเฉพาะรอบที่ดีที่สุด
3. ความทนทาน: ทำให้ส่วนหนึ่งล้มอย่างตั้งใจ ตรวจว่างาน Retry, Resume หรือคำนวณใหม่จาก Lineage และตรวจว่าคำตอบหลังการกู้ไม่ซ้ำหรือขาด
4. ต้นทุน: เวลาเร็วขึ้นอาจแลกกับ Memory, Storage, Network หรือความซับซ้อนในการดูแล ให้บันทึกสิ่งที่จ่ายเพิ่มเสมอ

31จาก Error Message ไปสู่คำอธิบาย

ให้เลือก Error จริงหนึ่งครั้งจากแล็บแล้วเขียน Postmortem สั้น ๆ โดยไม่กล่าวโทษผู้ใช้หรือเครื่องมือ เริ่มจากผลกระทบ ตามด้วยลำดับเหตุการณ์ สาเหตุใกล้ตัว สาเหตุราก วิธีแก้เฉพาะหน้า และวิธีป้องกันซ้ำ ตัวอย่างเช่น “Job ล้มเพราะ Out of Memory” ยังเป็นเพียงอาการ สาเหตุรากอาจเป็นการ collect ข้อมูลทั้งหมดมายัง Driver, Broadcast ตารางใหญ่เกินไป หรือ Partition หนึ่งก้อนใหญ่ผิดปกติ ซึ่งมีวิธีแก้คนละแบบ

จุดสำคัญคือแยก “แก้ให้งานผ่านครั้งนี้” ออกจาก “แก้ไม่ให้เกิดซ้ำ” การเพิ่ม Memory อาจช่วยงานรอบนี้ แต่ถ้าข้อมูลโตต่อเนื่อง ปัญหาจะกลับมา การเปลี่ยนรูปแบบ Aggregation, ลดข้อมูลก่อน Shuffle หรือเลิกดึงทุกอย่างเข้า Driver อาจแตะสาเหตุได้ตรงกว่า

32แบบฝึกสอนกลับ: อธิบายให้เพื่อนฟังในห้านาที

จับคู่กับเพื่อนแล้วผลัดกันอธิบาย Spark SQL และ DataFrame โดยห้ามเริ่มจากชื่อ Class หรือ API ให้เริ่มจากปัญหาที่ระบบกำลังแก้ ใช้ภาพหนึ่งภาพและตัวอย่างข้อมูลไม่เกินสิบแถว ผู้ฟังมีหน้าที่ถามว่า “รู้ได้อย่างไร” และ “ถ้าข้อมูลใหญ่ขึ้นจะเกิดอะไร” วิธีนี้ช่วยเปิดจุดที่เราจำคำสั่งได้แต่ยังเชื่อมเหตุผลไม่ครบ

• นาทีที่ 1: ปัญหาคืออะไร และเหตุใดเครื่องเดียวหรือวิธีตรงไปตรงมาจึงเริ่มไม่พอ
• นาทีที่ 2: ข้อมูลและงานถูกแบ่งอย่างไร ส่วนใดอยู่ที่ Driver และส่วนใดอยู่ที่ Executor
• นาทีที่ 3: จุดใดต้องสื่อสารข้ามเครื่อง และต้นทุนเกิดตรงไหน
• นาทีที่ 4: แสดงหลักฐานหนึ่งชิ้นจากแล็บแล้วตีความ
• นาทีที่ 5: บอกข้อจำกัดหนึ่งข้อและการทดลองถัดไปหนึ่งอย่าง