GFS/HDFS และ MapReduce

1GFS/HDFS: Distributed File System

Google สร้าง GFS ให้เหมาะกับ workload ภายในของตน ส่วน HDFS ได้แรงบันดาลใจอย่างมากจากหลักการของ GFS และนำมาใช้ในระบบ Hadoop ทั้งสองมีแนวคิดร่วม เช่น master metadata, block/chunk ขนาดใหญ่ และการทำสำเนา แต่ไม่ควรสรุปว่า HDFS เป็นเพียง clone ที่ต่างเฉพาะชื่อเรียก เพราะ implementation, interface และวิวัฒนาการต่างกัน

Distributed File System ไม่ใช่ Network Drive ขนาดใหญ่เฉย ๆ

Network drive ทั่วไปอาจเก็บไฟล์ไว้ศูนย์กลางแล้วให้ client อ่านผ่าน network แต่ HDFS กระจาย block ไปยัง DataNode หลายเครื่อง และพยายามรัน computation บนเครื่องเดียวกับ block ระบบไฟล์จึงเป็นส่วนหนึ่งของ execution architecture ไม่ใช่เพียงพื้นที่เก็บ

GFS: Design Decisions

• Chunk ขนาดใหญ่: GFS ดั้งเดิมใช้ chunk 64 MB เพื่อลดจำนวน metadata และให้การอ่านต่อเนื่องคุ้ม overhead
• Replication: เก็บสำเนาหลาย chunkserver ตาม placement policy เพื่อทนต่อเครื่องหรือดิสก์เสีย
• Master metadata: รวม namespace และแผนที่ chunk ไว้ที่ master เพื่อให้การตัดสินใจเรียบง่าย แต่ต้องออกแบบ recovery และ availability ของ master
• เหมาะกับ Streaming Read: ไม่เน้น cache block ขนาดใหญ่ซึ่งมักอ่านครั้งเดียว เพราะ page cache ของ OS ยังทำหน้าที่บางส่วนได้
• ไม่ยึด POSIX เต็มรูปแบบ: ยอมเปลี่ยน semantics บางอย่างให้เหมาะกับ workload ขนาดใหญ่และการกระจาย

Block ใหญ่ช่วยอะไร และสร้างปัญหาอะไร

Block ใหญ่ลด metadata ต่อไฟล์ ลดจำนวน seek และทำให้ task ทำงานนานพอคุ้ม scheduling แต่ block ใหญ่เกินไปลดจำนวนชิ้นงานที่แบ่งได้ และทำให้ retry หนึ่ง task แพงขึ้น ขนาด block ใน HDFS กำหนดค่าได้และมักใหญ่กว่าระบบไฟล์ทั่วไป จึงไม่ควรจำตัวเลขเดียวเป็นกฎถาวร

Block, InputSplit และ Record ไม่ใช่สิ่งเดียวกัน

InputSplit มักสอดคล้องกับ block เพื่อใช้ data locality แต่ไม่จำเป็นต้องเหมือนกันเสมอ RecordReader ต้องจัดการ record ที่ข้าม boundary เช่นบรรทัดข้อความซึ่งเริ่มท้าย split หนึ่งและจบใน split ถัดไป

HDFS Architecture

Hadoop 1 กับ YARN ไม่ควรปนกัน

HDFS จัดเก็บข้อมูล ส่วน MapReduce/YARN จัด execution ทั้งสองทำงานร่วมกันแต่เป็นคนละ subsystem การแยกนี้ช่วยให้ YARN รองรับ processing framework อื่นนอกจาก MapReduce ได้

เส้นทางการอ่าน

1. Client ขอเปิดไฟล์จาก NameNode
2. NameNode ส่งตำแหน่ง replica ของแต่ละ block โดยเรียงตามความใกล้ที่เหมาะสม
3. Client อ่าน block จาก DataNode โดยตรงและตรวจ checksum
4. หาก replica หนึ่งอ่านไม่ได้ Client เลือก replica อื่นและรายงานปัญหา

เส้นทางการเขียนแบบ Pipeline

Client ขอสร้างไฟล์ NameNode เลือก DataNode ตาม replication กับ rack policy จากนั้น Client ส่ง packet ไป DataNode ตัวแรก ซึ่งส่งต่อไปตัวที่สองและตัวที่สาม acknowledgement ไหลย้อนกลับมา วิธีนี้ทำให้ Client ไม่ต้องส่งสำเนาเต็มแยกสามครั้งจากต้นทาง

Heartbeat, Block Report และการซ่อมสำเนา

DataNode ส่ง heartbeat บอกว่ายังทำงาน และส่ง block report บอกว่ามี block ใดบ้าง หาก NameNode ไม่ได้รับ heartbeat ตามเงื่อนไข จะมอง DataNode ว่าใช้งานไม่ได้และสั่งสร้าง replica เพิ่มจากสำเนาที่เหลือ การซ่อมใช้ bandwidth จึงต้องควบคุมไม่ให้รบกวนงานปกติมากเกินไป

High Availability ไม่เท่ากับ Backup

HA NameNode ช่วยให้บริการ metadata ต่อได้เมื่อ Active NameNode มีปัญหา ส่วน backup หรือ snapshot ช่วยย้อนกลับจากการลบผิดหรือข้อมูลเสียหายเชิงตรรกะ ระบบที่มี HA ยังต้องมีนโยบายสำรองข้อมูล เพราะการลบไฟล์ผิดอาจถูกทำซ้ำอย่างถูกต้องไปยังระบบที่พร้อมใช้งานทุกชุด

Small Files Problem

ไฟล์เล็กจำนวนมากเพิ่ม metadata ใน NameNode และสร้าง Map task สั้นจำนวนมาก ค่า scheduling กับการเปิดไฟล์อาจมากกว่าการประมวลผล การรวมไฟล์ จัด partition และเลือกรูปแบบอย่าง Parquet จึงสำคัญพอ ๆ กับการเพิ่ม DataNode

HDFS เหมาะกับอะไร และไม่เหมาะกับอะไร

2แทนที่จะย้ายข้อมูลมาหาโปรแกรม เราส่งโปรแกรมไปหาข้อมูล

ในโปรแกรมทั่วไป เรามักคิดว่าเครื่องของเราเป็นศูนย์กลาง: เปิดไฟล์จากที่อื่น ดึงข้อมูลเข้ามา แล้วจึงเริ่มคำนวณ วิธีนี้ใช้ได้เมื่อข้อมูลมีขนาดเล็ก แต่กลับด้านทันทีเมื่อข้อมูลมีขนาดหลายเทราไบต์ เพราะสิ่งที่ต้องขนผ่านเครือข่ายมีขนาดใหญ่กว่าสิ่งที่ใช้คำนวณมหาศาล

HDFS รู้ว่าแต่ละ block อยู่ที่ DataNode ใด ส่วน scheduler รู้ว่า worker ใดว่าง ระบบจึงพยายามเริ่ม Map task บนเครื่องที่มี block นั้นอยู่แล้ว เรียกว่า data locality ถ้าทำไม่ได้จึงค่อยเลือกเครื่องใน rack เดียวกัน และใช้การส่งข้อมูลข้าม rack เป็นทางเลือกท้าย ๆ

