Spark SQL และ ML Pipeline

1RDD: รากฐานของทุกอย่างใน Spark

คำว่า immutable ไม่ได้แปลว่าเราทำความสะอาดหรือแก้ข้อมูลไม่ได้ เราทำได้โดยสร้างผลลัพธ์ชุดใหม่ เช่น clean = raw.filter(...).map(...) สิ่งที่ไม่เกิดขึ้นคือการแอบเปลี่ยนค่ากลางทางใน raw วิธีคิดนี้ช่วยลดการพึ่งพากันระหว่าง task เพราะแต่ละ task อ่าน input แล้วสร้าง output ของตัวเอง ไม่ต้องแย่งกันเขียนตัวแปรกลางและไม่ต้องถือ write lock บนข้อมูลชุดเดียวกัน

Partition: หน่วยที่ทำให้งานกระจายได้จริง

คำว่า “ข้อมูลกระจาย” จะยังเป็นเพียงภาพกว้างจนกว่าเราจะเห็น partition Spark ไม่ได้ส่ง record ทีละแถวไปหา CPU แบบสุ่ม แต่แบ่งข้อมูลเป็นก้อน แต่ละ task มักรับผิดชอบหนึ่ง partition ในหนึ่ง stage ดังนั้นจำนวน partition จึงสัมพันธ์กับจำนวนงานที่รันพร้อมกันได้ ขนาดข้อมูลต่อ task และ overhead จากการจัดตารางงาน

ดังนั้นคำถามที่ดีกว่า “ควรมี partition เท่าไร” คือ “แต่ละ partition ใหญ่แค่ไหน งานใน partition ใช้เวลาพอ ๆ กันหรือไม่ และจำนวน task ทำให้ทรัพยากรในคลัสเตอร์ทำงานต่อเนื่องหรือยัง” ไม่มีตัวเลขเดียวที่ใช้ได้กับทุกงาน เพราะการอ่าน Parquet การ Join ตารางใหญ่ และการประมวลผลข้อความมีต้นทุนต่อแถวต่างกัน

Lineage: สูตรสำหรับสร้างข้อมูลใหม่ ไม่ใช่สำเนาของข้อมูลทุกชุด

เมื่อเราเขียน raw.filter(...).map(...) Spark จำความสัมพันธ์ว่า RDD ปลายทางเกิดจาก RDD ใดและผ่าน transformation อะไร ถ้า executor เครื่องหนึ่งหาย Spark สามารถนำ partition ต้นทางที่เกี่ยวข้องมาคำนวณใหม่ได้ Lineage จึงคล้ายสูตรทำอาหารมากกว่ากล่องสำรองอาหาร เราไม่จำเป็นต้องถ่ายสำเนาผลลัพธ์ชั่วคราวทุกจาน แต่ต้องรู้ว่าถ้าจานหนึ่งหก จะหยิบวัตถุดิบใดมาทำใหม่

Transformation, Action และ Lazy Evaluation

RDD แบ่ง operation เป็นสองกลุ่มใหญ่ Transformation เช่น map, filter, flatMap และ join สร้าง RDD ใหม่และยังไม่จำเป็นต้องคำนวณทันที ส่วน Action เช่น count, take, collect และ saveAsTextFile ขอผลลัพธ์จริง จึงทำให้ Spark สร้าง job และเริ่มประมวลผล

val raw = sc.textFile("events.csv")          // ยังไม่อ่านไฟล์ทั้งหมด
val valid = raw.filter(isValid)               // บันทึกแผนไว้
val pairs = valid.map(toUserAndAmount)         // บันทึกแผนต่อ
val total = pairs.reduceByKey(_ + _)           // ยังไม่รันจนกว่าจะมี action

total.take(20)                                 // Action: เริ่มสร้าง job
total.saveAsTextFile("output")                // อีก Action: เป็นอีก job หนึ่ง

Lazy Evaluation เหมือนเราจดรายการธุระหลายอย่างไว้ก่อนออกจากบ้าน ถ้ารู้ทั้งหมดตั้งแต่ต้น เราอาจจัดเส้นทางให้แวะร้านที่อยู่ใกล้กัน ลดการย้อนกลับ และไม่ซื้อของที่ท้ายที่สุดไม่ได้ใช้ อย่างไรก็ตาม ความ “ขี้เกียจ” ไม่ได้แปลว่าระบบจะเก็บผลลัพธ์ให้เอง เมื่อ action สองตัวเรียกใช้สายการคำนวณเดียวกัน Spark อาจคำนวณซ้ำ เว้นแต่เราจะ cache หรือ persist ข้อมูลที่คุ้มแก่การใช้ซ้ำ

Narrow กับ Wide Transformation: จุดแบ่งที่สำคัญกว่าชื่อ operation

Job, Stage และ Task: สามคำที่ควรแยกให้ออก

• Job เกิดจาก Action หนึ่งครั้ง เช่น count() หรือการเขียนผลลัพธ์หนึ่งครั้ง
• Stage คือช่วงของงานที่ต่อกันได้โดยไม่ต้องรอ Shuffle ใหม่ ขอบเขตของ stage มักเกิดตรง wide dependency
• Task คือหน่วยงานที่ถูกส่งไปรันกับ partition หนึ่งใน stage นั้น

ลำดับจึงเป็น Action สร้าง Job, Job ถูกแบ่งเป็น Stage และแต่ละ Stage มี Task ตามจำนวน partition การอ่าน Spark UI ให้เริ่มจากโครงนี้ก่อน แล้วค่อยดูว่า stage ใดใช้เวลานาน มี Shuffle Read/Write สูง หรือมี task บางตัวช้ากว่ากลุ่มอย่างผิดปกติ

2จาก MapReduce สู่ SQL: เมื่อเราเลิกบอกทุกขั้นตอน

MapReduce เปิดทางให้ประมวลผลข้อมูลขนาดใหญ่ได้ แต่โปรแกรมเมอร์ต้องระบุเองว่าจะ Map อะไร ใช้ key ใด Shuffle ตอนไหน และ Reduce อย่างไร ต่อให้โจทย์เป็นเพียง “เลือกยอดขายของกรุงเทพฯ แล้วรวมตามสินค้า” ก็ยังต้องเขียนเส้นทางการคำนวณค่อนข้างละเอียด

