Spark SQL และ ML Pipeline

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

RDD รองรับสอง operation หลัก: Transformation (เช่น map, filter, join — คืนค่าเป็น RDD ใหม่ และขี้เกียจ (lazy) คือยังไม่รันจริงจนกว่าจะมี action) และ Action (เช่น count, collect, show — บังคับให้ Spark รัน transformation ทั้งหมดที่ค้างอยู่และคืนผลลัพธ์จริง)

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

3Spark SQL และ DataFrame

// อ่านข้อมูลแบบ 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 vs Action ใน DataFrame

ตัวอย่างการทำงานกับ 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(3).toInt)
}
val df = peopleRDD.toDF

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

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

4Spark ML Pipeline API

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)

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