Locality มีหลายระดับ และ Scheduler ต้องแลกกับเวลารอ

Node-local ใช้ local disk หรือ local cache ได้ดีที่สุด Rack-local ยังส่งผ่าน network แต่ไม่ข้ามลิงก์ส่วนกลางมาก ส่วน off-rack แย่ง bandwidth ข้าม rack อย่างไรก็ตาม หากเครื่องที่มีข้อมูลเต็มทั้งหมด การรอ locality นานเกินไปอาจช้ากว่าส่งข้อมูลไปเครื่องว่าง Scheduler จึงต้องแลกระหว่าง “ใกล้ข้อมูล” กับ “เริ่มงานได้เร็ว”

Code เล็กกว่า Data แต่ Environment อาจไม่เล็ก

แนวคิดส่ง code ไปหา data ใช้ได้ดีเมื่อ code กับ dependency มีขนาดเล็กและ worker มี runtime พร้อม หากแต่ละ task ต้องส่ง container image ขนาดใหญ่หรือติดตั้ง library ใหม่ ต้นทุนเริ่มงานจะสูง ระบบจึงใช้ image cache, shared runtime และ environment ที่สร้างไว้ล่วงหน้า

การส่ง Code ไปหา Data ไม่ได้แปลว่าไม่มี Data Movement

Map task อ่าน input แบบ local ได้ แต่เมื่อจัดกลุ่ม key เดียวกัน ผลจาก Mapper ทุกเครื่องต้องเดินทางไปยัง Reducer ที่รับผิดชอบ นี่คือ Shuffle ซึ่งมักเป็นช่วงที่ใช้ network มากที่สุด หลักการจึงเปลี่ยนจาก “ห้ามย้ายข้อมูล” เป็น “ย้ายเฉพาะข้อมูลที่จำเป็น และลดให้เล็กที่สุดก่อนย้าย”

เมื่อ Storage แยกจาก Compute

Object storage ไม่มีแนวคิดว่า task ควรรันบนเครื่องที่มีดิสก์ block นั้นโดยตรง ระบบสมัยใหม่จึงรักษาแก่นของ locality ด้วยวิธีอื่น:

• Cache ข้อมูลที่อ่านบ่อยใกล้ compute
• ใช้ columnar format เพื่ออ่านเฉพาะคอลัมน์
• ใช้ partition pruning และ predicate pushdown ลด object ที่ต้องอ่าน
• วาง compute กับ storage ใน region เดียวกัน
• รวม request และอ่านเป็นช่วงใหญ่เพื่อใช้ bandwidth คุ้ม

Data locality จึงไม่จำเป็นต้องหมายถึงดิสก์กับ CPU อยู่ในกล่องเดียวเสมอไป ความหมายที่กว้างกว่าคือทำให้ computation กับข้อมูลที่ต้องใช้มีระยะห่างทางต้นทุนต่ำที่สุด

Locality กับ Privacy และ Data Sovereignty

บางข้อมูลย้ายไม่ได้เพราะกฎ นโยบาย หรือความเป็นส่วนตัว เราอาจส่ง model หรือ query ไปประมวลผลในพื้นที่ แล้วนำกลับเฉพาะ aggregate ที่อนุญาต หลัก “ส่ง code ไปหา data” จึงเกี่ยวกับ governance ด้วย ไม่ใช่ performance อย่างเดียว

3MapReduce: Programming Model

ภายใน job และ partitioner เดียวกัน ค่าที่มี key เดียวกันจะถูกส่งไป partition เดียวและถูกจัดกลุ่มให้ Reducer ประมวลผล นี่คือภาพพื้นฐานของ distributed GROUP BY แต่ถ้า key กระจายไม่ดี Reducer บางตัวจะรับงานมากกว่าส่วนอื่น

Map, Shuffle และ Reduce ทำหน้าที่คนละอย่าง

Key เป็นการตัดสินใจด้านความหมายและตำแหน่ง

Key บอกทั้งว่า record ใดควรถูกรวมกันและควรเดินทางไป Reducer ใด หากเลือก key ละเอียดเกินไปจะมี key จำนวนมาก หากหยาบเกินไปข้อมูลกองที่ Reducer เดียว เช่นใช้ประเทศเป็น key ในข้อมูลที่ 90% มาจากประเทศเดียว

การออกแบบ key จึงต้องดู distribution ไม่ใช่ดู semantics อย่างเดียว บางกรณีเติม salt เพื่อกระจาย hot key แล้วมี Reduce รอบถัดไปรวมผลกลับมา

รากฐานจาก Functional Programming

MapReduce นำแรงบันดาลใจจาก map กับ fold/reduce ใน Functional Programming มาประกอบกับ execution framework แบบกระจาย Map ทำงานกับ record แต่ละตัวได้ค่อนข้างอิสระ ส่วน Reduce รวมค่าตาม key

// ตัวอย่างจากเอกสารต้นฉบับ — Scala functional style
scala> val t = Array(1, 2, 3, 4, 5)
scala> t.map(n => n*n)
res0: Array[Int] = Array(1, 4, 9, 16, 25)

scala> t.map(n => n*n).foldLeft(0)((m, n) => m + n)
res2: Int = 55
// map แต่ละค่าแยกกันได้
// แต่ foldLeft ระบุลำดับจากซ้ายไปขวา จึงยังไม่ใช่ตัวอย่างการ reduce แบบขนานโดยตรง

MapReduce จำกัดรูปแบบ เพื่อให้ Framework ช่วยได้มาก

ความยืดหยุ่นที่น้อยลงเป็นข้อดีส่วนหนึ่ง เมื่อ framework รู้ว่างานมี Map, Shuffle และ Reduce มันสามารถ retry Map task บนเครื่องอื่น partition ตาม key และเริ่มดึง map output ล่วงหน้าได้ หากผู้ใช้ส่งโปรแกรมที่แก้ shared state ทั่วคลัสเตอร์อย่างอิสระ การกู้คืนและจัดตารางจะยากกว่ามาก

ตัวอย่างคลาสสิก: Word Count

// map: อ่านแต่ละบรรทัด แยกคำ แล้ว emit (คำ, 1)
def map(key: LongWritable, value: Text) = {
  for (word <- value.toString.split("\\s+"))
    emit(word, 1)
}

// reduce: รวมค่า count ของแต่ละคำเข้าด้วยกัน
def reduce(key: Text, values: Iterable[Int]) = {
  emit(key, values.sum)
}

ตัวอย่างที่มีประโยชน์กว่า Word Count: Inverted Index

Search engine ต้องรู้ว่าคำหนึ่งปรากฏในเอกสารใด Mapper อ่านเอกสารแล้วส่ง (term, documentId) Reducer รวม documentId ของ term เดียวกันเป็น posting list รูปแบบนี้แสดงว่า MapReduce ไม่ได้จำกัดอยู่กับการนับ แต่สร้าง index แบบกระจายได้

// map(documentId, text)
for each distinct term in text:
    emit(term, documentId)

// reduce(term, documentIds)
emit(term, sortedUnique(documentIds))

Map-only และ Reduce-only ในทางแนวคิด