SQL เปลี่ยนบทบาทของเรา จากคนบอกเส้นทางทุกแยก มาเป็นคนบอกจุดหมาย เราเขียนว่า ต้องการผลลัพธ์อะไร ส่วนระบบเลือกเองว่า ควรคำนวณอย่างไร จุดเปลี่ยนสำคัญจึงไม่ใช่เพียงไวยากรณ์ที่สั้นลง แต่คือการแยก logical plan ออกจาก physical plan

SELECT product, SUM(amount)
FROM sales
WHERE city = 'Bangkok'
GROUP BY product;

Relational Algebra: ภาษากลางระหว่าง SQL กับการทำงานจริง

ก่อนที่ SQL จะกลายเป็น task บนคลัสเตอร์ ระบบจะแปลงคำสั่งให้เป็นชุด operator ทาง Relational Algebra แต่ละ operator รับตารางหนึ่งหรือหลายตาราง แล้วคืนตารางใหม่ จึงต่อกันเป็นสายการคำนวณได้

เส้นทางจาก Hadoop มาถึง Spark SQL

Declarative ไม่ได้แปลว่าเราไม่ต้องเข้าใจระบบ

SQL เป็นภาษาเชิงประกาศ เราบอกว่าต้องการแถวและคอลัมน์ใด แต่ไม่ได้กำหนดทุกขั้นตอนว่าจะอ่านไฟล์ไหนก่อนหรือใช้ Join แบบใด ข้อดีคือ optimizer มีอิสระเลือกแผนตามสถิติ ขนาดข้อมูล และความสามารถของแหล่งข้อมูล แต่ข้อเสียคือถ้าเราไม่อ่านแผนเลย เราอาจเขียน query ที่ถูกต้องทางตรรกะ แต่มีต้นทุนสูงมากทางกายภาพ

กฎของ Algebra ที่เปิดทางให้ optimizer จัดรูปใหม่

ประโยชน์ของ Relational Algebra ไม่ได้อยู่ที่การท่องสัญลักษณ์ แต่อยู่ที่การรู้ว่า operation ใดสลับหรือรวมกันได้โดยยังคงความหมายเดิม ตัวอย่างเช่น ถ้าเงื่อนไขกรองใช้เฉพาะคอลัมน์ของตารางลูกค้า ระบบอาจดันเงื่อนไขนั้นไปก่อน Join เพื่อลดจำนวนแถวที่จะข้ามเครือข่าย

-- เขียนแบบอ่านง่าย
SELECT c.segment, SUM(o.amount)
FROM customers c JOIN orders o ON c.customer_id = o.customer_id
WHERE c.country = 'TH' AND o.order_date >= DATE '2026-01-01'
GROUP BY c.segment;

-- แนวคิดของ optimizer
TH_customers = FILTER customers WHERE country = 'TH'
recent_orders = FILTER orders WHERE order_date >= DATE '2026-01-01'
joined        = JOIN TH_customers WITH recent_orders
result        = GROUP joined BY segment AND SUM(amount)

แผนหลังไม่ได้เปลี่ยนคำตอบ แต่ลดข้อมูลทั้งสองฝั่งก่อน Join นี่คือเหตุผลว่าทำไม “กรองให้เร็ว ตัดคอลัมน์ให้เร็ว” จึงเป็นหลักที่เห็นซ้ำบ่อย อย่างไรก็ตาม optimizer จะทำได้ดีเมื่อ expression อยู่ในภาษาที่ระบบเข้าใจ ถ้าเราซ่อนตรรกะทั้งหมดไว้ใน UDF ระบบอาจมองไม่เห็นเงื่อนไขภายในและพลาดโอกาสปรับแผน

Logical Plan กับ Physical Plan ต่างกันอย่างไร

ความผิดพลาดแต่ละชั้นจึงมีหน้าตาต่างกัน ถ้าพิมพ์ชื่อคอลัมน์ผิด ปัญหาเกิดตอน analysis ถ้า query ถูกต้องแต่ช้า เราต้องดู optimized และ physical plan ถ้าแผนดูสมเหตุสมผลแต่ task บางตัวช้ามาก ปัญหาอาจอยู่ที่ distribution ของข้อมูลหรือทรัพยากรตอน runtime

3Spark SQL และ DataFrame

DataFrame API กับ SQL ใช้ Catalyst และ execution engine ชุดเดียวกัน จึงมักสร้างแผนที่เหมือนหรือใกล้เคียงกันเมื่อบอกความหมายเดียวกัน แต่ไม่ควรสรุปว่า “เร็วเท่ากันเสมอ” เพราะ UDF, expression, hint, configuration และรายละเอียดของ query อาจทำให้แผนต่างกัน วิธีตรวจที่เชื่อถือได้คือดู explain() และวัดงานจริง ไม่ใช่เดาจากรูปแบบไวยากรณ์

// อ่านข้อมูลแบบ Python (PySpark)
df = spark.read \
     .format("json") \
     .option("samplingRatio", "0.1") \
     .load("/data/stuff.json")

df.write \
     .format("parquet") \
     .mode("append") \
     .partitionBy("year") \
     .saveAsTable("faster_stuff")

// เขียนแบบ Scala — เทียบเท่ากัน
val df = spark.read.format("json").load("/data/stuff.json")
val young = df.filter($"age" < 30)
young.createOrReplaceTempView("young")
spark.sql("SELECT count(*) FROM young").show()

Transformation กับ Action ใน DataFrame

Schema ไม่ได้มีไว้เพียงบอกชื่อคอลัมน์

Schema บอกชนิดข้อมูล โครงสร้างซ้อน และ nullability ทำให้ Spark ตรวจ expression ได้ก่อนรัน เลือก representation ที่เหมาะสม และอ่านเฉพาะคอลัมน์ที่ต้องการจาก columnar format ได้ แต่ schema ไม่ได้บอกความหมายทางธุรกิจทั้งหมด เช่น คอลัมน์ age เป็น integer ไม่ได้แปลว่าค่า -20 ถูกต้อง หรือคอลัมน์ country เป็น string ไม่ได้แปลว่ารหัสประเทศทุกค่าเป็นมาตรฐานเดียวกัน เราจึงยังต้องมี validation และ data quality rule แยกต่างหาก

กำหนด Schema เอง หรือให้ระบบเดา

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

ตัวอย่างการทำงานกับ Schema

