Spark MLlib: สร้าง Text Classifier ตัวแรก

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

from pyspark.sql import SparkSession

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

training = spark.createDataFrame([
    (0, "win free money now", 1.0),
    (1, "click here to claim your prize", 1.0),
    (2, "congratulations you won a lottery", 1.0),
    (3, "free gift card waiting for you", 1.0),
    (4, "meeting scheduled for tomorrow afternoon", 0.0),
    (5, "please review the attached document", 0.0),
    (6, "let us know if you have questions", 0.0),
    (7, "quarterly report is ready for review", 0.0),
    (8, "urgent claim your reward immediately", 1.0),
    (9, "team lunch this friday at noon", 0.0),
], ["id", "text", "label"])

training.show(truncate=False)

2สร้าง ML Pipeline

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml.classification import LogisticRegression

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol="words", outputCol="features", numFeatures=1000)
lr = LogisticRegression(maxIter=10, regParam=0.01, labelCol="label", featuresCol="features")

pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])
model = pipeline.fit(training)

print("Train เสร็จแล้ว — pipeline มี", len(pipeline.getStages()), "ขั้นตอน")

3ทดสอบกับข้อความใหม่

test = spark.createDataFrame([
    (10, "you have won free money click now"),
    (11, "can we reschedule the meeting"),
    (12, "claim your urgent prize today"),
    (13, "attached is the report you requested"),
], ["id", "text"])

predictions = model.transform(test)
predictions.select("id", "text", "prediction", "probability").show(truncate=False)

4ประเมินผลอย่างถูกวิธี: Train/Test Split และ Metric

from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator

# สมมติมี dataset ใหญ่กว่านี้ — สาธิตวิธีแบ่งข้อมูลที่ถูกต้อง
train_df, test_df = training.randomSplit([0.8, 0.2], seed=42)
model = pipeline.fit(train_df)
result = model.transform(test_df)

evaluator_auc = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC")
evaluator_acc = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="accuracy")

print("AUC:", evaluator_auc.evaluate(result))
print("Accuracy:", evaluator_acc.evaluate(result))

5บันทึกและโหลดโมเดลกลับมาใช้

# บันทึกโมเดล
model.write().overwrite().save("/home/jovyan/data/spam-classifier-model")

# โหลดกลับมาใช้ในเซสชันใหม่ (หรือ notebook อื่น)
from pyspark.ml import PipelineModel
loaded_model = PipelineModel.load("/home/jovyan/data/spam-classifier-model")

loaded_model.transform(test).select("text", "prediction").show(truncate=False)

6ตรวจ Transformer, Estimator และ Model ทีละ Stage

อย่ามอง Pipeline เป็นกล่องเดียว ให้พิมพ์ Schema/ตัวอย่างหลังแต่ละ Stage เพื่อรู้ว่าคอลัมน์เปลี่ยนอย่างไร

tokenized = tokenizer.transform(training)
tokenized.select("text", "words").show(truncate=False)

featured = hashingTF.transform(tokenized)
featured.select("words", "features").show(truncate=False)

# LogisticRegression เป็น Estimator ต้อง fit ก่อน
lr_model = lr.fit(featured)
lr_model.transform(featured).select(
    "text", "rawPrediction", "probability", "prediction"
).show(truncate=False)

Tokenizer/HashingTF เป็น Transformer เพราะไม่ต้องเรียนสถิติชุดข้อมูล ส่วน LogisticRegression เป็น Estimator และคืน Model ซึ่งเป็น Transformer Pipeline.fit จึงเดินผ่าน Stage และ Fit เฉพาะ Estimator

7แก้ Data Leakage ด้วยการแบ่งก่อน Fit

ถ้าเพิ่ม IDF, StringIndexer, StandardScaler หรือ Feature Selector ขั้นเหล่านี้เรียนจากข้อมูล ห้าม Fit บนข้อมูลทั้งหมดก่อนแบ่ง Train/Test

train, test = data.randomSplit([0.8, 0.2], seed=42)
pipeline_model = pipeline.fit(train)
predictions = pipeline_model.transform(test)

# ผิดหลัก: fitted = pipeline.fit(data) แล้วค่อยแบ่ง prediction
# เพราะ preprocessing และ model เห็น distribution ของ test

8Dataset เล็กเกินไปทำให้ Metric แกว่ง

ข้อมูลตัวอย่างไม่กี่สิบแถวมีไว้เรียน API ไม่เหมาะสรุปคุณภาพโมเดล สุ่ม Split คนละ Seed อาจทำให้ Test มีเพียงคลาสเดียว ทดลองหลาย Seed แล้วบันทึกจำนวน Positive/Negative

from pyspark.sql.functions import count

for seed in [1, 2, 3, 42, 99]:
    tr, te = data.randomSplit([0.8, 0.2], seed=seed)
    print("seed", seed, "train", tr.count(), "test", te.count())
    te.groupBy("label").count().show()