งานที่แต่ละ record แปลงได้อิสระและไม่ต้องรวม key อาจใช้ Map-only job เช่นแปลง format หรือกรองข้อมูล ส่วนงาน Reduce ต้องอาศัย input ที่จัดกลุ่มไว้ก่อน ใน framework จริง configuration กับ input อาจต่างกัน แต่การแยกช่วยถามว่า Shuffle จำเป็นจริงหรือไม่

Physical View: เกิดอะไรขึ้นจริงในคลัสเตอร์

Intermediate Data อยู่ไหน

Map output ภายใน job มักอยู่ local disk ของ Map worker ไม่ได้เขียน HDFS เพราะเป็นข้อมูลชั่วคราวและสร้างใหม่ได้ หาก worker ที่เก็บ map output หาย ระบบรัน Map task นั้นใหม่ ส่วน output สุดท้ายของ job จึงเขียนไปยัง distributed file system เพื่อความทนทาน

Failure Recovery

Map task เป็น deterministic และไม่มี side effect ภายนอกจะ retry ได้ง่าย Reducer ที่ล้มสามารถดึง map output ใหม่และคำนวณซ้ำ แต่ถ้า Mapper ส่งอีเมลหรือแก้ฐานข้อมูลภายนอก การ retry อาจทำงานซ้ำ Framework ไม่สามารถแก้ semantics ของ side effect แทนผู้ใช้ได้

Speculative Execution

ถ้า task หนึ่งช้ากว่ากลุ่มมาก ระบบอาจเปิดสำเนาบน worker อื่นและรับผลจากตัวที่เสร็จก่อน วิธีนี้ช่วยเมื่อเครื่องช้าผิดปกติ แต่ไม่ช่วย data skew ที่ทำให้ task รับข้อมูลมากจริง และไม่เหมาะกับ task ที่มี side effect ซึ่งทำซ้ำไม่ได้

Counters และการสังเกตคุณภาพข้อมูล

นอกจากผลลัพธ์ Job สามารถนับ record เสีย ค่า missing หรือเหตุการณ์พิเศษผ่าน counter ช่วยให้ pipeline รายงานคุณภาพข้อมูลพร้อมการประมวลผล แต่ counter จำนวนมากหรือ key ที่ไม่จำกัดอาจเพิ่ม overhead จึงควรใช้กับตัวชี้วัดที่กำหนดไว้

4รวมผลก่อนส่ง: ลดเวลารอด้วยโครงสร้างของการคำนวณ

เวลาทำงานบนคลัสเตอร์ สิ่งที่แพงมักไม่ใช่เครื่องหมายบวก แต่เป็นการส่งข้อมูลข้ามเครือข่ายและการรอให้เครื่องอื่นทำงานเสร็จ ถ้า Mapper พบคำว่า data หนึ่งหมื่นครั้ง การส่งคู่ (data, 1) ออกไปหนึ่งหมื่นคู่ย่อมสิ้นเปลืองกว่า การรวมในเครื่องนั้นก่อนแล้วส่งเพียง (data, 10000)

Combiner และ In-mapper Combining

Monoid: เหตุผลที่บางงานรวมแบบขนานได้

ผลรวมแบบขนานทำได้ง่ายเมื่อเราสามารถจัดกลุ่มใหม่ได้โดยคำตอบไม่เปลี่ยน เช่น (a+b)+c = a+(b+c) และมีค่าเริ่มต้น เช่น 0 สำหรับการบวก แนวคิดนี้เรียกว่า Monoid ชื่ออาจดูเป็นคณิตศาสตร์ แต่ความหมายตรงไปตรงมา: แต่ละเครื่องรวมผลของตนก่อน แล้วค่อยนำผลย่อยมารวมกันเป็นต้นไม้ ไม่จำเป็นต้องต่อแถวรอเครื่องเดียว

เงื่อนไขสามข้อที่ควรเห็นให้ชัด

คำว่า Monoid ไม่ได้หมายถึงอัลกอริทึมชนิดเดียว แต่เป็นโครงสร้างที่บอกว่า “ผลย่อยรูปแบบนี้นำมาต่อกันได้อย่างปลอดภัย” โดยมีองค์ประกอบสามอย่าง

Commutativity หรือการสลับที่ เช่น a+b=b+a เป็นคุณสมบัติที่มีประโยชน์มาก เพราะข้อมูลอาจมาถึงคนละลำดับ แต่ไม่ใช่เงื่อนไขบังคับของ Monoid ทุกชนิด การต่อข้อความเป็น Monoid ได้ แม้ "ab" จะไม่เท่ากับ "ba" เพียงแต่ถ้าผลลัพธ์ขึ้นกับลำดับ เราต้องกำหนดลำดับนั้นเอง ไม่ควรหวังว่า Shuffle จะจัดให้โดยอัตโนมัติ

จากเส้นตรงเป็นต้นไม้: ลดช่วงที่ต้องรอ

ถ้ามีผลย่อย 1,024 ชิ้นแล้วรวมทีละชิ้น เส้นทางที่ยาวที่สุดต้องผ่านการรวมประมาณ 1,023 ครั้ง แต่ถ้าจับคู่พร้อมกันเป็นชั้น ๆ ความลึกของต้นไม้เหลือประมาณ log₂(1024) = 10 ชั้น จำนวนการรวมทั้งหมดไม่ได้หายไปมากนัก แต่ critical path หรือช่วงเวลาที่ยาวที่สุดซึ่งทุกคนต้องรอ ลดลงอย่างมาก นี่เป็นแก่นของ parallel reduction

อย่างไรก็ตาม ต้นไม้ไม่ได้โตฟรี ทุกชั้นมีค่าใช้จ่ายในการส่งข้อมูล สร้าง task และประสานงาน หากข้อมูลมีเพียงสิบค่า การสร้างงานบนคลัสเตอร์อาจแพงกว่าการบวกบนเครื่องเดียว การออกแบบที่ดีจึงถามทั้ง “ขนานได้ไหม” และ “ขนาดเท่าใดจึงคุ้มกับ overhead”

Monoidification: เปลี่ยนสิ่งที่ถือไว้ เพื่อให้รวมได้

ปัญหาหลายอย่างดูเหมือนรวมผลย่อยไม่ได้ เพราะเราเลือกตัวแทนของผลย่อยไม่ดี Monoidification คือการเปลี่ยนรูปข้อมูลกลางให้เก็บข้อมูลพอสำหรับรวมต่อ โดยไม่จำเป็นต้องเก็บข้อมูลดิบทั้งหมด เป็นเทคนิคที่ใช้ตั้งแต่ Combiner, distributed aggregation, streaming ไปจนถึง query optimizer

สมมุติกลุ่ม A มี 10 ค่า ค่าเฉลี่ย 80 และกลุ่ม B มี 90 ค่า ค่าเฉลี่ย 60 หากเอา 80 กับ 60 มาเฉลี่ยตรง ๆ จะได้ 70 ซึ่งผิด เพราะสองกลุ่มมีน้ำหนักไม่เท่ากัน แต่ถ้าส่ง (800,10) และ (5400,90) เรารวมเป็น (6200,100) แล้วจึงได้ค่าเฉลี่ย 62 อย่างถูกต้อง

ความแปรปรวนก็รวมได้ แต่ต้องเก็บสถานะให้พอ

