Spark x Scala : ทำความรู้จักกับ RDD

ถ้าคุณกำลังเริ่มต้นกับการประมวลผลข้อมูลขนาดใหญ่ คุณอาจจะได้ยินคำว่า Spark และ RDD บ่อยๆ วันนี้เรามาทำความรู้จักกับ RDD กันให้มากขึ้นกันเถอะ!

RDD คืออะไร?

RDD ย่อมาจาก Resilient Distributed Dataset เป็นโครงสร้างข้อมูลพื้นฐานใน Apache Spark ที่ช่วยให้เราจัดการกับข้อมูลขนาดใหญ่ได้อย่างมีประสิทธิภาพ

ความหมายของแต่ละคำใน RDD

• Resilient (ทนทาน): สามารถฟื้นฟูข้อมูลได้เมื่อเกิดความล้มเหลว
• Distributed (กระจาย): ข้อมูลถูกกระจายไปยังหลายๆ node ในคลัสเตอร์
• Dataset (ชุดข้อมูล): คอลเลกชันของข้อมูลที่สามารถทำงานร่วมกันได้

ทำไมต้องใช้ RDD ?

1. การประมวลผลแบบขนาน (Parallel processing): RDD ช่วยให้เราแบ่งงานออกเป็นส่วนย่อยๆ และประมวลผลพร้อมกันได้
2. ความทนทานต่อความล้มเหลว (Fault tolerance): หากเกิดปัญหากับ node ใดๆ Spark สามารถสร้างข้อมูลขึ้นมาใหม่ได้
3. Lazy evaluation (การประเมินอย่างล่าช้า): RDD จะไม่คำนวณผลลัพธ์จนกว่าจะจำเป็นต้องใช้จริงๆ ช่วยประหยัดทรัพยากร
4. Immutability (ความไม่สามารถเปลี่ยนแปลงได้): RDD ไม่สามารถเปลี่ยนแปลงได้หลังจากสร้างขึ้น ช่วยลดข้อผิดพลาดและทำให้โปรแกรมเสถียรขึ้น

การทำงานกับ RDD ใน Scala

import org.apache.spark.rdd.RDD

// สร้าง RDD จาก collection
val numbers: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5))

// การ transform RDD
val doubled: RDD[Int] = numbers.map(_ * 2)

// การ action บน RDD
val sum: Int = doubled.reduce(_ + _)

println(s"ผลรวมหลังจากคูณ 2: $sum")

ในตัวอย่างนี้ เราสร้าง RDD จากลิสต์ธรรมดา ทำการ transform โดยคูณทุกตัวด้วย 2 และสุดท้ายคำนวณผลรวม

ใน Spark มีการแบ่งการทำงานเป็นสองประเภท

1. Transformations (การแปลง): สร้าง RDD ใหม่จาก RDD ที่มีอยู่ เช่น map(), filter(), flatMap()
2. Actions (การกระทำ): คำนวณผลลัพธ์และส่งกลับไปยังdriver program หรือเขียนลงในระบบจัดเก็บ เช่น reduce(), collect(), count()

เรื่องสำคัญคือ transformations เป็นแบบ lazy evaluation ส่วน actions จะทำให้เกิดการคำนวณจริงๆ

Summary

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

การเข้าใจ RDD เป็นพื้นฐานสำคัญในการใช้งาน Spark อย่างมีประสิทธิภาพ แม้ว่าในปัจจุบันจะมี DataFrame และ Dataset ที่ใช้งานง่ายกว่า แต่การรู้จัก RDD ก็ยังมีประโยชน์อยู่เสมอ!

** แถมส่งท้าย **

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

File income_data.txt , DemoRdd.scala

A,2020,50000
B,2020,55000
C,2020,48000
A,2021,52000
B,2021,57000
C,2021,50000
A,2022,54000
B,2022,59000
C,2022,52000

package demoSpark
import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.log4j.{Level, Logger}
object Rdd extends App {

  // Set only the Spark logger to ERROR level
  Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

  // Create SparkContext
  val conf = new SparkConf().setAppName("Income Analysis").setMaster("local[*]")
  val sc = new SparkContext(conf)

  // Load data from file
  val rawData: RDD[String] = sc.textFile("data/income_data.txt")

  // Transform data into tuple (employee, year, income)
  val incomeData: RDD[(String, Int, Int)] = rawData.map { line =>
    val parts = line.split(",")
    (parts(0), parts(1).toInt, parts(2).toInt)
  }

  // Calculate average income for each employee
  val avgIncomeByEmployee: RDD[(String, Double)] = incomeData
    .map { case (employee, _, income) => (employee, (income, 1)) }
    .reduceByKey { case ((sum1, count1), (sum2, count2)) => (sum1 + sum2, count1 + count2) }
    .mapValues { case (sum, count) => sum.toDouble / count }

  // Calculate total income for each year
  val totalIncomeByYear: RDD[(Int, Int)] = incomeData
    .map { case (_, year, income) => (year, income) }
    .reduceByKey(_ + _)

  // Show results
  println("Average income by employee:")
  avgIncomeByEmployee.collect().foreach { case (employee, avgIncome) =>
    println(s"$employee: $avgIncome")
  }

  println("\nTotal income by year:")
  totalIncomeByYear.collect().foreach { case (year, totalIncome) =>
    println(s"$year: $totalIncome")
  }

  // Stop SparkContext
  sc.stop()
}

ลองทำความเข้าใจดูครับไม่ยากเกินความตั้งใจ จะเห็นว่าคล้ายการทำงานของ Mapreduce แต่มีความยืดหยุ่นกว่ามากครับ

ถ้ายังไม่ค่อยรู้ Mapreduce อ่านได้ที่นี่ครับ …

result

Average income by employee:
B: 57000.0
A: 52000.0
C: 50000.0

Total income by year:
2022: 165000
2020: 153000
2021: 159000

Process finished with exit code 0