// ให้ Spark เดา schema จาก case class
case class Person(firstName: String, lastName: String, age: Int)
val rdd = sc.textFile("people.csv")
val peopleRDD = rdd.map { line =>
  val cols = line.split(",")
  Person(cols(0), cols(1), cols(2).toInt)
}
val df = peopleRDD.toDF

// หรือกำหนด schema เอง (แม่นยำกว่า infer อัตโนมัติ)
val schema = StructType(
  StructField("firstName", StringType, nullable = false) ::
  StructField("lastName", StringType, nullable = false) ::
  StructField("age", IntegerType, nullable = true) :: Nil
)
val people = spark.read
  .option("header", "true")
  .schema(schema)
  .csv("people.csv")

people.printSchema()
// root
// |-- firstName: string (nullable = true)
// |-- lastName: string (nullable = true)
// |-- age: integer (nullable = true)

Dataset, DataFrame และ RDD ควรเลือกอย่างไร

แนวทางทั่วไปคือเริ่มจาก DataFrame/SQL เพราะระบบเห็นความหมายและช่วย optimize ได้มากกว่า แล้วลงไปใช้ RDD เฉพาะส่วนที่ abstraction เชิงตารางอธิบายไม่ได้ การเลือก abstraction ระดับสูงไม่ใช่การเสียความเป็นวิศวกร แต่คือการส่งข้อมูลความหมายให้ optimizer ช่วยตัดสินใจ

4อ่านและเขียนข้อมูล: จุดที่ประสิทธิภาพเริ่มต้นจริง

หลายงานช้าไม่ใช่เพราะ SQL ซับซ้อน แต่เพราะอ่านข้อมูลมากเกินจำเป็น เลือก file format ไม่เหมาะ หรือมีไฟล์เล็กนับแสนไฟล์ ก่อนปรับจำนวน executor เราควรถามก่อนว่า Spark ต้องอ่านกี่ byte ใช้กี่คอลัมน์ กรอง partition ได้หรือไม่ และจำนวนไฟล์สร้าง overhead มากเพียงใด

CSV/JSON กับ Parquet/ORC

สี่วิธีลดข้อมูลก่อนมันไหลเข้าคลัสเตอร์

1. Column pruning: เลือกเฉพาะคอลัมน์ที่ใช้ ระบบอ่านข้อมูลน้อยลงและทำให้ record ระหว่างทางเล็กลง
2. Predicate pushdown: ส่งเงื่อนไขกรองลงไปยัง data source หรือ file reader เพื่อข้ามกลุ่มข้อมูลที่ไม่เกี่ยวข้อง
3. Partition pruning: ถ้าข้อมูลแบ่ง directory ตามวันที่ การกรองวันที่ที่ตรงรูปแบบช่วยข้าม directory อื่นได้
4. Data skipping: ใช้สถิติ เช่น min/max ของก้อนข้อมูล เพื่อไม่อ่านก้อนที่เป็นไปไม่ได้ว่าจะมีค่าตรงเงื่อนไข

val sales = spark.read.parquet("/lake/sales")

val result = sales
  .select("sale_date", "product_id", "amount")
  .filter($"sale_date" >= "2026-07-01" && $"sale_date" < "2026-08-01")
  .groupBy("product_id")
  .agg(sum("amount").as("revenue"))

result.explain("formatted")

ควรตรวจแผนว่ามี PartitionFilters, PushedFilters และ ReadSchema ตามที่คาดหรือไม่ ถ้าเราเขียน function ครอบคอลัมน์วันที่จน data source มองเงื่อนไขไม่ออก ระบบอาจต้องเปิดทุก partition แม้สุดท้ายใช้ข้อมูลเพียงเดือนเดียว

Partitioning ใน Storage ไม่ใช่ Partition ของ Spark เสมอไป

คำว่า partition ใช้สองบริบท storage partition คือการจัด directory หรือ table ตามค่า เช่น year=2026/month=07 เพื่อช่วยค้นหาและตัดข้อมูล ส่วน Spark partition คือก้อนงานในหน่วยความจำหรือระหว่าง execution ทั้งสองสัมพันธ์กันตอนอ่าน แต่ไม่ใช่สิ่งเดียวกันตลอด pipeline

ปัญหา Small Files

ไฟล์ขนาดเล็กจำนวนมากทำให้ driver ต้องจัดการรายการไฟล์จำนวนมาก executor ต้องเปิดและปิดไฟล์ซ้ำ และ object store มี request overhead สูง แม้จำนวน byte รวมไม่มาก งาน streaming หรือการเขียนแบบถี่อาจสร้างปัญหานี้โดยไม่รู้ตัว วิธีแก้มีทั้งการ compact ไฟล์เป็นระยะ การควบคุมจำนวน output partition และการเลือก table format ที่รองรับ maintenance แต่ต้องระวังไม่รวมไฟล์จนใหญ่เกินกว่าจะกระจายงานได้ดี

Repartition กับ Coalesce

อย่าใส่ repartition(1) เพียงเพราะอยากได้ไฟล์เดียวโดยไม่คิดถึงขนาดข้อมูล เพราะงานเขียนช่วงสุดท้ายจะบีบทุกอย่างผ่าน task เดียว เหมือนสร้างทางด่วนสิบช่องแล้วบังคับให้รถทั้งหมดออกด่านเดียว

5Join, Shuffle และ Data Skew

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

เหตุใด Join จึงมักต้อง Shuffle

สมมุติ orders กระจายตาม order_id ส่วน customers กระจายแบบอื่น แต่เราต้อง Join ด้วย customer_id ข้อมูลลูกค้าหมายเลข 42 ที่อยู่คนละเครื่องต้องถูกส่งไปยัง partition เดียวกันก่อน งานนี้คล้ายเอาบัตรจากหลายสาขามาจัดตู้ใหม่ตามเลขสมาชิก ค่าใช้จ่ายไม่ได้อยู่ที่การเทียบเลขเพียงอย่างเดียว แต่อยู่ที่การขนและจัดบัตรใหม่ทั้งระบบ

กลยุทธ์ Join ที่ควรรู้

val customers = spark.read.parquet("/lake/customers")
val orders = spark.read.parquet("/lake/orders")

// ถ้าตาราง customers เล็กจริง เราอาจให้ hint ได้
val enriched = orders.join(
  broadcast(customers.select("customer_id", "segment")),
  Seq("customer_id"),
  "left"
)