ถ้าต้องการ variance การส่งเพียงค่าเฉลี่ยของแต่ละกลุ่มยังไม่พอ วิธีง่ายในชั้นเรียนคือส่ง (count, sum, sumOfSquares) แล้วรวมองค์ประกอบตามตำแหน่ง ก่อนคำนวณ variance ตอนท้าย ในงานตัวเลขจริงต้องระวังความคลาดเคลื่อนจาก floating point และอาจใช้สถานะ (count, mean, M2) ตามแนวคิด Welford/Chan ซึ่งเสถียรกว่า ประเด็นสำคัญไม่ใช่จำสูตร แต่ให้ถามว่า ผลย่อยต้องถืออะไรไว้ จึงรวมกับเครื่องอื่นได้โดยไม่ย้อนอ่านข้อมูลดิบ

Median เตือนว่า Summary อาจใหญ่กว่าที่คิด

Median ไม่มีผลสรุปขนาดคงที่แบบ (sum,count) ที่ให้คำตอบแม่นยำเสมอ หากต้องการคำตอบแน่นอน เราอาจต้องเก็บหรือจัดลำดับข้อมูลจำนวนมาก แต่ถ้ายอมคลาดเคลื่อนได้ อาจใช้ quantile sketch ซึ่งรวมผลย่อยได้ นี่ทำให้เห็นการแลกระหว่าง ความแม่นยำ หน่วยความจำ ปริมาณข้อมูลที่ส่ง และเวลา

ข้อควรระวังเรื่อง Floating Point

ในจำนวนจริง การบวกเป็น associative แต่ในคอมพิวเตอร์ (a+b)+c อาจไม่เท่ากับ a+(b+c) ทุกบิต เพราะ floating point มีการปัดเศษ การเปลี่ยนจำนวน partition หรือรูปต้นไม้อาจทำให้หลักทศนิยมท้าย ๆ เปลี่ยน งานที่ต้องการ reproducibility สูงจึงควรกำหนดลำดับ ใช้ชนิดข้อมูลที่เหมาะสม หรือใช้อัลกอริทึมสะสมค่าที่เสถียรกว่า

Combiner ไม่ใช่ Reducer ขนาดย่อมที่รับประกันว่าจะทำงาน

หลายคนใช้โค้ดเดียวกับ Reducer แล้วเผลอคิดว่าระบบจะเรียกหนึ่งครั้งหลัง Mapper ทุกตัว ความจริง Framework มีสิทธิ์ไม่เรียก เรียกหนึ่งครั้ง หรือเรียกหลายครั้งระหว่าง spill และ merge เพราะฉะนั้นผลสุดท้ายต้องถูกต้องในทุกกรณี

ผลรวม ตัวนับ ค่าต่ำสุด และค่าสูงสุดมักเหมาะ ส่วนค่าเฉลี่ยที่ส่งเป็นตัวเลขเดียวไม่เหมาะ แต่ค่าเฉลี่ยที่เปลี่ยนเป็น (sum,count) เหมาะกว่า นอกจากนี้ชนิด key/value ที่ Combiner ส่งออกต้องเข้ากับ input ที่ Reducer รอรับด้วย

In-mapper Combining: ควบคุมมากขึ้น ภาระก็มากขึ้น

In-mapper Combining ให้ Mapper เก็บ map เช่น word → count ไว้เอง แล้ว emit ตอนท้ายของ split ข้อดีคือรับประกันว่า key เดียวกันภายใน Mapper ถูกยุบก่อนส่ง ลดจำนวน object และข้อมูลผ่าน framework แต่หน่วยความจำอาจโตตามจำนวน key ที่ไม่ซ้ำ และเราสูญเสียความยืดหยุ่นบางส่วนที่ framework มีในการ spill เป็นช่วง ๆ

วิธีใช้จริงควรมี flush policy เช่น เมื่อ map มีเกินจำนวน key ที่กำหนดหรือใช้หน่วยความจำถึงเกณฑ์ ให้ emit ผลบางส่วนและล้างตาราง แล้วเริ่มสะสมใหม่ วิธีนี้ลดโอกาส OutOfMemory แม้ key เดียวกันอาจถูกส่งหลายครั้ง ซึ่งไม่เป็นปัญหาหาก Reducer รวมได้ถูกต้อง

จาก Combiner ไปสู่ SQL Partial Aggregation

เมื่อเขียน GROUP BY department และ SUM(sales) ใน SQL engine แบบกระจาย optimizer มักให้แต่ละ partition รวมยอดภายในเครื่องก่อน แล้วค่อย Shuffle ยอดย่อยตาม department ไปยังขั้นสุดท้าย หลักการเดียวกับ Combiner ยังอยู่ เพียงผู้ใช้เขียน “สิ่งที่ต้องการ” ด้วยภาษาระดับสูง และระบบเลือก “วิธีทำ” จากคุณสมบัติทาง Algebra ของฟังก์ชันรวม

5InputSplit และ Data Flow แบบเต็มรูปแบบ

Input file ถูกแบ่งเป็น InputSplit หลายชิ้น แต่ละชิ้นผ่าน RecordReader เพื่อแปลงเป็น record ที่ Mapper อ่านได้ จากนั้นข้อมูลผ่าน Mapper, buffer, spill, sort, Combiner ที่อาจมีหรือไม่มีก็ได้, Partitioner, Shuffle, merge และ Reducer ก่อนเขียนผลลัพธ์ การเห็นสายพานทั้งเส้นช่วยให้วินิจฉัยได้ว่า job ช้าตรงการอ่าน การคำนวณ การส่งข้อมูล หรือการรวมผล

เส้นทางของหนึ่ง Record

Block, Split และ Record: เส้นแบ่งสามชนิด

Block เป็นหน่วยจัดเก็บของ HDFS, InputSplit เป็นหน่วยวางแผนงานของ MapReduce และ Record เป็นหน่วยที่โปรแกรมผู้ใช้เห็น สามอย่างนี้อาจมีขนาดใกล้กันในบางกรณี แต่ไม่จำเป็นต้องตรงกัน

ไฟล์ข้อความอาจถูกแบ่งเป็น block ตรงกลางบรรทัด RecordReader ของ split ถัดไปต้องรู้ว่าจะข้ามเศษบรรทัดหรืออ่านต่อจนจบบรรทัดอย่างไร เพื่อไม่ให้ record หายหรือซ้ำ ส่วนไฟล์ gzip ทั่วไปแบ่งอ่านกลาง stream ได้ยาก แม้ HDFS จะแบ่งเก็บเป็นหลาย block ก็ไม่ได้แปลว่า Map task หลายตัวจะเริ่มถอดรหัสจากทุก block ได้อิสระ

จำนวน Mapper ไม่ได้เท่ากับจำนวนไฟล์เสมอไป

ไฟล์ใหญ่หนึ่งไฟล์อาจสร้างหลาย split และ Mapper หลายตัว ขณะที่ไฟล์เล็กจำนวนมากอาจถูกรวมเป็น split เดียวผ่าน CombineFileInputFormat เพื่อลด overhead จำนวน task การกำหนด split size จึงเป็นการแลกระหว่าง parallelism กับค่าใช้จ่ายในการสร้าง task และอ่าน metadata

