Spark Core & RDD

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

%%writefile /home/jovyan/data/sample.txt
big data analytics is about scale
spark makes big data processing fast
big data needs distributed systems
scale out beats scale up for big data

2สร้าง SparkContext และโหลดข้อมูล

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Lab1-RDD") \
    .master("spark://spark-master:7077") \
    .getOrCreate()
sc = spark.sparkContext

lines = sc.textFile("/home/jovyan/data/sample.txt")
print("จำนวนบรรทัด:", lines.count())   # action ตัวแรก — บังคับให้อ่านไฟล์จริง

3Word Count แบบเต็มรูปแบบ

word_counts = (
    lines
    .flatMap(lambda line: line.split(" "))   # map: แตกแต่ละบรรทัดเป็นคำ
    .map(lambda word: (word, 1))              # map: แปลงเป็น (คำ, 1)
    .reduceByKey(lambda a, b: a + b)          # reduce: รวมค่าของคำเดียวกัน
)

for word, count in word_counts.collect():     # action: ดึงผลลัพธ์ทั้งหมดกลับมา
    print(f"{word}: {count}")

big: 4
data: 4
analytics: 1
scale: 3
...

4ทดลอง: Transformation เป็น Lazy จริงไหม

def debug_map(word):
    print(f"กำลังประมวลผลคำ: {word}")   # ไม่ควรเห็น print นี้จนกว่าจะมี action
    return (word, 1)

mapped = lines.flatMap(lambda l: l.split(" ")).map(debug_map)
print("--- ตอนนี้ยังไม่มี print ใด ๆ ปรากฏ เพราะยังไม่มี action ---")

result = mapped.reduceByKey(lambda a, b: a + b).collect()   # action — ตอนนี้ print ถึงจะทำงาน
print("--- ตอนนี้ print ปรากฏแล้ว เพราะ collect() บังคับให้รันทั้ง chain ---")

5ทดลอง: ผลของ cache() ต่อความเร็ว

import time

# โหลดไฟล์ที่ใหญ่กว่าเดิม (ทำซ้ำเนื้อหาเดิม 50,000 ครั้ง เพื่อให้เห็นผลชัด)
big_lines = sc.parallelize(["big data analytics is about scale"] * 50000)

# ไม่ cache — ต้องคำนวณ transformation ใหม่ทุกครั้งที่มี action
start = time.time()
big_lines.flatMap(lambda l: l.split(" ")).count()
big_lines.flatMap(lambda l: l.split(" ")).count()
print("ไม่ cache:", time.time() - start, "วินาที")

# cache — คำนวณครั้งแรกแล้วเก็บผลไว้ใน memory
cached = big_lines.flatMap(lambda l: l.split(" ")).cache()
start = time.time()
cached.count()   # ครั้งแรก: คำนวณจริงและเก็บลง cache
cached.count()   # ครั้งที่สอง: อ่านจาก cache โดยตรง เร็วกว่ามาก
print("cache แล้ว:", time.time() - start, "วินาที")

6อ่าน Lineage และแบ่ง Narrow/Wide Transformation

RDD ไม่ได้เก็บเพียงข้อมูล แต่จำสูตรว่าผลลัพธ์มาจากไหน toDebugString() ช่วยเห็น Dependency Chain และจุดที่เกิด Shuffle