enriched.explain("formatted")

Data Skew: เมื่อค่าเฉลี่ยหลอกเรา

สมมุติข้อมูล 1 TB แบ่งเป็น 1,000 partition ค่าเฉลี่ยคือ partition ละ 1 GB ดูเหมือนสมดุล แต่ถ้า key UNKNOWN กินข้อมูล 300 GB อยู่ partition เดียว task นั้นจะทำงานนานมาก ขณะที่ task อื่นเสร็จและเครื่องส่วนใหญ่รออยู่ Job จึงช้าตาม task ที่ช้าที่สุด ไม่ใช่ตามค่าเฉลี่ย

อาการที่พบได้คือ stage เกือบเสร็จแต่ค้างอยู่ไม่กี่ task, Shuffle Read ของ task บางตัวสูงกว่ากลุ่มมาก, spill ลงดิสก์มาก หรือ executor หลุดเพราะ memory ไม่พอ สิ่งแรกที่ควรทำไม่ใช่เพิ่มเครื่องทันที แต่ตรวจ distribution ของ join key และ group key

วิธีรับมือกับ Skew

• แยก key พิเศษออก: ค่า null หรือ UNKNOWN อาจไม่มีความหมายในการ Join จึงกรองหรือประมวลผลแยกได้
• Broadcast ฝั่งเล็ก: หลีกเลี่ยงการ Shuffle ตารางใหญ่ทั้งสองฝั่งเมื่อเงื่อนไขเหมาะสม
• Salting: เติมเลขย่อยให้ hot key กระจายไปหลาย partition แล้วรวมผลภายหลัง
• Pre-aggregation: รวมข้อมูลบางส่วนก่อน Shuffle เช่นใช้ reduceByKey แทนการส่งทุก record แบบ groupByKey
• Adaptive Query Execution: เปิดโอกาสให้ Spark ใช้สถิติระหว่างรันเพื่อจัดการ skewed partition หรือปรับแผนบางส่วน

// แนวคิด Salting สำหรับ hot key
val saltedOrders = orders
  .withColumn("salt", pmod(hash($"order_id"), lit(16)))

val expandedCustomers = customers
  .withColumn("salt", explode(sequence(lit(0), lit(15))))

val joined = saltedOrders.join(
  expandedCustomers,
  Seq("customer_id", "salt")
)

Salting แลกการลดก้อนใหญ่หนึ่งก้อนกับการทำสำเนาหรือรวมผลเพิ่ม จึงควรใช้กับ key ที่มีปัญหาจริงและเลือกจำนวน salt จากขนาดข้อมูล ไม่ใช่ใส่ทุก Join โดยอัตโนมัติ

GroupByKey กับการรวมบางส่วน

ถ้าเป้าหมายคือหาผลรวม เราไม่จำเป็นต้องส่งค่าทุกตัวข้ามเครือข่ายแล้วเก็บเป็นรายการยาว reduceByKey หรือ aggregation ที่ Spark รู้จักสามารถรวมบางส่วนในฝั่ง map ก่อน Shuffle ได้ หลักการเหมือนให้แต่ละสาขารวมยอดของตนก่อนส่งยอดสรุปเข้าศูนย์กลาง แทนที่จะขนใบเสร็จทุกใบเข้ากรุงเทพฯ

6Catalyst, AQE และการอ่านแผนการทำงาน

Catalyst เป็นกรอบสำหรับวิเคราะห์และปรับแผนเชิงตาราง ส่วน execution engine แปลงแผนไปเป็นงานที่ทำกับข้อมูลจริง เราไม่จำเป็นต้องจำ rule ทุกตัว แต่ควรอ่านเส้นทางจาก query ไปสู่ physical operator และรู้ว่าคำสำคัญใดบอกต้นทุน

สิ่งที่ควรมองหาใน explain()

df.explain()                  // physical plan แบบย่อ
df.explain("extended")       // parsed, analyzed, optimized และ physical plan
df.explain("formatted")      // อ่าน operator และรายละเอียดเป็นส่วน ๆ
df.explain("cost")           // ดูสถิติ/ค่าประมาณเมื่อระบบมีข้อมูลเพียงพอ

Adaptive Query Execution: ปรับแผนเมื่อเห็นข้อมูลจริง

แผนก่อนรันอาศัยสถิติและค่าประมาณ ซึ่งอาจไม่ตรงกับข้อมูลจริง AQE ใช้ข้อมูลที่เกิดหลัง Shuffle เพื่อปรับบางส่วนของ physical plan ระหว่างรัน เช่น รวม shuffle partition ที่เล็กเกินไป เปลี่ยนกลยุทธ์ Join เมื่อพบว่าฝั่งหนึ่งเล็กจริง หรือแบ่ง skewed partition ที่ใหญ่ผิดปกติ

Built-in Expression กับ UDF

Built-in function เช่น lower, substring, when, regexp_extract และ SQL expression เป็นภาษาที่ Catalyst เข้าใจ ระบบจึงตรวจชนิดข้อมูล จัดรูป expression และสร้างโค้ดประมวลผลได้ดี UDF มีประโยชน์เมื่อไม่มี operation ที่ต้องการ แต่ทำให้ optimizer เห็นเพียงกล่องดำ และใน PySpark ยังอาจมีต้นทุนการส่งข้อมูลข้ามขอบเขต JVM/Python เพิ่มขึ้น

// ควรเริ่มจาก built-in expression
val cleaned = df.withColumn(
  "normalized_email",
  lower(trim($"email"))
)

// ใช้ UDF เมื่อความหมายเฉพาะจริง ๆ และวัดต้นทุนแล้ว
val custom = udf((x: String) => domainSpecificTransform(x))

หลักนี้ไม่ได้หมายความว่า “ห้ามใช้ UDF” แต่ให้ถามก่อนว่า built-in expression, higher-order function สำหรับ array/map หรือ vectorized UDF ตอบโจทย์ได้หรือไม่ และหลังเลือกแล้วควรวัดด้วยข้อมูลที่ใกล้เคียงของจริง

Cache, Persist และ Checkpoint