ถ้า split ใหญ่มาก เรามี task น้อย ใช้คลัสเตอร์ไม่เต็มและเสี่ยงมี straggler ถ้า split เล็กมาก ระบบเสียเวลาสร้าง task เปิดไฟล์ และจัดตารางมากเกินไป เป้าหมายไม่ใช่ “แบ่งให้มากที่สุด” แต่เป็น “แบ่งพอให้กระจายงานและเหลืองานสำรองให้ scheduler จัดสมดุล”

Partitioner: ผู้คุมทางแยกของ Shuffle

Partitioner รับ key แล้วตอบหมายเลข Reducer ปกติใช้ hash คล้าย hash(key) mod R เงื่อนไขสำคัญคือ key ที่ถือว่าเท่ากันต้องไป Reducer เดียวกัน มิฉะนั้นผลรวมของ key นั้นจะแตกเป็นหลายคำตอบ แต่ hash อย่างเดียวไม่ได้รับประกันว่าแต่ละ Reducer จะมีงานเท่ากัน เพราะความถี่ของ key อาจต่างกันมาก

Data Skew: เมื่อ key เดียวทำให้ทุกคนต้องรอ

สมมุติ log 40% มี key เป็น unknown ต่อให้มี Reducer หนึ่งร้อยตัว ข้อมูลของ unknown ยังต้องไปจบที่ Reducer เดียวตาม semantics ของ group-by เครื่องอื่นอาจเสร็จหมดแล้ว แต่ job ยังรอเครื่องนี้อยู่ การเพิ่มจำนวน Reducer อย่างเดียวจึงไม่แก้ heavy key

• แยก key ผิดปกติออกต่างหาก หาก unknown ไม่ได้มีความหมายเชิงวิเคราะห์เท่ากับ key ปกติ
• Salting เติมหมายเลขย่อย เช่น (unknown,0..9) เพื่อกระจายการรวมรอบแรก แล้วมีรอบที่สองรวมสิบผลย่อยกลับเป็น unknown
• Sampling สำรวจ distribution ก่อนสร้าง partition boundary โดยเฉพาะงาน sort ตามช่วงค่า
• Map-side aggregation ลด record ของ heavy key ก่อน Shuffle หาก operation รวมผลย่อยได้

Salting เป็นการเปลี่ยน key ชั่วคราวเพื่อให้แบ่งงานได้ แล้วค่อยลบ salt และรวมอีกครั้ง แต่ใช้ไม่ได้กับทุกงาน หาก Reducer ต้องเห็น record ของ key เดียวกันทั้งหมดพร้อมกันโดยแยกไม่ได้ เราต้องออกแบบอัลกอริทึมใหม่หรือยอมรับคอขวดนั้น

Shuffle คือ Distributed Sort ที่มีค่าใช้จ่ายสูง

ระหว่าง Shuffle ข้อมูลต้องถูก serialize ออกจาก object, เขียน buffer, spill ลงดิสก์เมื่อจำเป็น, sort ตาม partition และ key, ส่งผ่านเครือข่าย, รับและ merge ที่ Reducer แล้ว deserialize กลับมา ค่าใช้จ่ายจึงมาจาก CPU, memory, disk และ network พร้อมกัน

การปรับ Mapper ให้เร็วขึ้น 20% อาจแทบไม่ช่วย หาก job ส่งข้อมูลกลางหลายเท่าของ input ในทางกลับกัน การใช้ Combiner ลด map output จากหนึ่งพันล้านคู่เหลือสิบล้านคู่อาจเปลี่ยนเวลาทั้ง job โดยไม่แก้สูตรหลักเลย metrics อย่าง map output records, spilled records, shuffle bytes และเวลารอ fetch จึงสำคัญพอ ๆ กับเวลาของฟังก์ชัน Map

Sort กับ Group ไม่ใช่ความหมายเดียวกัน

Framework เรียง intermediate key เพื่อให้ค่าของ key เดียวกันอยู่ติดกัน แล้วจึง group เป็นรายการที่ Reducer เห็น Comparator สำหรับ sort และ grouping อาจกำหนดต่างกันได้ เทคนิคนี้นำไปสู่ Secondary Sort

ตัวอย่างเช่น key เชิงประกอบเป็น (customerId, timestamp) เรา sort ด้วยทั้งสองช่องเพื่อให้รายการของลูกค้าเรียงตามเวลา แต่ group ด้วย customerId เท่านั้น Reducer จึงได้รับเหตุการณ์ของลูกค้าหนึ่งคนตามลำดับเวลา โดยไม่ต้องเก็บทุกเหตุการณ์ไว้แล้ว sort เองอีกครั้ง

Map-side Join กับ Reduce-side Join

แนวคิดนี้กลับมาใน SQL optimizer: broadcast join คล้ายการแจกตารางเล็กไปหา data partition ส่วน shuffle join คล้าย reduce-side join ภาษาที่ผู้ใช้เห็นต่างกัน แต่คำถามทางกายภาพยังเหมือนเดิมว่า “ควรย้ายข้อมูลฝั่งไหน และย้ายมากเท่าใด”

จำนวน Reducer และจำนวนไฟล์ผลลัพธ์

โดยทั่วไป Reducer หนึ่งตัวเขียน output partition หนึ่งชุด การเพิ่ม Reducer ช่วยกระจายงานแต่สร้างไฟล์ผลลัพธ์มากขึ้น หากตั้งศูนย์ Reducer จะเป็น map-only job และ Mapper เขียนผลลัพธ์สุดท้ายโดยตรง เหมาะกับงานแปลงรูปหรือกรองที่ไม่ต้องรวม key ข้าม split

จำนวน Reducer มากเกินไปทำให้เกิดไฟล์เล็กและ overhead ส่วนน้อยเกินไปทำให้แต่ละ task หนักและใช้คลัสเตอร์ไม่เต็ม ควรพิจารณาปริมาณ shuffle, memory ต่อ task, ขนาดไฟล์ปลายทาง และระบบที่จะอ่านผลต่อ ไม่ควรเลือกจากสูตรเดียวโดยไม่ดู workload

Output Commit: ป้องกันผลครึ่งเดียวและผลซ้ำ

Task อาจ retry หรือมี speculative copy มากกว่าหนึ่งตัว หากทุกตัวเขียนชื่อไฟล์สุดท้ายโดยตรง ผลอาจชนกันหรือเผยไฟล์ครึ่งเดียว Framework จึงมักให้ task เขียนไปยังพื้นที่ชั่วคราว แล้วผู้ชนะ commit ผลอย่างเป็นระบบ คล้ายการเขียนร่างก่อนแล้วค่อยประกาศฉบับจริง

แต่หากโค้ดผู้ใช้เขียนออกไปยังฐานข้อมูลภายนอก ส่งข้อความ หรือเรียก API เอง กลไก commit ของไฟล์คุม side effect เหล่านั้นไม่ได้ การออกแบบ sink จึงควรรองรับ idempotency, transaction หรือ deduplication ตามความเหมาะสม

6ความทนทาน: ทำใหม่ได้ เพราะงานถูกออกแบบให้ทำซ้ำได้