lines = sc.textFile("/home/jovyan/data/text.txt")
words = lines.flatMap(lambda line: line.lower().split())
pairs = words.map(lambda word: (word, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
print(counts.toDebugString().decode("utf-8"))

map และ filter มักเป็น Narrow Dependency เพราะ Partition ปลายทางใช้ Parent จำนวนน้อย ส่วน reduceByKey, groupByKey, sortByKey และ join มักต้อง Shuffle ข้อมูลตาม Key จึงเกิด Stage Boundary

7พิสูจน์ว่า Action สร้าง Job

เปิด Spark UI แล้วรัน Action ทีละคำสั่ง จดจำนวน Job, Stage และ Task อย่ารัน Cell ทั้งก้อน เพราะจะจับคู่ Code กับ UI ยาก

base = sc.parallelize(range(1_000_000), 8)
even = base.filter(lambda x: x % 2 == 0)
squared = even.map(lambda x: x * x)

# จุดนี้ยังไม่ควรมี Job ใหม่จาก Transformation
print("Partitions:", squared.getNumPartitions())

print("Count:", squared.count())       # Action 1
print("First five:", squared.take(5)) # Action 2

Action สองครั้งอาจคำนวณ Lineage ซ้ำ ถ้าไม่ได้ Cache Spark ไม่สรุปเองว่าเรา “น่าจะใช้ซ้ำ” เพราะ Cache มีราคาและอาจไม่คุ้ม

8reduceByKey กับ groupByKey: คำตอบเหมือน ต้นทุนไม่เหมือน

data = sc.parallelize([
    ("A", 10), ("A", 20), ("B", 5), ("A", 7), ("B", 8)
], 4)

good = data.reduceByKey(lambda a, b: a + b)
expensive = data.groupByKey().mapValues(lambda xs: sum(xs))

print(good.collect())
print(expensive.collect())

reduceByKey รวมบางส่วนใน Partition ก่อนส่งข้าม Network ส่วน groupByKey ส่งค่าทุกตัวไปกองที่ปลายทางแล้วค่อย Sum หลักนี้เหมือนให้แต่ละสาขารวมยอดก่อนส่งสำนักงานใหญ่ แทนการขนใบเสร็จทุกใบ

9Partition และ Load Balance

rdd = sc.parallelize(range(100), 6)
sizes = rdd.mapPartitionsWithIndex(
    lambda idx, it: [(idx, sum(1 for _ in it))]
).collect()
print(sizes)

จากนั้นสร้างข้อมูล Key เอียงและดูจำนวน Record หลัง partitionBy

skewed = sc.parallelize(
    [("HOT", i) for i in range(100_000)] +
    [(f"K{i}", i) for i in range(10_000)], 8
)
partitioned = skewed.partitionBy(8)
print(partitioned.mapPartitionsWithIndex(
    lambda idx, it: [(idx, sum(1 for _ in it))]
).collect())

ถ้า HOT Key ไป Partition เดียว Task นั้นจะช้ากว่ากลุ่ม Job รอ Task ช้าที่สุด การเพิ่ม Partition ไม่แยก Key เดียวออกเอง ต้องเปลี่ยน Algorithm เช่น Salting หรือ Pre-aggregation

10Cache, Persist และ Materialization

cache() ยัง Lazy การเรียกไม่ทำให้ข้อมูลเข้า Memory จนมี Action แรก ควรแยกเวลารอบ Warm-up กับเวลารอบ Reuse

from pyspark import StorageLevel
import time

base = sc.parallelize(range(5_000_000), 16).map(lambda x: (x % 1000, x * x))
base.persist(StorageLevel.MEMORY_AND_DISK)

t0 = time.perf_counter(); base.count(); first = time.perf_counter() - t0
t0 = time.perf_counter(); base.count(); second = time.perf_counter() - t0
print("materialize:", first, "reuse:", second)
base.unpersist()

11Fault Tolerance จาก Lineage

สร้าง RDD จาก Deterministic Transformation แล้วสังเกตว่า Spark รู้ Parent Dependency แต่ Lineage ยาวมากทำให้ Recovery แพง Checkpoint ตัด Lineage ด้วยการบันทึกข้อมูลที่เชื่อถือได้

sc.setCheckpointDir("/home/jovyan/data/rdd-checkpoints")
rdd = sc.parallelize(range(1_000_000), 16)
for _ in range(20):
    rdd = rdd.map(lambda x: (x * 17 + 3) % 1000003)

print(rdd.toDebugString().decode())
rdd.checkpoint()
rdd.count()  # materialize checkpoint
print(rdd.toDebugString().decode())

Checkpoint ต่างจาก Cache: Cache เร่ง Reuse และอาจสูญหายได้; Checkpoint ตัด Dependency เพื่อ Recovery แต่มี I/O เพิ่ม บางงานใช้ทั้งสองร่วมกัน

12กับดัก collect() และ Driver

collect() ดึงทุก Record กลับ Driver งานกระจายอาจล้มตรงปลายเมื่อข้อมูลทั้งหมดอัดเข้าเครื่องเดียว ระหว่างสำรวจใช้ take, takeOrdered, Aggregate หรือเขียน Storage

# ปลอดภัยกว่าสำหรับดูตัวอย่าง
print(counts.takeOrdered(10, key=lambda kv: -kv[1]))

# หลีกเลี่ยงกับผลลัพธ์ใหญ่
# everything = huge_rdd.collect()

13Mini Experiment: วัด Partition, Cache และ Shuffle

ออกแบบตารางทดลอง 3 ปัจจัย: Partition 4/8/16, Cache yes/no และ Aggregation reduceByKey/groupByKey รันซ้ำอย่างน้อย 3 ครั้ง บันทึก Median, Stage Count, Shuffle Read/Write และข้อสังเกต

รายงานผลที่ขัดสมมุติฐานได้ เช่นข้อมูลเล็กจน Overhead กลบความต่าง นั่นเป็นผลการทดลอง ไม่ใช่ข้อผิดพลาดที่ต้องซ่อน

14สิ่งที่ต้องส่งและสรุปขั้นตอนถัดไป

• Notebook ที่รันจากต้นจนจบ
• ภาพ DAG/Stage ของ Narrow และ Wide Transformation
• ตาราง Distribution ต่อ Partition
• ผล Cache แยก Materialization กับ Reuse
• เปรียบเทียบ reduceByKey/groupByKey พร้อม Shuffle Metric
• คำอธิบายหนึ่งกรณีที่เพิ่ม Partition แล้วไม่เร็วขึ้น

สรุปและขั้นตอนถัดไป: Lab นี้มองข้อมูลเป็น Record ที่ Spark ยังไม่เข้าใจความหมายคอลัมน์ Lab 2 จะยกระดับเป็น DataFrame/SQL ให้ Catalyst เห็น Schema และ Algebra แล้วเราจะตรวจว่า Optimization ใดเกิดจริงจาก explain() ไม่ใช่สรุปว่า SQL เร็วเพราะเขียนสั้นกว่า

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 Core และ RDD

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 Core และ RDD ตัดสินใจอย่างไรเมื่อข้อมูล งาน และทรัพยากรถูกกระจายออกจากกัน เมื่อทำเสร็จแล้วควรตอบได้สามเรื่อง: ระบบทำอะไร หลักฐานใดแสดงว่ามันทำเช่นนั้น และเงื่อนไขใดทำให้พฤติกรรมเปลี่ยน

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

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 Core และ RDD โดยห้ามเริ่มจากชื่อ Class หรือ API ให้เริ่มจากปัญหาที่ระบบกำลังแก้ ใช้ภาพหนึ่งภาพและตัวอย่างข้อมูลไม่เกินสิบแถว ผู้ฟังมีหน้าที่ถามว่า “รู้ได้อย่างไร” และ “ถ้าข้อมูลใหญ่ขึ้นจะเกิดอะไร” วิธีนี้ช่วยเปิดจุดที่เราจำคำสั่งได้แต่ยังเชื่อมเหตุผลไม่ครบ

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