ข้อมูลที่คุ้มแก่การ cache คือผลลัพธ์ที่สร้างแพงและถูกใช้ซ้ำหลายครั้ง เช่น feature table เดียวกันที่ใช้ฝึกหลายโมเดล แต่ถ้า DataFrame ถูกใช้ครั้งเดียว การ cache เพิ่มงานเขียนและกินพื้นที่โดยไม่คืนประโยชน์ ควร materialize ด้วย action ที่เหมาะสมแล้วดู Storage tab ว่าเก็บได้จริงเท่าไร

7SQL ที่ควรรู้ให้ลึกกว่าการเขียน SELECT

คนใช้ Spark มักรู้ไวยากรณ์ SQL อยู่แล้ว แต่เมื่อนำไปใช้กับข้อมูลกระจาย เราต้องสนใจ semantics และต้นทุนมากขึ้น โดยเฉพาะ NULL, aggregation, window function และความแตกต่างระหว่างการเรียงในกลุ่มกับการเรียงทั้งระบบ

NULL และตรรกะสามค่า

ใน SQL ค่า NULL หมายถึงไม่ทราบหรือไม่มีค่า การเปรียบเทียบ amount = NULL ไม่ได้คืน true แต่เป็น unknown จึงต้องใช้ IS NULL หรือ IS NOT NULL ความจริงของเงื่อนไขจึงมีสามสถานะคือ true, false และ unknown ไม่ใช่ Boolean สองค่าแบบที่หลายคนคุ้นจากภาษาโปรแกรม

-- ไม่ถูกตาม semantics ที่ต้องการ
SELECT * FROM sales WHERE amount = NULL;

-- ถูกต้อง
SELECT * FROM sales WHERE amount IS NULL;

-- เลือกค่าแทนเมื่อ NULL
SELECT COALESCE(amount, 0) AS amount_or_zero FROM sales;

Aggregation และการรวมแบบกระจาย

Aggregation ที่ดีสามารถรวมบางส่วนในแต่ละ partition แล้วส่งผลสรุปข้ามเครือข่าย ตัวอย่างเช่น การหาผลรวมใช้สถานะเพียงหนึ่งค่า แต่การหา median แบบ exact อาจต้องข้อมูลมากกว่ามาก ระบบ Big Data จึงมี approximate aggregation เช่น approximate percentile หรือ approximate distinct count เพื่อแลกความคลาดเคลื่อนเล็กน้อยกับเวลาและหน่วยความจำที่ลดลง

Window Function: มองเพื่อนบ้านโดยไม่ยุบแถว

GROUP BY ยุบหลายแถวเหลือหนึ่งแถวต่อกลุ่ม แต่ Window Function คำนวณข้ามแถวในกลุ่มแล้วคงจำนวนแถวเดิม จึงเหมาะกับ ranking, running total, moving average และการเทียบค่ากับแถวก่อนหน้า

SELECT
  customer_id,
  order_date,
  amount,
  SUM(amount) OVER (
    PARTITION BY customer_id
    ORDER BY order_date
    ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  ) AS cumulative_amount,
  LAG(amount) OVER (
    PARTITION BY customer_id ORDER BY order_date
  ) AS previous_amount
FROM orders;

คำว่า PARTITION BY ใน window หมายถึงการแบ่งกลุ่มทางตรรกะของการคำนวณ ไม่ควรสับสนกับ storage partition หรือ Spark execution partition ระบบอาจต้อง Shuffle และ Sort ตาม customer เพื่อให้ลำดับถูกต้อง ดังนั้น window หลายชุดที่ใช้ partition/order ต่างกันอาจสร้าง Shuffle หลายรอบ

ORDER BY กับ SORT WITHIN PARTITIONS

ORDER BY ต้องให้ผลลัพธ์มีลำดับรวมทั้งชุด ซึ่งเป็นข้อกำหนดแพงในระบบกระจาย ส่วนการเรียงภายใน partition รับประกันเฉพาะลำดับของแต่ละก้อน เหมาะกับงานเขียนหรือ algorithm บางชนิดที่ไม่ต้องการลำดับทั่วโลก ถ้า dashboard ต้องการเพียง 100 รายการสูงสุด ใช้ ORDER BY ... LIMIT 100 และตรวจแผนย่อมดีกว่าเรียงข้อมูลทั้งหมดแล้วค่อยตัดด้วยโค้ดภายนอก

8Spark ML Pipeline: จากตารางไปสู่โมเดลที่ทำซ้ำได้

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.classification.LogisticRegression

val tokenizer = new Tokenizer()
  .setInputCol("text").setOutputCol("words")

val hashingTF = new HashingTF()
  .setNumFeatures(1000)
  .setInputCol(tokenizer.getOutputCol)
  .setOutputCol("features")

val lr = new LogisticRegression()
  .setMaxIter(10).setRegParam(0.01)

val pipeline = new Pipeline()
  .setStages(Array(tokenizer, hashingTF, lr))

val model = pipeline.fit(trainingData)
val predictions = model.transform(testData)

Transformer กับ Estimator ต่างกันตรง “ต้องเรียนจากข้อมูลหรือไม่”

วิธีจำที่แม่นกว่าการจำชื่อ class คือถามว่า operation นั้นต้องดูข้อมูลทั้งชุดเพื่อหาค่าพารามิเตอร์หรือไม่ Tokenizer ใช้กฎที่กำหนดไว้แปลงข้อความ จึงเป็น Transformer ได้ทันที แต่ StandardScaler ต้องหาค่าเฉลี่ยและส่วนเบี่ยงเบนมาตรฐานจากชุดฝึกก่อน จึงเป็น Estimator เมื่อ fit() แล้วจึงได้ StandardScalerModel ซึ่งเป็น Transformer

Data Leakage: Pipeline ช่วยจัดขั้นตอน แต่ไม่ป้องกันให้เอง

Data leakage เกิดเมื่อข้อมูลจากชุดทดสอบหรืออนาคตรั่วเข้าไปในขั้นตอนฝึก ตัวอย่างเช่น เราหาค่าเฉลี่ยและส่วนเบี่ยงเบนจากข้อมูลทั้งหมดก่อนแบ่ง train/test แม้โมเดลยังไม่เห็น label ของ test โดยตรง แต่ feature transformation ได้เรียน distribution จาก test แล้ว คะแนนจึงดูดีกว่าความสามารถจริง

// วิธีที่ถูกหลัก: แบ่งก่อน แล้ว fit ทุก Estimator บน training เท่านั้น
val Array(training, test) = data.randomSplit(Array(0.8, 0.2), seed = 42)