คลัสเตอร์ขนาดใหญ่ไม่ได้ตั้งอยู่บนสมมุติฐานว่าเครื่องทุกตัวจะทำงานสมบูรณ์ แต่ถือว่าเครื่อง ดิสก์ กระบวนการ และเครือข่ายบางส่วนจะมีปัญหาเป็นเรื่องปกติ วิธีรับมือของ MapReduce คือเก็บ input ที่ทนทาน และทำให้ task เป็นหน่วยงานที่รันใหม่ได้

Retry ไม่ใช่เวทมนตร์ ถ้า Function มี Side Effect

ถ้า Mapper อ่าน record แล้วคำนวณ key/value โดยไม่เปลี่ยนโลกภายนอก การรันใหม่ให้ผลเชิงตรรกะเหมือนเดิม แต่ถ้า Mapper หักเงิน ส่งอีเมล หรือเพิ่ม counter ในฐานข้อมูลทุกครั้งที่ถูกเรียก การ retry อาจทำสิ่งนั้นซ้ำ แม้ output file สุดท้ายจะถูก commit เพียงครั้งเดียว

นี่เป็นเหตุผลที่ pure function และ immutability เข้ากับ distributed processing ได้ดี ไม่ใช่เพราะ syntax สั้นกว่าเท่านั้น แต่เพราะงานที่ผลลัพธ์ขึ้นกับ input ชัดเจนสามารถย้าย รันใหม่ และตรวจสอบได้ง่ายกว่า

Map Output สร้างใหม่ได้ จึงไม่ต้อง Replicate แบบ HDFS

Map output เป็นข้อมูลกลางและมักอยู่ local disk หาก worker หาย ข้อมูลนั้นหายตามไปด้วย แต่ระบบทราบว่า output มาจาก split ใด จึงรัน Map task ใหม่บน worker อื่นได้ การ replicate intermediate ทุกชิ้นไป HDFS อาจแพงกว่าการคำนวณใหม่ นี่คือการเลือกระหว่าง เก็บสำเนา กับ เก็บที่มาของการคำนวณแล้วสร้างใหม่

แนวคิด lineage นี้พัฒนาต่อใน Spark RDD: หาก partition หาย ระบบใช้ลำดับ transformation สร้างใหม่ได้ แต่ถ้า lineage ยาวหรือคำนวณแพง อาจใช้ checkpoint ตัดสายและเก็บสถานะที่ทนทานเป็นระยะ

Straggler: เครื่องไม่ตาย แต่งานทั้งระบบยังรอ

เวลาจบของงานขนานไม่ได้ขึ้นกับ task โดยเฉลี่ย แต่ขึ้นกับ task ตัวท้าย เครื่องหนึ่งอาจช้าเพราะดิสก์มีปัญหา แย่ง CPU หรือ network congestion ระบบจึงอาจเปิด speculative copy บนเครื่องอื่น แล้วใช้ผลจากตัวที่เสร็จก่อน

แต่ speculative execution ไม่ได้แก้ทุกอย่าง หาก task ช้าเพราะรับ heavy key ที่มีข้อมูลมากจริง สำเนาทุกตัวก็ยังต้องทำงานก้อนใหญ่เท่าเดิม การวัด distribution และแก้ skew จึงสำคัญกว่าการเปิดสำเนาแบบเหวี่ยงแห

Determinism ช่วยการ Debug และทำซ้ำผล

หาก Map/Reduce เป็น deterministic เรานำ input split ที่ทำให้เกิดปัญหามารันซ้ำได้ง่ายขึ้น แต่ลำดับ values จาก Shuffle ไม่ควรถูกสมมุติว่าเหมือนเดิมเสมอ เว้นแต่กำหนด secondary sort ชัดเจน โค้ดที่เลือก “ค่าตัวแรก” โดยไม่กำหนด order อาจได้คำตอบต่างกันเมื่อจำนวน task หรือ timing เปลี่ยน

Counters, Logs และข้อมูลที่ถูกทิ้ง

Pipeline ที่จบด้วย success ไม่ได้แปลว่าข้อมูลถูกต้อง Mapper อาจ parse ไม่ได้แล้วข้าม record จำนวนมาก ควรมี counters เช่น input records, malformed rows, missing key, filtered rows และ output records แล้วตรวจความสัมพันธ์ เช่น input เท่ากับ accepted + rejected หรือไม่

ไม่ควร log record เสียทุกตัวจน I/O หนัก อาจเก็บตัวอย่างจำนวนจำกัดและเขียนข้อมูลเสียไปยัง dead-letter output แยกต่างหาก เพื่อสืบค้นโดยไม่ทำให้ log หลักท่วม

Exactly-once เป็นคุณสมบัติปลายทาง

MapReduce ให้ output file สำเร็จหนึ่งชุดผ่าน commit แต่เมื่อต่อระบบภายนอก เราต้องแยกคำว่า task รันอย่างน้อยหนึ่งครั้ง ออกจากผลทางธุรกิจเกิดหนึ่งครั้ง วิธีทั่วไปคือ idempotent key, upsert, transactional sink หรือ staging แล้ว commit ด้วย transaction

7จาก MapReduce สู่ Relational Algebra และ SQL

MapReduce มีพลังเพราะให้โครงสร้างพื้นฐานรับผิดชอบการกระจายงาน แต่การเขียนทุกคำถามเป็น Mapper และ Reducer ทำให้ผู้ใช้ต้องอธิบาย “วิธีทำ” มากเกินไป นักวิเคราะห์ส่วนใหญ่ต้องการบอกว่าอยากได้ข้อมูลอะไร เช่น กรองรายการ Join ตาราง และรวมยอดตามจังหวัด มากกว่าจะจัดการ serializer, partitioner และจำนวน Reducer เอง

MapReduce เป็นกลไก ส่วน Algebra เป็นภาษาสำหรับประกอบการคำนวณ

Relational Algebra มองข้อมูลเป็น relation และมีตัวดำเนินการที่ประกอบกันได้ เช่น selection, projection, join, grouping และ aggregation เมื่อคำสั่งมีโครงสร้างทางคณิตศาสตร์ ระบบจึงแปลงรูปได้โดยยังรักษาความหมาย เช่นผลัก filter ลงไปก่อน join หรือตัด column ที่ไม่ใช้ก่อน Shuffle

ตัวอย่างการแปลง SQL เป็น Dataflow

SELECT province, SUM(amount) AS total
FROM sales
WHERE status = 'paid'
GROUP BY province;

แผนอย่างง่ายอาจทำตามลำดับนี้:

1. อ่านเฉพาะคอลัมน์ province, amount และ status
2. กรองแถวที่ status ไม่ใช่ paid ใกล้แหล่งข้อมูลที่สุด
3. แต่ละ partition รวมยอดของ province ภายในเครื่องก่อน
4. Shuffle คู่ (province, partialSum) แทนรายการขายทุกแถว
5. รวม partial sum ของ province เดียวกัน แล้วเขียนผล

ผู้ใช้เขียน SQL เพียงไม่กี่บรรทัด แต่ข้างใต้ยังมี Map, partial aggregation, partition, Shuffle และ Reduce สิ่งที่เปลี่ยนคือ optimizer ช่วยเลือกแผนและปรับตามสถิติข้อมูล