งานส่งควรเพิ่มข้อมูลอย่างน้อยหลักร้อยและแยก Holdout ที่ไม่แตะระหว่างพัฒนา หากข้อความจากผู้ส่งคนเดียวอยู่ทั้ง Train/Test โมเดลอาจจำรูปแบบผู้ส่ง จึงพิจารณา Group Split

9Confusion Matrix และ Metric ที่อ่านได้

predictions.groupBy("label", "prediction").count() \
    .orderBy("label", "prediction").show()

tp = predictions.filter("label = 1 AND prediction = 1").count()
fp = predictions.filter("label = 0 AND prediction = 1").count()
fn = predictions.filter("label = 1 AND prediction = 0").count()

precision = tp / (tp + fp) if tp + fp else 0
recall = tp / (tp + fn) if tp + fn else 0
print("precision", precision, "recall", recall)

Spam Filter อาจเลือก Threshold ตาม Cost: False Positive ทำให้เมลสำคัญหาย ส่วน False Negative ทำให้ผู้ใช้รำคาญ ค่า 0.5 ไม่ใช่กฎศักดิ์สิทธิ์ ทดลอง Threshold หลายค่าและสร้างตาราง Precision/Recall

10Hash Collision และ Feature Space

HashingTF ไม่เก็บ Vocabulary แต่คำต่างกันชน Index เดียวได้ ทดลอง numFeatures 16, 128, 1024, 16384 แล้วดู Metric, Model Size และเวลา

เพิ่ม IDF เพื่อให้น้ำหนักคำหายาก แต่ IDF เป็น Estimator ต้อง Fit จาก Train เท่านั้น

from pyspark.ml.feature import IDF
idf = IDF(inputCol="rawFeatures", outputCol="features")

11Cross-validation และต้นทุนที่คูณ

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

grid = ParamGridBuilder() \
    .addGrid(hashingTF.numFeatures, [128, 1024, 4096]) \
    .addGrid(lr.regParam, [0.0, 0.01, 0.1]) \
    .build()

cv = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=grid,
    evaluator=evaluator_auc,
    numFolds=3,
    parallelism=2
)
cv_model = cv.fit(train)

3 × 3 Parameter Maps × 3 Folds = 27 Fits ยังไม่รวม Final Model การขนานลด Wall-clock แต่ไม่ลด Compute Cost ควรเริ่ม Baseline และ Search Space ที่มีเหตุผล

12Error Analysis: อ่านสิ่งที่โมเดลทายผิด

predictions.filter("label != prediction") \
  .select("text", "label", "prediction", "probability") \
  .show(50, truncate=False)

จัดกลุ่ม Error เช่นคำโฆษณาในข้อความปกติ, ภาษาไทย/อังกฤษ, URL, ข้อความสั้น, Spam แบบใหม่ แล้วตัดสินว่าจะเพิ่มข้อมูล ปรับ Tokenization หรือเปลี่ยน Model อย่า Tune ตัวเลขโดยไม่อ่านตัวอย่างผิด

13Training–Serving Skew และ Model Contract

โหลด PipelineModel ช่วยให้ Tokenizer/Hashing เหมือนตอนฝึก แต่ Production ยังต้องตรงเรื่อง Schema, Null, Encoding และข้อความยาว สร้าง Contract และ Test ตัวอย่างเดียวกันก่อน/หลัง Save–Load

before = model.transform(test).select("text", "prediction").orderBy("text")
loaded = PipelineModel.load("/home/jovyan/data/spam-classifier-model")
after = loaded.transform(test).select("text", "prediction").orderBy("text")

assert before.collect() == after.collect()

บันทึก Model Version, Data Version, Code/Library Version, Parameters, Metric และ Training Time ข้าง Model File ไม่ใช่เก็บเพียงโฟลเดอร์ชื่อ final_model_really_final

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

• Dataset ที่เพิ่มเองพร้อมคำอธิบาย Label
• Pipeline Diagram และ Schema หลังแต่ละ Stage
• Split Strategy และหลักฐานไม่มี Leakage
• Confusion Matrix, Precision, Recall, AUC/PR ตามที่เหมาะ
• ผล Hash Space/Regularization อย่างน้อย 6 Configurations
• Error Analysis อย่างน้อย 10 ตัวอย่าง
• Save–Load Reproducibility Test และ Model Card ย่อ

สรุปและขั้นตอนถัดไป: Lab นี้ฝึก Batch ML ที่มีจุดจบ Lab 4 จะเปลี่ยน Input เป็น Stream ที่ไม่มีวันจบ เราต้องเพิ่ม Event Time, Window, Watermark, State และ Checkpoint โดยยังรักษาหลักเดียวกันคืออย่าให้ผลลัพธ์ “รันได้” บังคำถามว่ามันนับอะไรและลืมเมื่อใด

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 MLlib

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

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

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

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