val pipelineModel = pipeline.fit(training)
val trainPredictions = pipelineModel.transform(training)
val testPredictions  = pipelineModel.transform(test)

// ไม่ควร fit preprocessing บน data ทั้งหมดแล้วค่อยแบ่ง
// เพราะ statistics และ category mapping ของ test จะรั่วเข้าชุดฝึก

Pipeline ทำให้ stage อยู่ด้วยกันและนำ transformation เดิมไปใช้กับ test ได้สะดวก แต่ถ้าเราเรียก fit() ผิดชุด Pipeline ก็ทำผิดอย่างเป็นระเบียบเท่านั้น เครื่องมือช่วยรักษากระบวนการ ไม่ได้แทนความเข้าใจเรื่องการทดลอง

การแบ่งข้อมูลต้องสอดคล้องกับโลกจริง

randomSplit เหมาะเมื่อแถวเป็นอิสระพอสมควรและโจทย์ในอนาคตมี distribution คล้ายข้อมูลปัจจุบัน แต่หลายกรณีต้องแบ่งแบบอื่น

• Time split: ฝึกด้วยอดีตและทดสอบด้วยอนาคต เพื่อเลียนแบบ deployment จริง
• Group split: ผู้ใช้หรือคนไข้คนเดียวกันไม่ควรปรากฏทั้ง train และ test เพราะโมเดลอาจจำเอกลักษณ์บุคคล
• Stratified split: รักษาสัดส่วน class โดยเฉพาะเมื่อ positive class หายาก
• Geographic split: ทดสอบความสามารถในการใช้กับพื้นที่ใหม่ ไม่ใช่เพียงแถวสุ่มจากพื้นที่เดิม

9Feature Engineering บนข้อมูลขนาดใหญ่

โมเดลใน Spark ML รับคอลัมน์ features ซึ่งมักเป็น vector และคอลัมน์ label การเดินทางจากตารางดิบไปถึง vector จึงเป็นส่วนสำคัญของระบบ ไม่ใช่งานเตรียมข้อมูลเล็ก ๆ ที่ทำครั้งเดียวแล้วลืม เพราะ mapping, vocabulary, scaling และการจัดการ missing value ต้องเหมือนกันตอนฝึกและตอนใช้งานจริง

ข้อมูลหมวดหมู่: StringIndexer และ OneHotEncoder

StringIndexer เรียน mapping จากข้อความเป็นดัชนี เช่น gold → 0, silver → 1 จากความถี่ในชุดฝึก ส่วน OneHotEncoder แปลงดัชนีเป็นเวกเตอร์เพื่อไม่ให้โมเดลเชิงเส้นตีความว่า gold มากกว่า silver ทางตัวเลข

val indexer = new StringIndexer()
  .setInputCol("segment")
  .setOutputCol("segment_index")
  .setHandleInvalid("keep")

val encoder = new OneHotEncoder()
  .setInputCol("segment_index")
  .setOutputCol("segment_vec")

setHandleInvalid("keep") ช่วยรองรับ category ที่ไม่เคยเห็นตอนฝึกโดยจัดเข้ากลุ่มพิเศษ แต่เรายังต้องเฝ้าดูอัตรา unknown ถ้าสูงขึ้นมากอาจแปลว่าข้อมูลเปลี่ยนหรือ mapping ต้นทางมีปัญหา การ “ไม่ล้ม” ไม่ได้แปลว่าคุณภาพโมเดลยังดี

ข้อความ: Tokenizer, HashingTF และ IDF

HashingTF แปลง token ไปยังตำแหน่งใน vector ด้วย hash จึงไม่ต้องเก็บพจนานุกรมขนาดใหญ่ แต่ token ต่างกันอาจชนตำแหน่งเดียวกัน การเพิ่มจำนวน feature ลดโอกาสชนแต่เพิ่มขนาด vector ส่วน IDF เป็น Estimator เพราะต้องเรียนว่า token ใดพบทั่วไปหรือหายากใน corpus ชุดฝึก

val tokenizer = new RegexTokenizer()
  .setInputCol("text")
  .setOutputCol("tokens")
  .setPattern("\\W+")

val hashingTF = new HashingTF()
  .setInputCol("tokens")
  .setOutputCol("tf")
  .setNumFeatures(1 << 18)

val idf = new IDF()
  .setInputCol("tf")
  .setOutputCol("tfidf")

ตัวเลข: Imputation, Scaling และความหมายของศูนย์

ค่า missing ต้องจัดการตามความหมาย บางกรณีใช้ median พร้อมเพิ่ม flag ว่าเดิม missing บางกรณีต้องแยก record ออกเพราะไม่มีทางตีความ การเติมศูนย์ทุกคอลัมน์สะดวกแต่เสี่ยงทำให้ “ไม่มีข้อมูล” กลายเป็น “ค่าจริงเท่ากับศูนย์”

โมเดลที่อาศัยระยะทางหรือ gradient เช่น Logistic Regression มักได้ประโยชน์จากการ scale feature ให้มีช่วงใกล้กัน ส่วน tree-based model มักไม่จำเป็นต้อง scale เพราะ split ตามลำดับค่า อย่างไรก็ดี scaler ต้อง fit บน training set เท่านั้นเพื่อหลีกเลี่ยง leakage

VectorAssembler: จุดรวม feature

val assembler = new VectorAssembler()
  .setInputCols(Array(
    "age_scaled",
    "income_scaled",
    "segment_vec",
    "tfidf"
  ))
  .setOutputCol("features")

val lr = new LogisticRegression()
  .setFeaturesCol("features")
  .setLabelCol("label")

ลำดับคอลัมน์ใน vector เป็นส่วนหนึ่งของสัญญาโมเดล ถ้า training ใช้ลำดับหนึ่ง แต่ production ประกอบ vector คนละลำดับ โมเดลยังรันได้แต่ตีความผิดอย่างเงียบ ๆ การบันทึก fitted PipelineModel จึงช่วยเก็บลำดับและพารามิเตอร์ transformation ไว้ด้วยกัน

Feature Engineering คือการคำนวณแบบกระจายเช่นกัน