ทำไม Algebra จึงทำให้ Optimizer กล้าเปลี่ยนแผน

ถ้าผู้ใช้ส่งฟังก์ชันที่เป็นกล่องดำ ระบบไม่รู้ว่าฟังก์ชันกรองแถว เปลี่ยน schema หรือมี side effect จึงย้ายตำแหน่งโดยไม่แน่ใจว่าจะเปลี่ยนคำตอบหรือไม่ แต่ operator ที่นิยามด้วย Algebra มีคุณสมบัติชัด ระบบจึงใช้กฎได้ เช่น:

• Predicate pushdown: กรองให้เร็ว เพื่อลดข้อมูลเข้าสู่ขั้นตอนถัดไป
• Projection pruning: อ่านและส่งเฉพาะคอลัมน์ที่จำเป็น
• Partial aggregation: รวมในเครื่องก่อน Shuffle เมื่อ aggregate แยกส่วนได้
• Join reordering: เปลี่ยนลำดับ join เพื่อลด intermediate data
• Broadcast selection: ส่งตารางเล็กไปทุก worker แทน Shuffle ตารางใหญ่ทั้งคู่

นี่คือจุดเชื่อมระหว่าง Functional Programming กับฐานข้อมูล: operator ที่ไม่มี side effect และมี semantics ชัด ทำให้ระบบจัดลำดับใหม่ แบ่งส่วน และคำนวณแบบ lazy ได้ปลอดภัยกว่า

จาก Java MapReduce ไปสู่ Pig และ Hive

ในยุค Hadoop เครื่องมืออย่าง Pig และ Hive ช่วยให้ผู้ใช้เขียน dataflow หรือ SQL-like query แล้ว compiler สร้าง MapReduce jobs หลายขั้นให้ เช่น query หนึ่งอาจกลายเป็น job สำหรับ join แล้วตามด้วย job สำหรับ aggregation ผลกลางระหว่าง job มักต้อง materialize ลง HDFS ทำให้ workflow ซับซ้อนหรือ iterative มีค่า I/O สูง

แม้เครื่องมือเหล่านี้ทำให้เขียนง่ายขึ้น แต่ข้อจำกัดของ execution engine ยังอยู่ หาก query ต้องผ่านหลาย stages ก็ยังมี job startup, HDFS read/write และ synchronization หลายรอบ จุดนี้เปิดทางให้ engine ที่สร้าง DAG ทั้งแผนและส่งต่อข้อมูลระหว่าง stages ได้ยืดหยุ่นกว่า

SQL ไม่ได้ทำให้ต้นทุนหายไป เพียงทำให้มองเห็นในอีกระดับ

คำสั่งสั้นไม่ได้แปลว่างานเบา GROUP BY บน key ที่ skew ยังทำให้ partition หนึ่งหนัก, ORDER BY ทั้งชุดยังต้องแลกข้อมูล, และ join ตารางใหญ่สองชุดยังอาจเกิด Shuffle มหาศาล ผู้ใช้ระดับสูงจึงควรอ่าน physical plan เป็น เข้าใจ partitioning และดู metrics แม้ไม่ต้องเขียน Mapper/Reducer เอง

8เมื่อยุคเปลี่ยน: จาก MapReduce สู่ Spark และไกลกว่านั้น

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

แยก Intermediate ภายใน Job ออกจากผลระหว่าง Jobs

ภายใน MapReduce job เดียว map output มักอยู่ local disk แล้ว Shuffle ไป Reducer ไม่ได้เขียน HDFS ทุก record แต่เมื่อ Reducer ของ job แรกจบและ job ที่สองต้องอ่านต่อ ผลลัพธ์ระหว่าง job มักถูกเขียนลง HDFS การพูดว่า “MapReduce เขียนผลกลางทุกอย่างลง HDFS” จึงกว้างเกินจริง เราควรแยกระหว่าง intermediate ภายใน job กับ materialized output ระหว่าง jobs

DAG เปิดทางให้เห็นทั้งแผน ไม่ใช่ทีละ Job

Directed Acyclic Graph หรือ DAG แสดง transformation และ dependency หลายขั้นในแผนเดียว Engine จึงแบ่ง stage ตามจุดที่ต้อง Shuffle, pipeline operator ที่แคบเข้าด้วยกัน และหลีกเลี่ยงการเขียนข้อมูลทนทานระหว่างทุกขั้นหากไม่จำเป็น อีกทั้งใช้ข้อมูลของทั้งแผนเพื่อ optimize ได้มากขึ้น

Transformation แบบ narrow dependency เช่น map/filter มักคำนวณ partition ปลายทางจาก partition ต้นทางจำนวนน้อย จึง pipeline ได้ง่าย ส่วน wide dependency เช่น groupByKey/join ต้องดึงข้อมูลจากหลาย partition และก่อ Shuffle เส้นแบ่งนี้สำคัญกว่าการจำชื่อ API เพราะบอกต้นทุนทางกายภาพ

Spark ไม่ได้เก็บทุกอย่างใน RAM โดยอัตโนมัติ

Spark สามารถ cache หรือ persist dataset ที่เลือกไว้เพื่อใช้ซ้ำ แต่ transformation ทั่วไปเป็น lazy และ partition ที่ไม่ถูก persist อาจถูก pipeline หรือคำนวณใหม่ตาม lineage หาก memory ไม่พอ ข้อมูลที่ persist อาจ spill ลงดิสก์หรือถูก evict ตาม storage level การกล่าวว่า Spark “เก็บผลกลางทั้งหมดใน memory” จึงทำให้เข้าใจผิด

งาน iterative เช่น machine learning หรือ graph algorithm ได้ประโยชน์มากเมื่อใช้ dataset เดิมหลายรอบและ cache ได้พอดี แต่งาน scan ครั้งเดียวที่ I/O-bound อาจไม่ได้เร็วกว่า MapReduce หลายสิบเท่าเสมอ ผลขึ้นกับ format, serialization, memory, skew, Shuffle และโค้ดผู้ใช้

Lazy Evaluation: รอก่อน เพื่อวางแผนให้ดีกว่า

ใน Spark การเรียก transformation เช่น map, filter หรือ select มักยังไม่ลงมืออ่านทุก record ทันที แต่สร้างคำอธิบายแผนไว้ จนเจอ action เช่น count, collect หรือ write จึงเริ่มทำงาน ความ “ขี้เกียจ” นี้ไม่ใช่การผัดวันประกันพรุ่ง แต่เปิดโอกาสให้ engine รวมขั้นตอน ตัดสิ่งที่ไม่ใช้ และวางแผน execution

Lazy evaluation เชื่อมกับ immutability และ lineage: dataset ใหม่อธิบายว่ามาจาก dataset เดิมอย่างไร แทนการแก้ก้อนเดิมแบบสุ่ม ระบบจึงเก็บ dependency เป็นกราฟ คำนวณเมื่อจำเป็น และสร้าง partition ที่หายใหม่ได้ แต่แนวคิดเหล่านี้ไม่ใช่สิ่งเดียวกัน—immutability พูดถึงการไม่แก้ค่าเดิม, laziness พูดถึงเวลาที่ลงมือคำนวณ, และ cache พูดถึงการเก็บผลที่คำนวณแล้วไว้ใช้ซ้ำ