การสร้าง feature จากประวัติผู้ใช้ 90 วันอาจต้อง Join, Window และ Aggregate หลาย TB ก่อนถึงโมเดล ดังนั้นความรู้เรื่อง partition, Shuffle และ skew ยังใช้เต็มที่ บ่อยครั้งต้นทุนหลักของ ML ไม่ได้อยู่ที่ fit() แต่อยู่ที่การสร้าง feature table ซ้ำหลายรอบ วิธีที่ดีคือแยกจุดที่ deterministic และใช้ซ้ำได้ วาง partition ตามรูปแบบการอ่าน เก็บ metadata และกำหนดเวลาตัดข้อมูลให้ชัด

10ประเมินโมเดลและปรับพารามิเตอร์อย่างไม่หลอกตัวเอง

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

Accuracy อาจดีมากทั้งที่โมเดลไร้ประโยชน์

ถ้าธุรกรรมฉ้อโกงมี 0.1% โมเดลที่ทายว่า “ปกติ” ทุกแถวได้ accuracy 99.9% แต่จับการฉ้อโกงไม่ได้เลย เราจึงต้องแยก Precision, Recall, F1, ROC-AUC และ PR-AUC ตามโจทย์

CrossValidator กับ TrainValidationSplit

CrossValidator แบ่ง training data เป็นหลาย fold แล้วฝึกซ้ำเพื่อประเมินชุดพารามิเตอร์อย่างเสถียรกว่าการแบ่งครั้งเดียว แต่ต้นทุนคร่าว ๆ โตตามจำนวน parameter map คูณจำนวน fold ถ้ามี 30 ชุดและ 5 folds อาจต้องฝึก 150 ครั้งก่อน refit โมเดลที่เลือก

val paramGrid = new ParamGridBuilder()
  .addGrid(lr.regParam, Array(0.001, 0.01, 0.1))
  .addGrid(lr.elasticNetParam, Array(0.0, 0.5, 1.0))
  .build()

val evaluator = new BinaryClassificationEvaluator()
  .setLabelCol("label")
  .setMetricName("areaUnderPR")

val cv = new CrossValidator()
  .setEstimator(pipeline)
  .setEstimatorParamMaps(paramGrid)
  .setEvaluator(evaluator)
  .setNumFolds(3)
  .setParallelism(4)

val cvModel = cv.fit(training)
val finalScore = evaluator.evaluate(cvModel.transform(test))

Threshold เป็นการตัดสินใจ ไม่ใช่ค่าศักดิ์สิทธิ์ 0.5

โมเดล classification มักให้ score หรือ probability แล้วเราค่อยกำหนด threshold การเลือก threshold ควรอิงต้นทุน false positive/false negative ความสามารถของทีมปฏิบัติการ และข้อจำกัด เช่นตรวจสอบได้วันละ 1,000 รายการ โมเดลเดียวกันจึงมีผลทางธุรกิจต่างกันมากตาม threshold

Baseline ก่อนโมเดลซับซ้อน

ควรเริ่มจากกฎง่าย ๆ ค่าเฉลี่ยตามกลุ่ม หรือ Logistic Regression ก่อน เพื่อรู้ว่าความซับซ้อนเพิ่มคุณค่าจริงเพียงใด ถ้าโมเดลใหญ่ขึ้น 20 เท่าแต่ metric ดีขึ้นเล็กน้อยและ latency สูงขึ้นมาก ระบบโดยรวมอาจแย่กว่า การวัดเวลาสร้าง feature, เวลา fit, เวลา transform, ขนาดโมเดล และต้นทุน inference จึงควรอยู่ข้าง metric ทางสถิติ

11จาก Notebook ไปสู่ระบบที่ใช้งานจริง

งานใน notebook มักจบที่ตารางคะแนน แต่ระบบจริงต้องตอบต่อว่าโมเดลและ preprocessing เวอร์ชันใดใช้กับข้อมูล snapshot ใด จะโหลดกลับอย่างไร ใครอนุมัติ และจะรู้ได้อย่างไรว่าคุณภาพตกลงหลังใช้งาน

บันทึกทั้งกระบวนการ ไม่ใช่เฉพาะตัวโมเดล

// บันทึก fitted pipeline ซึ่งรวม preprocessing และ model
cvModel.bestModel
  .asInstanceOf[PipelineModel]
  .write
  .overwrite()
  .save("/models/churn/2026-08-02")

val loaded = PipelineModel.load("/models/churn/2026-08-02")
val predictions = loaded.transform(newData)

นอกจากไฟล์โมเดล ควรบันทึก code version, Spark version, parameter, schema, ช่วงเวลาของ training data, metric, random seed และเจ้าของโมเดล ข้อมูลเหล่านี้ช่วยตอบคำถามเมื่อผลลัพธ์เปลี่ยน ไม่เช่นนั้นเรามีโมเดลหนึ่งไฟล์แต่ไม่มีเรื่องราวว่ามันเกิดขึ้นอย่างไร

Training–Serving Skew

Training–serving skew คือ feature ตอนใช้งานจริงคำนวณไม่เหมือนตอนฝึก เช่น ตอนฝึกตัดช่องว่างและแปลงตัวอักษรเป็นเล็ก แต่ service ลืมทำ หรือ training ใช้อัตราแลกเปลี่ยนสิ้นวัน ขณะที่ online service ใช้อัตราปัจจุบัน PipelineModel ช่วยบรรจุ transformation ที่อยู่ใน Pipeline แต่ไม่แก้ความแตกต่างจากระบบภายนอกทั้งหมด เราจึงต้องมีสัญญา feature, integration test และตรวจ distribution ของ output จริง

Data Drift, Concept Drift และ Data Quality

การ monitor จึงต้องมากกว่า latency ของ job เราควรดู null rate, unknown category rate, distribution ของ feature, prediction rate, metric ที่ได้เมื่อ label มาถึง และจำนวนแถวที่ถูก quarantine หากข้อมูลผิดแต่ pipeline ยังรันสำเร็จ ความสำเร็จทางระบบอาจซ่อนความล้มเหลวทางความหมาย

Batch Inference กับ Online Inference

Spark เหมาะกับ batch scoring ข้อมูลจำนวนมาก เช่นคำนวณคะแนนลูกค้าทุกคืน แต่ถ้าต้องตอบภายในไม่กี่สิบมิลลิวินาทีต่อ request อาจต้อง export หรือให้บริการโมเดลผ่านระบบ online แยกต่างหาก จุดท้าทายคือทำให้ feature semantics ตรงกัน ไม่ใช่เพียงย้ายไฟล์โมเดลสำเร็จ