Cache เป็นการตัดสินใจด้านต้นทุน

การ cache ทุกอย่างอาจทำให้ช้าลง เพราะเสียเวลา serialize ใช้ memory และเบียดข้อมูลที่มีประโยชน์กว่า ควร cache เมื่อ dataset มีราคาแพงในการสร้างและจะถูกใช้ซ้ำหลายครั้ง หลังใช้งานควร unpersist เพื่อคืนพื้นที่ หาก dataset ใช้ครั้งเดียว การ pipeline ใหม่อาจถูกกว่าเก็บ

จาก RDD ไปสู่ DataFrame และ SQL Engine

RDD เปิดให้เขียน function ได้ยืดหยุ่น แต่ engine มองความหมายภายใน function ได้จำกัด DataFrame และ SQL มี schema กับ operator ชัด จึงเปิดทางให้ optimizer เช่น Catalyst เลือกแผนเชิงตรรกะและกายภาพ รวมถึง code generation และ columnar execution ในระบบรุ่นใหม่ สิ่งนี้ทำให้เราเห็นวิวัฒนาการจาก “กระจายฟังก์ชัน” ไปสู่ “ประกาศการคำนวณที่ระบบวิเคราะห์ได้”

Object Storage กับการแยก Storage ออกจาก Compute

บน cloud หลายระบบเก็บข้อมูลระยะยาวใน object storage แล้วสร้าง compute cluster ตามต้องการ ข้อดีคือปรับขนาดและอายุของสองส่วนแยกกันได้ แต่ locality แบบ “task อยู่เครื่องเดียวกับ disk block” ลดลง Engine จึงใช้วิธีอื่น เช่น column pruning, predicate pushdown, partition pruning, caching, local SSD และการลดจำนวน request

Object storage ไม่ควรถูกอธิบายว่าเป็น HDFS ที่เปลี่ยนชื่อ แม้ทั้งคู่แบ่งและทำสำเนาข้อมูลภายใน แต่ interface, consistency semantics, metadata architecture, rename cost และรูปแบบการวาง compute ต่างกัน ความเข้าใจ GFS/HDFS ยังมีค่าเพราะสอนหลักการจัดเก็บแบบกระจาย แต่ต้องแยกหลักการร่วมออกจากรายละเอียดของแต่ละระบบ

แนวคิดเดิมที่ยังอยู่ในระบบใหม่

9ตัวอย่างออกแบบงานจากต้นจนจบ

สมมุติเรามี log การซื้อสินค้าหลายเทราไบต์ ต้องการยอดขายรวมต่อจังหวัดเฉพาะรายการที่ชำระสำเร็จ และต้องการเก็บจำนวน record เสียด้วย เราจะค่อย ๆ ตัดสินใจจาก storage ถึง output แทนการเริ่มเขียน Mapper ทันที

ขั้นที่ 1: ทำความเข้าใจ Input

ข้อมูลเป็น text, JSON, Avro หรือ Parquet? record แบ่งอ่านได้หรือไม่? มีไฟล์เล็กกี่ไฟล์? partition ตามวันหรือไม่? ถ้าเป็น columnar format เราอาจอ่านเฉพาะสามคอลัมน์และกรอง status ที่ storage layer ได้ ถ้าเป็น gzip ก้อนใหญ่ การขนานการอ่านอาจติดข้อจำกัดของ codec

ขั้นที่ 2: เลือก Key ตามคำถาม

คำถามต้องการยอดต่อจังหวัด ดังนั้น key กลางควรเป็น province และ value เป็น amount ไม่ควรใช้ record ทั้งก้อนเป็น value เพราะเพิ่ม Shuffle โดยไม่จำเป็น รายการที่ parse ไม่ได้ไม่ควรถูกปล่อยเงียบ ๆ แต่เพิ่ม counter และอาจเขียนตัวอย่างไปยัง output แยก

ขั้นที่ 3: รวมในเครื่องก่อน

ยอดรวมเป็น associative และมี identity 0 จึงใช้ Combiner หรือ partial aggregation ได้ Mapper แต่ละตัวส่งยอดย่อยต่อจังหวัด แทนการส่งทุกรายการซื้อ หากจังหวัดมีไม่มาก In-mapper Combining ก็อาจเหมาะและใช้หน่วยความจำคาดการณ์ได้

ขั้นที่ 4: ตรวจ Skew

หาก province กระจายค่อนข้างปกติ hash partitioner อาจเพียงพอ แต่ถ้าข้อมูลจำนวนมากมี province ว่าง key เดียวจะหนัก ควรแยก missing ออกเป็น data-quality output หากต้องนับจริง อาจ salt และรวมสองรอบ

ขั้นที่ 5: เลือกจำนวน Output Partition

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

ขั้นที่ 6: วัดผล ไม่เดาจากความรู้สึก

ดู input bytes, map output records, combine input/output records, shuffle bytes, spilled records, task duration distribution, counters ของข้อมูลเสีย และขนาด output หาก task ตัวท้ายช้ากว่ากลุ่มมาก ให้แยกว่ามาจากเครื่องช้าหรือ key skew ก่อนแก้

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

บทนี้เริ่มจากปัญหาว่าข้อมูลใหญ่เกินกว่าจะโยกกลับมาประมวลผลบนเครื่องเดียวอย่างคุ้มค่า GFS/HDFS จึงแบ่งไฟล์เป็น block กระจายสำเนา และเปิดให้ scheduler ส่งงานไปใกล้ข้อมูล MapReduce เพิ่มกรอบที่ทำให้ framework แบ่งงาน จัดกลุ่มข้อมูล retry และติดตามสถานะได้

หัวใจที่ควรติดตัวไปไม่ใช่การจำชื่อ class ของ Hadoop สิ่งสำคัญกว่าคือ:

• แยกหน่วยจัดเก็บ (block) หน่วยวางแผนงาน (split) และหน่วยข้อมูล (record) ออกจากกัน
• รู้ว่าการออกแบบ key กำหนดทั้งความหมายและเส้นทางของข้อมูล
• มอง Shuffle เป็นค่าใช้จ่ายหลัก และหาทางรวมผลในเครื่องก่อนส่ง
• ตรวจว่า operation มีโครงสร้างแบบ Monoid หรือปรับรูปด้วย monoidification ได้หรือไม่
• ออกแบบ function ให้ทำซ้ำได้ และอย่าวาง side effect ที่รับ retry ไม่ได้ไว้ใน task
• เข้าใจว่า SQL และ Spark ไม่ได้ลบต้นทุนทางกายภาพ แต่ช่วยวางแผนและซ่อนรายละเอียดที่ไม่จำเป็น

บทถัดไปจะลงไปที่ Scala และ Functional Programming เพื่อทำให้แนวคิด pure function, immutability, higher-order function, lazy evaluation และการรวมผลมีภาษาที่ชัดขึ้น เมื่อกลับมาดู Spark ในภายหลัง เราจะเห็นว่า API รุ่นใหม่ไม่ได้เกิดขึ้นลอย ๆ แต่ต่อยอดจากหลักการในบทนี้