Reproducibility มีหลายชั้น

• Code: ระบุ commit หรือ artifact ที่ใช้
• Data: ระบุ snapshot/version และเวลาที่อ่านข้อมูล
• Environment: ระบุ Spark, library, runtime และ configuration สำคัญ
• Randomness: กำหนด seed แต่ยอมรับว่าการคำนวณแบบกระจายบางอย่างอาจไม่ deterministic แบบ bit-for-bit
• Decision: บันทึก metric, threshold และเหตุผลที่เลือกโมเดล

12เมื่อยุคเปลี่ยน: Spark SQL/ML ในปี 2026

สิ่งที่เปลี่ยนกับสิ่งที่ยังเป็นแก่นเดิม

ชื่อ API และความสามารถของ optimizer เปลี่ยนตามรุ่น แต่ต้นทุนพื้นฐานยังวนกลับมาที่เรื่องเดิม คืออ่านข้อมูลเท่าไร ย้ายข้อมูลข้ามเครือข่ายเท่าไร งานกระจายสมดุลหรือไม่ และมีจุดใดบีบทุกอย่างกลับไปเครื่องเดียว เครื่องมือใหม่ช่วยตัดสินใจได้ดีขึ้น แต่ไม่ลบข้อจำกัดของ CPU, memory, disk และ network

13กรณีศึกษา: จากคำถามทางธุรกิจถึง ML Pipeline

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

ขั้นที่ 1: กำหนด Observation Time และ Label Window

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

ขั้นที่ 2: ลดข้อมูลก่อน Join

ตาราง event อาจมีหลายพันล้านแถว เรากรองเฉพาะช่วง 90 วัน เลือกคอลัมน์ที่จำเป็น แล้ว aggregate ต่อ customer ก่อน Join กับตารางลูกค้า การรวมใกล้แหล่งข้อมูลช่วยลด Shuffle จาก event หลายพันล้านแถวเหลือ feature หนึ่งแถวต่อลูกค้า

WITH usage_90d AS (
  SELECT
    customer_id,
    COUNT(*) AS session_count_90d,
    SUM(bytes_used) AS bytes_90d,
    MAX(event_time) AS last_seen
  FROM usage_events
  WHERE event_date >= DATE '2026-05-01'
    AND event_date <  DATE '2026-08-01'
  GROUP BY customer_id
)
SELECT c.customer_id, c.segment, c.tenure_months,
       u.session_count_90d, u.bytes_90d, u.last_seen,
       l.churned_30d AS label
FROM customer_snapshot c
LEFT JOIN usage_90d u USING (customer_id)
JOIN labels l USING (customer_id);

ขั้นที่ 3: ตรวจคุณภาพและ Distribution

ก่อนฝึกโมเดล เราตรวจจำนวนแถวซ้ำ null rate สัดส่วน label ช่วงค่าของ tenure และความสดของ last_seen ถ้า Join ทำให้ลูกค้าหนึ่งคนกลายเป็นหลายแถวโดยไม่ตั้งใจ ทั้ง metric และ distribution จะผิด แม้ SQL ไม่มี error

ขั้นที่ 4: แบ่งตามเวลา แล้ว Fit Pipeline

ใช้ snapshot เดือนก่อนเป็น training และเดือนล่าสุดเป็น test ภายใน Pipeline จัดการ missing value, index segment, assemble feature และฝึก Logistic Regression จุดสำคัญคือ Estimator ทุกตัวเรียนจาก training เท่านั้น และเก็บ fitted mapping ไว้ใน PipelineModel

ขั้นที่ 5: ประเมินทั้งโมเดลและระบบ

นอกจาก PR-AUC และ Recall ที่งบแจ้งเตือนกำหนดไว้ เราวัดเวลา query, ปริมาณ Shuffle, จำนวนไฟล์ที่อ่าน, เวลา fit และเวลา batch scoring ถ้า feature pipeline ใช้เวลา 12 ชั่วโมงต่อวัน โมเดลแม่นขึ้นเล็กน้อยอาจไม่ทันเวลาที่ทีมรักษาลูกค้าจะดำเนินการ

ขั้นที่ 6: นำไปใช้และเฝ้าดู

บันทึก PipelineModel พร้อม metadata เขียนคะแนนลงตารางที่มี model version และ scoring time แล้วติดตาม unknown category, feature drift, สัดส่วนผู้ถูกแจ้งเตือน และผลลัพธ์เมื่อ label มาถึง ถ้าสัดส่วนลูกค้าแพ็กเกจใหม่เพิ่มขึ้น แต่ training ไม่เคยเห็น ระบบอาจยังรันได้โดยไม่ error ขณะที่คุณภาพลดลง

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

บทนี้เริ่มจาก RDD ซึ่งทำให้เราเห็นข้อมูลเป็น partition ที่มี lineage และคำนวณแบบ Lazy จากนั้นยกระดับไปสู่ DataFrame และ SQL ซึ่งเปิดเผย schema กับความสัมพันธ์ของ operation ให้ Catalyst ช่วยจัดแผน เมื่ออ่าน physical plan เราจึงเชื่อมคำสั่งระดับสูงกลับไปยังต้นทุนจริง เช่น Scan, Shuffle, Sort, Broadcast และ Data Skew ได้

ในส่วน Machine Learning เราไม่ได้หยุดที่การเรียก fit() แต่ประกอบการเตรียม feature, model และ transformation เป็น Pipeline แบ่งข้อมูลอย่างไม่ให้อนาคตรั่ว เลือก metric ตามต้นทุนของความผิดพลาด และบันทึกกระบวนการให้ทำซ้ำได้ แก่นสำคัญคือโมเดลเป็นเพียงหนึ่ง stage ในระบบข้อมูลขนาดใหญ่

ขั้นตอนถัดไป คือบท Streaming แบบเรียลไทม์ ซึ่งเพิ่มมิติของเวลาเข้ามา งานไม่ได้มีเพียงข้อมูลก้อนหนึ่งที่รู้จุดจบ แต่มีข้อมูลไหลเข้าต่อเนื่อง เราจะต้องคิดเรื่อง event time, processing time, window, watermark, state และความหมายของผลลัพธ์เมื่อข้อมูลมาช้า แนวคิด DataFrame, SQL, partition และ state ที่เรียนในบทนี้จะเป็นฐานโดยตรง