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

Donald Knuth, “The Errors of TeX” (1989)

ที่ผ่านมา หนังสือเล่มนี้ส่วนใหญ่พูดถึง requests และ queries รวมถึง responses หรือ results ที่สอดคล้องกัน ระบบข้อมูลสมัยใหม่หลายระบบถูกออกแบบบนสมมติฐานแบบนี้: คุณส่ง request หรือคำสั่งเข้าไป แล้วระบบจะพยายามตอบกลับให้เร็วที่สุด

browser ที่ขอหน้าเว็บ, service ที่เรียก API ระยะไกล, database, cache, search index และระบบอื่นๆ อีกมากมายล้วนทำงานแบบนี้ เราเรียกสิ่งเหล่านี้ว่า online systems โดยปกติ response time คือตัววัดประสิทธิภาพหลัก และมักต้องการ fault tolerance เพื่อรับประกันความพร้อมใช้งานสูง

อย่างไรก็ตาม บางครั้งคุณจำเป็นต้องรันการคำนวณขนาดใหญ่หรือประมวลผลข้อมูลปริมาณมากกว่าที่คุณสามารถทำได้ใน request แบบโต้ตอบ บางทีคุณอาจต้องเทรน AI model หรือแปลงข้อมูลปริมาณมากจากรูปแบบหนึ่งไปเป็นอีกรูปแบบหนึ่ง หรือคำนวณ analytics บน dataset ที่มีขนาดใหญ่มาก เราเรียกงานเหล่านี้ว่า batch processing jobs และระบบที่จัดการกับงานเหล่านี้บางครั้งเรียกว่า offline systems

batch processing job จะรับ input data (ซึ่งเป็น read-only) และสร้าง output data (ซึ่งถูกสร้างขึ้นมาใหม่ทุกครั้งที่ job รัน) โดยทั่วไปแล้วมันจะไม่ mutate ข้อมูลแบบที่ read/write transaction ทำ Output จึงเป็น derived จาก input (ตามที่กล่าวถึงใน "Systems of Record and Derived Data" ) หากคุณไม่ชอบ output คุณสามารถลบทิ้ง ปรับ logic ของ job แล้วรัน job ใหม่ได้

การถือว่า input เป็น immutable และหลีกเลี่ยง side effect (เช่น การเขียนไปยัง external database) ทำให้ batch job ทำงานได้มีประสิทธิภาพ และยังมีข้อดีอื่นๆ ดังนี้:

• หากคุณนำบั๊กเข้าไปในโค้ดและ output นั้นผิดหรือเสียหาย คุณสามารถย้อนกลับไปใช้โค้ดเวอร์ชันก่อนหน้าและ rerun job ได้อย่างง่ายดาย และ output จะกลับมาถูกต้องอีกครั้ง หรือยิ่งง่ายกว่านั้น คุณสามารถเก็บ output เก่าไว้ใน directory อื่นแล้วสลับกลับไปใช้ได้ Object store และ open table formats ส่วนใหญ่ (ดู "Cloud Data Warehouses" ) รองรับฟีเจอร์นี้ ซึ่งเรียกว่า time travel database ส่วนใหญ่ที่รองรับ read/write transaction ไม่มีความสามารถนี้: หากคุณ deploy โค้ดที่มีบั๊กซึ่งเขียนข้อมูลเสียหายลงใน database การย้อนกลับโค้ดจะไม่สามารถแก้ไขข้อมูลนั้นได้ แนวคิดที่ว่าสามารถกู้คืนจากโค้ดที่มีบั๊กได้นี้เรียกว่า human fault tolerance [ 1 ]

• เพราะย้อนกลับได้ง่าย การพัฒนา feature จึงทำได้เร็วกว่าในสภาพแวดล้อมที่ความผิดพลาดอาจหมายถึงความเสียหายที่ย้อนกลับไม่ได้ หลักการ minimizing irreversibility นี้มีประโยชน์ต่อการพัฒนาซอฟต์แวร์แบบ Agile [ 2 ]

• ชุดไฟล์เดียวกันสามารถใช้เป็น input สำหรับงานประเภทต่างๆ ได้ รวมถึง monitoring jobs ที่คำนวณ metrics และประเมินว่า output ของ job มีคุณลักษณะตามที่คาดหวังหรือไม่ (เช่น โดยการเปรียบเทียบกับ output จากการรันครั้งก่อนหน้าและวัดความแตกต่าง)

• Batch processing framework ใช้ทรัพยากรคำนวณได้อย่างมีประสิทธิภาพ แม้จะ batch-process ข้อมูลผ่าน online data system เช่น OLTP database และ application server ได้ แต่การทำแบบนั้นอาจกินทรัพยากรมากกว่ามาก

Batch processing ได้พิสูจน์แล้วว่ามีประโยชน์ในกรณีการใช้งานที่หลากหลาย ซึ่งเราจะกล่าวถึงอีกครั้งใน "Batch Use Cases" อย่างไรก็ตาม มันก็มีความท้าทายเช่นกัน ใน framework ส่วนใหญ่ job อื่นจะนำ output ไปประมวลผลต่อได้ก็ต่อเมื่อ job ทั้งหมดเสร็จสิ้นเท่านั้น Batch processing ยังอาจไม่มีประสิทธิภาพ การเปลี่ยนแปลงใดๆ ใน input data แม้เพียง 1 byte ก็ทำให้ batch job ต้องประมวลผล input dataset ทั้งหมดใหม่

batch job อาจใช้เวลานานในการรัน: นาที ชั่วโมง หรือแม้แต่วัน Job อาจถูกตั้งเวลาให้รันเป็นระยะ (เช่น วันละครั้ง) ตัววัดประสิทธิภาพหลักมักจะเป็น throughput: ปริมาณข้อมูลที่ job สามารถประมวลผลได้ต่อหน่วยเวลา ระบบ batch บางระบบจัดการ fault โดยการยกเลิกและ restart job ทั้งหมด ในขณะที่บางระบบมี fault tolerance เพื่อให้ job สามารถเสร็จสมบูรณ์ได้แม้บาง node จะ crash

เส้นแบ่งระหว่าง online และ batch processing systems ไม่ได้ชัดเจนเสมอไป การ query ฐานข้อมูลที่ใช้เวลานานก็ดูคล้ายกับ batch process มาก แต่ batch processing ก็มีคุณลักษณะเฉพาะที่ทำให้มันเป็น building block ที่มีประโยชน์สำหรับการสร้าง application ที่ reliable, scalable, และ maintainable ตัวอย่างเช่น มันมักมีบทบาทใน data integration —การประกอบระบบข้อมูลหลายระบบเข้าด้วยกันเพื่อให้บรรลุสิ่งที่ระบบเดียวไม่สามารถทำได้ ETL ตามที่กล่าวถึงใน "Data Warehousing" เป็นตัวอย่างหนึ่งของเรื่องนี้

Note

ทางเลือกหนึ่งสำหรับ batch processing คือ stream processing ซึ่ง job จะไม่หยุดทำงานเมื่อประมวลผล input เสร็จ แต่จะคอยเฝ้าดู input ต่อเนื่อง และประมวลผลการเปลี่ยนแปลงของ input ไม่นานหลังจากที่เกิดขึ้น เราจะพูดถึง stream processing ใน Chapter 12 .

batch processing สมัยใหม่ได้รับอิทธิพลอย่างมากจาก MapReduce ซึ่งเป็น algorithm การประมวลผลแบบ batch ที่เผยแพร่โดย Google ในปี 2004 [ 3 ] และต่อมาถูกนำไป implement ในระบบข้อมูล open source ต่างๆ รวมถึง Hadoop, CouchDB, และ MongoDB MapReduce เป็น programming model ที่ค่อนข้างระดับต่ำ ซับซ้อนน้อยกว่า parallel query execution engine ที่พบใน data warehouse เป็นต้น [ 4 , 5 ] ในยุคที่ MapReduce ยังใหม่ มันเป็นก้าวสำคัญในแง่ของขนาดการประมวลผลที่ทำได้บน commodity hardware แต่ปัจจุบันล้าสมัยไปเป็นส่วนใหญ่และไม่ได้ถูกใช้ที่ Google อีกต่อไป [ 6 , 7 ]

ปัจจุบัน batch processing มักทำโดยใช้ frameworks เช่น Spark หรือ Flink หรือ data warehouse query engines เช่นเดียวกับ MapReduce พวกมันพึ่งพา sharding (ดู Chapter 7 ) และการประมวลผลแบบขนานอย่างมาก แต่มีกลยุทธ์การ caching และ execution ที่ซับซ้อนกว่ามาก เมื่อระบบเหล่านี้เติบโตเต็มที่ ปัญหาด้านการ operate ก็ถูกแก้ไปเป็นส่วนใหญ่ ความสนใจจึงเปลี่ยนไปที่ความง่ายในการใช้งาน โมเดลการประมวลผลใหม่เช่น dataflow APIs, query languages, และ DataFrame APIs ได้รับการสนับสนุนอย่างกว้างขวาง การ orchestrate job และ workflow ก็เติบโตเต็มที่เช่นกัน Hadoop-centric workflow schedulers เช่น Oozie และ Azkaban ถูกแทนที่ด้วย solution ที่ใช้ได้กว้างกว่า เช่น Airflow, Dagster, และ Prefect ซึ่งสนับสนุน batch processing frameworks และ cloud data warehouses ที่หลากหลาย

เมื่อ cloud computing แพร่หลาย storage layer ของ batch processing ก็กำลังเปลี่ยนจาก distributed filesystems (DFSs) เช่น HDFS (Hadoop Distributed File System), GlusterFS, และ CephFS ไปยัง object storage systems เช่น S3 Cloud data warehouses ที่ปรับขนาดได้ เช่น BigQuery และ Snowflake กำลังทำให้เส้นแบ่งระหว่าง data warehouses และ batch processing เลือนลาง

เพื่อให้เห็นภาพว่า batch processing คืออะไร เราจะเริ่มบทนี้ด้วยตัวอย่างที่ใช้เครื่องมือ Unix มาตรฐานบนเครื่องเดียว จากนั้นจะดูว่าขยายการประมวลผลข้อมูลไปยังหลายเครื่องใน distributed system ได้อย่างไร เราจะเห็นว่า distributed batch processing framework มี scheduler และ filesystem คล้ายกับ operating system มาก จากนั้นจะสำรวจโมเดลการประมวลผลต่างๆ ที่ใช้เขียน batch job สุดท้าย เราจะพูดถึงกรณีการใช้งาน batch processing ทั่วไป

Batch Processing with Unix Tools (การประมวลผลแบบแบตช์ด้วยเครื่องมือ Unix)

สมมติว่าคุณมี web server ที่เขียนต่อท้าย log file หนึ่งบรรทัดทุกครั้งที่รับ request ตัวอย่างเช่น การใช้รูปแบบ NGINX default access log หนึ่งบรรทัดของ log อาจมีลักษณะดังนี้:

216.58.210.78 - - [27/Jun/2025:17:55:11 +0000] "GET /css/typography.css HTTP/1.1"
200 3377 "https://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X
10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36

(นั่นคือหนึ่งบรรทัดจริงๆ มันถูกแบ่งเป็นหลายบรรทัดที่นี่เพื่อให้อ่านง่าย) มีข้อมูลมากมายในบรรทัดนั้น ในการตีความ คุณต้องดูที่นิยามของรูปแบบ log ซึ่งมีดังนี้:

$remote_addr - $remote_user [$time_local] "$request"
$status $body_bytes_sent "$http_referer" "$http_user_agent"

ดังนั้น log หนึ่งบรรทัดนี้บ่งชี้ว่าในวันที่ 27 มิถุนายน 2025 เวลา 17:55:11 UTC เซิร์ฟเวอร์ได้รับ request สำหรับไฟล์ /css/typography.css จาก IP address ของ client 216.58.210.78 ผู้ใช้ไม่ได้ผ่านการ authentication ดังนั้น $remote_user จึงถูกตั้งค่าเป็นยัติภังค์ ( - ) status ของ response คือ 200 (หมายถึง request สำเร็จ) และ response มีขนาด 3,377 bytes browser คือ Chrome 137 และมันโหลดไฟล์เพราะถูกอ้างอิงในหน้าที่ URL [_https://martin.kleppmann.com/](https://martin.kleppmann.com/)_ .

การ parse log อาจฟังดูเป็นแค่ตัวอย่างสมมติ แต่จริงๆ แล้วเป็นส่วนสำคัญในการทำงานของบริษัทเทคโนโลยีสมัยใหม่หลายแห่ง ตั้งแต่ ad pipeline ไปจนถึงการประมวลผลการชำระเงิน และเป็นแรงผลักดันให้ MapReduce และกระแส "big data" ถูกนำไปใช้อย่างรวดเร็ว

Simple Log Analysis (การวิเคราะห์ Log อย่างง่าย)

มีเครื่องมือต่างๆ ที่สามารถนำ log files เหล่านี้ไปสร้างรายงานที่สวยงามเกี่ยวกับ traffic บนเว็บไซต์ของคุณ แต่เพื่อการฝึกฝน มาสร้างของเราเองโดยใช้เครื่องมือ Unix พื้นฐาน ตัวอย่างเช่น สมมติว่าคุณต้องการห้าหน้าเว็บที่ได้รับความนิยมมากที่สุดบนเว็บไซต์ของคุณ คุณสามารถทำได้ใน Unix shell ดังนี้:

cat /var/log/nginx/access.log | 
  awk '{print $7}' | 
  sort             | 
  uniq -c          | 
  sort -r -n       | 
  head -n 5

อ่าน log file (พูดอย่างเคร่งครัด cat ไม่จำเป็นที่นี่ เพราะ input file สามารถระบุเป็น argument โดยตรงให้กับ awk ได้ แต่การเขียนเป็น pipeline แบบเส้นตรงจะเห็นภาพได้ชัดเจนกว่า)

แบ่งแต่ละบรรทัดออกเป็นฟิลด์ด้วย whitespace และแสดงเฉพาะฟิลด์ที่เจ็ดจากแต่ละบรรทัด ซึ่งก็คือ URL ที่ถูกร้องขอ ในตัวอย่างของเรา URL นี้คือ /css/typography.css .

เรียงลำดับตามตัวอักษร sort รายการของ URL ที่ถูกร้องขอ เหตุผลที่ต้อง sort คือเพื่อให้แน่ใจว่า ถ้า URL ถูกขอ n ครั้ง ไฟล์ที่ sort แล้วจะมี URL เดียวกันซ้ำกัน n ครั้งติดต่อกัน

คำสั่ง uniq กรองบรรทัดที่ซ้ำกันใน input โดยการตรวจสอบว่าสองบรรทัดที่ติดกันเหมือนกันหรือไม่ option -c บอกให้มันแสดง counter ด้วย: สำหรับทุกๆ URL ที่แตกต่างกัน มันจะรายงานว่า URL นั้นปรากฏกี่ครั้งใน input

sort ครั้งที่สองจะเรียงลำดับตามตัวเลข ( -n ) ที่จุดเริ่มต้นของแต่ละบรรทัด ซึ่งคือจำนวนครั้งที่ URL ถูกขอ จากนั้นจะแสดงผลลัพธ์ในลำดับย้อนกลับ ( -r ) — โดยนำตัวเลขมากที่สุดขึ้นก่อน

สุดท้าย head จะแสดงเฉพาะห้าบรรทัดแรก ( -n 5 ) ของ input และทิ้งส่วนที่เหลือ

output ของชุดคำสั่งนั้นจะมีลักษณะประมาณนี้:

4189 /favicon.ico
3631 /2016/02/08/how-to-do-distributed-locking.html
2124 /2020/11/18/distributed-systems-and-elliptic-curves.html
1369 /
 915 /css/typography.css

คำสั่งบรรทัดด้านบนอาจอ่านยากถ้าคุณไม่คุ้นกับเครื่องมือ Unix แต่ทรงพลังมาก มันจะประมวลผล log files ขนาดหลาย gigabytes ภายในไม่กี่วินาที และคุณสามารถปรับเปลี่ยนการวิเคราะห์ให้เหมาะกับความต้องการของคุณได้อย่างง่ายดาย ตัวอย่างเช่น หากคุณต้องการละเว้นไฟล์ CSS จากรายงาน คุณสามารถเปลี่ยน awk argument เป็น $7 !~ /\.css$/ {print $7} และหากคุณต้องการนับ top client IP addresses แทนที่จะเป็น top pages คุณสามารถเปลี่ยน awk argument เป็น {print $1} เป็นต้น

หนังสือเล่มนี้ไม่มีที่พอจะอธิบายเครื่องมือ Unix โดยละเอียด แต่ก็คุ้มค่าที่จะเรียนรู้ การวิเคราะห์ข้อมูลจำนวนมากสามารถทำได้ภายในไม่กี่นาทีโดยใช้การผสมผสานของ awk , sed , grep , sort , uniq , และ xargs และพวกมันทำงานได้ดีอย่างน่าประหลาดใจ [ 8 ]

Chain of Commands Versus Custom Program (ห่วงโซ่คำสั่งเทียบกับโปรแกรมที่กำหนดเอง)

แทนที่ จะใช้ห่วงโซ่คำสั่ง Unix คุณสามารถเขียนโปรแกรมง่ายๆ เพื่อทำสิ่งเดียวกันได้ ตัวอย่างเช่น ใน Python มันอาจมีลักษณะดังนี้:

from collections import defaultdict

counts = defaultdict(int) 

with open('/var/log/nginx/access.log', 'r') as file:
    for line in file:
        url = line.split()[6] 
        counts[url] += 1 

top5 = sorted(((count, url) for url, count in counts.items()),
              reverse=True)[:5] 

for count, url in top5:  
    print(f"{count} {url}")

เริ่มต้น counts ให้เป็น hash table ที่เก็บ counter สำหรับจำนวนครั้งที่เราเห็นแต่ละ URL ค่าเริ่มต้นของแต่ละ counter คือ 0

รับ URL ที่ถูกร้องขอ ซึ่งเป็นฟิลด์ที่เจ็ดที่คั่นด้วย whitespace จากแต่ละบรรทัดของ log (index ของ array คือ 6 เพราะ array ของ Python นับจากศูนย์)

เพิ่ม counter สำหรับ URL ในบรรทัดปัจจุบันของ log

เรียงลำดับเนื้อหาของ hash table ตามค่า counter (จากมากไปน้อย) และนำห้ารายการแรก

แสดงห้ารายการแรกนั้นออกมา

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

Sorting Versus In-Memory Aggregation (การเรียงลำดับเทียบกับการรวมในหน่วยความจำ)

สคริปต์ Python เก็บ hash table ของ URL ไว้ในหน่วยความจำ โดย map แต่ละ URL ไปยังจำนวนครั้งที่พบ ส่วน Unix pipeline ไม่มี hash table แบบนั้น แต่อาศัยการ sort รายการ URL เพื่อให้ URL เดียวกันที่ซ้ำกันมาเรียงติดกัน

วิธีไหนดีกว่า? ขึ้นอยู่กับว่ามี URL กี่รายการ สำหรับเว็บไซต์ขนาดเล็กถึงกลางส่วนใหญ่ คุณอาจเก็บ URL ที่ไม่ซ้ำกันทั้งหมดพร้อม counter ของแต่ละ URL ไว้ในหน่วยความจำ (สมมติ) 1 GB ได้ ในตัวอย่างนี้ working set ของ job (ปริมาณหน่วยความจำที่ job ต้องการเข้าถึงแบบสุ่ม) ขึ้นอยู่กับจำนวน URL ที่ไม่ซ้ำกันเท่านั้น ถ้ามี log หนึ่งล้านรายการของ URL เดียว hash table ก็ยังใช้พื้นที่แค่ URL นั้นหนึ่งรายการบวกกับ counter ถ้า working set นี้เล็กพอ hash table ในหน่วยความจำก็ทำงานได้ดี—แม้แต่บน laptop

ในทางกลับกัน ถ้า working set ของ job มีขนาดใหญ่กว่าหน่วยความจำที่มีอยู่ วิธี sort ได้เปรียบตรงที่ใช้ disk ได้อย่างมีประสิทธิภาพ ซึ่งเป็นหลักการเดียวกับที่กล่าวถึงใน "Log-Structured Storage" : sort ข้อมูลทีละ chunk ในหน่วยความจำแล้วเขียนลง disk เป็น segment file จากนั้น merge segment ที่ sort แล้วหลายๆ ไฟล์เป็นไฟล์ที่ใหญ่ขึ้น Mergesort เข้าถึงข้อมูลแบบ sequential ซึ่งทำงานได้ดีบน disk (ดู "Sequential Versus Random Writes on SSDs" )

utility sort ใน GNU Coreutils (Linux) จัดการ dataset ที่ใหญ่กว่าหน่วยความจำให้อัตโนมัติด้วยการ spill ลง disk และ sort แบบขนานข้ามหลาย CPU core ให้เอง [ 9 ] ซึ่งหมายความว่าห่วงโซ่คำสั่ง Unix ง่ายๆ ที่เห็นก่อนหน้านี้ scale ไปยัง dataset ขนาดใหญ่ได้โดยไม่ทำให้หน่วยความจำหมด คอขวดน่าจะเป็นความเร็วในการอ่าน input file จาก disk

ข้อจำกัดของเครื่องมือ Unix คือทำงานได้บนเครื่องเดียว Dataset ที่ใหญ่เกินกว่าจะใส่ในหน่วยความจำหรือ local disk จะกลายเป็นปัญหา และตรงนี้เองที่ distributed batch processing framework เข้ามามีบทบาท

Batch Processing in Distributed Systems (การประมวลผลแบบแบตช์ในระบบกระจาย)

เครื่องที่รันตัวอย่างเครื่องมือ Unix ของเรามีส่วนประกอบหลายอย่างที่ทำงานร่วมกันเพื่อประมวลผล log data:

• อุปกรณ์จัดเก็บข้อมูลที่เข้าถึงผ่านอินเทอร์เฟซ filesystem ของ operating system

• Scheduler ที่กำหนดว่า process จะได้รันเมื่อใดและจะจัดสรรทรัพยากร CPU ให้กับ process อย่างไร

• ชุดของโปรแกรม Unix ที่ standard input และ standard output ( stdin และ stdout ) เชื่อมต่อกันด้วย pipes

distributed data processing framework ก็มีส่วนประกอบเหล่านี้เช่นกัน จะมองว่า framework เหล่านี้คือ distributed operating system ก็ได้ เพราะมี filesystem, job scheduler และโปรแกรมที่ส่งข้อมูลถึงกันผ่าน filesystem หรือช่องทางสื่อสารอื่นๆ

Distributed Filesystems (ระบบไฟล์แบบกระจาย)

ระบบ filesystem ที่ operating system ของคุณมีให้ประกอบด้วยหลายชั้น:

• ชั้นล่างสุดคือ block device driver ที่คุยกับ disk โดยตรง และให้ชั้นที่อยู่เหนือขึ้นไปอ่านเขียน raw block ได้

• เหนือชั้น block คือ page cache ที่เก็บ block ที่เพิ่งเข้าถึงไว้ในหน่วยความจำ เพื่อให้เข้าถึงได้เร็วขึ้น

• block API ถูกห่อไว้ด้วยชั้น filesystem ซึ่งแบ่งไฟล์ขนาดใหญ่เป็น block และติดตาม metadata ของไฟล์ เช่น inode, directory และ file ตัวอย่าง implementation ที่พบบ่อยบน Linux คือ ext4 และ XFS

• สุดท้าย operating system เปิดให้ application ใช้ filesystem ต่างๆ ผ่าน API กลางที่เรียกว่า virtual filesystem (VFS) VFS ทำให้ application อ่านเขียนข้อมูลด้วยวิธีเดียวกันได้ ไม่ว่า filesystem เบื้องล่างจะเป็นแบบใด

DFS ทำงานคล้ายกันมาก ไฟล์ถูกแบ่งเป็น block ที่กระจายอยู่บนหลายเครื่อง DFS block มักใหญ่กว่า local block มาก HDFS ใช้ค่าเริ่มต้น 128 MB ส่วน JuiceFS และ object store หลายตัวใช้ block ขนาด 4 MB ซึ่งก็ยังใหญ่กว่า 4,096 bytes ของ ext4 มาก block ที่ใหญ่ขึ้นหมายถึงมี metadata ให้ติดตามน้อยลง ซึ่งช่วยได้มากกับ dataset ระดับหลาย petabyte และยังลด overhead ของการ seek ไปยัง block เมื่อเทียบกับเวลาที่ใช้อ่าน block นั้น

อุปกรณ์จัดเก็บข้อมูลทางกายภาพส่วนใหญ่เขียน block ไม่เต็มไม่ได้ operating system จึงกำหนดให้การ write ต้องใช้ทั้ง block แม้ข้อมูลจะไม่เต็ม block ก็ตาม แต่ distributed filesystem มี block ที่ใหญ่กว่าและมักสร้างอยู่บน filesystem ของ operating system จึงไม่มีข้อกำหนดนี้ ตัวอย่างเช่น ไฟล์ 900 MB ที่เก็บด้วย block ขนาด 128 MB จะมี 7 block ที่ใช้ 128 MB และ 1 block ที่ใช้ 4 MB

การอ่าน DFS block ทำโดยส่ง network request ไปยังเครื่องใน cluster ที่เก็บ block นั้น แต่ละเครื่องรัน daemon ที่เปิด API ให้ process ระยะไกลอ่านและเขียน block ซึ่งเก็บเป็นไฟล์บน local filesystem ของเครื่อง HDFS เรียก daemon เหล่านี้ว่า DataNode ส่วน GlusterFS เรียก process นี้ว่า glusterfsd เราจะเรียกมันว่า data nodes ในหนังสือเล่มนี้

DFS ยังมี page cache แบบกระจายด้วย เพราะ DFS block ถูกเก็บเป็นไฟล์บน data node การอ่านและเขียนจึงผ่าน operating system ของแต่ละ data node ซึ่งรวมถึง in-memory page cache ด้วย ทำให้ block ของข้อมูลที่ถูกอ่านบ่อยๆ อยู่ในหน่วยความจำบน data node DFS บางระบบยังมี caching tier เพิ่มเติม เช่น client-side cache และ local disk cache ที่พบใน JuiceFS

filesystem อย่าง ext4 และ XFS เก็บ metadata ของการจัดเก็บ เช่น พื้นที่ว่าง ตำแหน่ง block ของไฟล์ โครงสร้าง directory การตั้งค่าสิทธิ์ และอื่นๆ DFS ก็ต้องมีวิธีติดตามว่าไฟล์ที่กระจายอยู่บนหลายเครื่องอยู่ที่ไหน รวมถึงการตั้งค่าสิทธิ์และอื่นๆ เช่นกัน Hadoop มี service ชื่อ NameNode ที่ดูแล metadata ของทั้ง cluster ส่วน 3FS ของ DeepSeek มี metadata service ที่เก็บข้อมูลลงใน key-value store เช่น FoundationDB

ใน operating system VFS อยู่เหนือ filesystem ส่วนสิ่งที่เทียบได้ใน batch processing คือ protocol ของ distributed filesystem DFS ต้องมี protocol หรือ interface ให้ batch processing system อ่านและเขียนไฟล์ได้ protocol นี้ทำหน้าที่เป็น pluggable interface คือจะใช้ DFS ตัวไหนก็ได้ ขอแค่ implement protocol นั้น ตัวอย่างเช่น API ของ Amazon S3 ถูกระบบจัดเก็บข้อมูลอย่าง MinIO, Cloudflare's R2, Tigris, Backblaze's B2 และอื่นๆ อีกมากมาย implement ตามกันอย่างกว้างขวาง Batch processing system ที่รองรับ S3 จึงใช้ระบบจัดเก็บข้อมูลเหล่านี้ได้ทั้งหมด

บาง DFS implement filesystem ที่ compliant กับ POSIX ซึ่งปรากฏต่อ VFS ของ operating system เหมือน filesystem อื่นๆ โดยมักใช้ Filesystem in Userspace (FUSE) หรือ Network File System (NFS) protocol เพื่อเชื่อมเข้ากับ VFS NFS อาจเป็น distributed filesystem protocol ที่รู้จักกันดีที่สุด และถูกพัฒนาขึ้นเพื่อให้ client หลายตัวอ่านและเขียนข้อมูลบน server เดียวได้ เมื่อไม่นานมานี้ filesystem อย่าง Amazon Elastic File System (EFS) และ Archil ได้ออก implementation ของ distributed filesystem ที่เข้ากันได้กับ NFS และ scale ได้มากกว่า NFS client ยังคงเชื่อมต่อกับ endpoint เดียว แต่เบื้องหลัง ระบบเหล่านี้จะสื่อสารกับ distributed metadata service และ data node เพื่ออ่านและเขียนข้อมูล

Distributed Filesystems and Network Storage (ระบบไฟล์แบบกระจายและพื้นที่จัดเก็บเครือข่าย)

Distributed filesystems อยู่บนพื้นฐานของหลักการ shared-nothing (ดู "Shared-Memory, Shared-Disk, and Shared-Nothing Architectures" ) ซึ่งต่างจากแนวทาง shared-disk ของ architecture แบบ network attached storage (NAS) และ storage area network (SAN) Shared-disk storage ถูก implement ด้วย centralized storage appliance ซึ่งมักใช้ฮาร์ดแวร์เฉพาะและโครงสร้างพื้นฐานเครือข่ายพิเศษอย่าง Fibre Channel ในทางกลับกัน แนวทาง shared-nothing ไม่ต้องใช้ฮาร์ดแวร์พิเศษ ขอแค่คอมพิวเตอร์ที่เชื่อมต่อกันด้วยเครือข่ายใน datacenter ทั่วไป

Distributed filesystem หลายระบบสร้างบน commodity hardware ซึ่งราคาถูกกว่าแต่มีอัตราความล้มเหลวสูงกว่าฮาร์ดแวร์ระดับองค์กร เพื่อทนต่อความล้มเหลวของเครื่องและ disk block ของไฟล์จะถูก replicate ไว้บนหลายเครื่อง วิธีนี้ยังช่วยให้ scheduler กระจายภาระงานได้สม่ำเสมอขึ้นด้วย เพราะสามารถรันงานบน node ใดก็ได้ที่มี replica ของ input data ของงานนั้น

replication อาจหมายถึงการเก็บสำเนาของข้อมูลเดียวกันหลายชุดบนหลายเครื่อง ตามที่อธิบายใน Chapter 6 หรือการใช้ erasure coding scheme เช่น Reed–Solomon codes ซึ่งช่วยให้กู้คืนข้อมูลที่สูญหายได้ด้วย storage overhead ที่ต่ำกว่า full replication [ 10 , 11 , 12 ] เทคนิคเหล่านี้คล้ายกับ RAID ซึ่งสร้างความซ้ำซ้อนข้าม disk หลายตัวที่ต่อกับเครื่องเดียวกัน ส่วนที่ต่างคือใน distributed filesystem การเข้าถึงไฟล์และการ replicate ทำผ่านเครือข่าย datacenter ทั่วไป โดยไม่ต้องใช้ฮาร์ดแวร์พิเศษ

Object Stores (พื้นที่จัดเก็บออบเจ็กต์)

บริการ object storage เช่น Amazon S3, Google Cloud Storage, Azure Blob Storage, และ OpenStack Swift กลายเป็นทางเลือกยอดนิยมแทน distributed filesystems สำหรับ batch processing jobs อันที่จริง เส้นแบ่งระหว่างทั้งสองค่อนข้างเลือนลาง ดังที่เราเห็นในส่วนก่อนหน้าและ "Databases Backed by Object Storage" FUSE driver ช่วยให้ใช้ object store อย่าง S3 เหมือนเป็น filesystem ได้ DFS บางตัว เช่น JuiceFS และ Ceph มีให้ทั้ง object storage API และ filesystem API อย่างไรก็ตาม API, performance และการรับประกันด้านความสอดคล้องของข้อมูลต่างกันมาก จึงต้องระวังเมื่อนำระบบแบบนี้มาใช้ และตรวจให้แน่ใจว่าทำงานตามที่คาดหวัง แม้จะดูเหมือน implement API ที่ต้องการครบแล้วก็ตาม

object แต่ละตัวใน object store มี URL เช่น s3://my-photo-bucket/2025/04/01​/⁠birthday.png ส่วน host ของ URL ( my-photo-bucket ) ระบุ bucket ที่เก็บ object และส่วนที่ตามมาคือ key ของ object ( /2025/04/01​/⁠birthday.png ในตัวอย่างของเรา) ชื่อ bucket ต้องไม่ซ้ำกันทั่วโลก และ key ของแต่ละ object ต้องไม่ซ้ำกันภายใน bucket เดียวกัน

เราอ่าน object ด้วยการเรียก get และเขียนด้วยการเรียก put ต่างจากไฟล์บน filesystem ตรงที่ object แก้ไขไม่ได้เมื่อเขียนแล้ว ถ้าจะอัปเดต object ต้องเขียนทับทั้งตัวด้วยการเรียก put คล้ายกับ key-value store Azure Blob Storage และ S3 Express One Zone รองรับการ append แต่ store อื่นๆ ส่วนใหญ่ไม่รองรับ และไม่มี file handle API ที่มีฟังก์ชันอย่าง fopen และ fseek

object อาจดูเหมือนถูกจัดเป็น directory ซึ่งชวนสับสน เพราะ object store ไม่มี concept ของ directory โครงสร้าง path เป็นเพียงข้อตกลง และเครื่องหมาย slash ก็เป็นส่วนหนึ่งของ key ของ object ข้อตกลงนี้ทำให้ทำสิ่งที่คล้ายการ list directory ได้ โดยขอรายชื่อ object ที่มี prefix ที่ระบุ แต่การ list object ด้วย prefix ต่างจากการ list directory ของ filesystem ในสองด้าน:

• การ list แบบ prefix ทำงานเหมือนการเรียก ls -R แบบ recursive บน Unix คือจะคืน object ทั้งหมดที่ขึ้นต้นด้วย prefix รวมถึง object ใน subpath ด้วย

• directory ว่างเปล่าไม่มีอยู่จริง ถ้าคุณลบ object ทั้งหมดภายใต้ s3://my-photo-bucket/2025/04/01 แล้ว 01 จะไม่โผล่ขึ้นมาอีกเมื่อคุณเรียก list บน s3://my-photo-bucket/2025/04 วิธีที่ทำกันทั่วไปคือสร้าง object ขนาดศูนย์ไบต์ไว้แทน directory ว่าง (เช่น สร้างไฟล์ว่าง s3://my-photo-bucket/2025/04/01 เพื่อให้ path นั้นยังอยู่แม้ object ลูกทั้งหมดถูกลบไปแล้ว)

DFS มักรองรับการทำงานของ filesystem ทั่วไป เช่น hard link, symbolic link, file locking และ atomic rename ส่วน object store ไม่มีความสามารถเหล่านี้ โดยมักไม่รองรับ linking และ locking ส่วน rename ก็ไม่ atomic เพราะทำด้วยการ copy object ไปยัง key ใหม่แล้วลบ object เก่า ถ้าต้องการ rename directory ก็ต้อง rename ทุก object ข้างในทีละตัว เพราะชื่อ directory เป็นส่วนหนึ่งของ key

Key-value store ที่กล่าวถึงใน Chapter 4 ถูก optimize ให้เหมาะกับค่าขนาดเล็ก (โดยทั่วไปเป็น kilobytes) และการอ่าน/เขียนบ่อยๆ ที่ latency ต่ำ ส่วน distributed filesystem และ object store โดยทั่วไปถูก optimize ให้เหมาะกับ object ขนาดใหญ่ (megabytes ถึง gigabytes) และการอ่านที่ก้อนใหญ่กว่าแต่ไม่บ่อยเท่า อย่างไรก็ตาม เมื่อเร็วๆ นี้ object store เริ่มรองรับการอ่าน/เขียนที่เล็กลงและถี่ขึ้น ตัวอย่างเช่น S3 Express One Zone ตอนนี้มี latency เพียงไม่กี่มิลลิวินาที และมีรูปแบบราคาที่คล้าย key-value store มากขึ้น

อีกความต่างระหว่าง distributed filesystem กับ object store คือ DFS อย่าง HDFS ให้งานคำนวณรันบนเครื่องที่เก็บสำเนาของไฟล์นั้นอยู่ ทำให้ task อ่านไฟล์ได้โดยไม่ต้องส่งผ่านเครือข่าย ซึ่งประหยัด bandwidth ถ้า executable code ของ task มีขนาดเล็กกว่าไฟล์ที่ต้องอ่าน ส่วน object store มักแยก storage ออกจาก computation การทำแบบนี้อาจใช้ bandwidth มากกว่า แต่เครือข่าย datacenter สมัยใหม่เร็วมาก จึงมักยอมรับได้ architecture นี้ยังทำให้ปรับ scale ทรัพยากรของเครื่องอย่าง CPU และ memory ได้อิสระจาก storage เพราะทั้งสองถูกแยกออกจากกัน

Distributed Job Orchestration (การจัดประสานงาน Job แบบกระจาย)

การเปรียบเทียบกับ operating system ใช้กับการ orchestrate job ได้เช่นกัน เมื่อคุณรัน Unix batch job จะมี awk , sort , uniq , และ head process ที่ต้องรันจริงๆ ข้อมูลต้องถูกส่งจาก output ของ process หนึ่งไปเป็น input ของอีก process หนึ่ง หน่วยความจำต้องถูกจัดสรรให้แต่ละ process คำสั่งของแต่ละ process ต้องถูก schedule อย่างยุติธรรมแล้วรันบน CPU ต้องบังคับใช้ขีดจำกัดของหน่วยความจำและ I/O และอื่นๆ บนเครื่องเดียว หน้าที่เหล่านี้เป็นของ kernel ของ operating system ส่วนในสภาพแวดล้อมแบบกระจาย นี่คือบทบาทของ job orchestrator

Batch processing framework ส่ง request ไปยัง scheduler ของ orchestrator เพื่อรัน job โดย request สำหรับเริ่ม job จะมี metadata เช่น:

• จำนวน task ที่จะรัน

• ปริมาณ memory, CPU และ disk ที่แต่ละ task ต้องการ

• Job identifier

• ข้อมูลรับรองการเข้าถึง

• parameter ของ job เช่น input และ output data

• ฮาร์ดแวร์ที่ต้องการ เช่น GPU หรือประเภท disk

• ตำแหน่งของ executable code ของ job

Orchestrator เช่น Kubernetes และ Hadoop YARN (Yet Another Resource Negotiator) [ 13 ] นำข้อมูลนี้ไปรวมกับ cluster metadata เพื่อรัน job โดยใช้ส่วนประกอบต่อไปนี้:

Task executors

Executor daemon เช่น NodeManager ของ YARN หรือ kubelet ของ Kubernetes รันอยู่บนทุก node ใน cluster executor มีหน้าที่รัน task ของ job, ส่ง heartbeat เพื่อบอกว่ายังทำงานอยู่ และติดตามสถานะ task กับการจัดสรรทรัพยากรบน node เมื่อ executor ได้รับ request ให้เริ่ม task มันจะดึง executable code ของ job มา แล้วรันคำสั่งเพื่อเริ่ม task จากนั้นจะติดตาม process นั้นจนกว่าจะเสร็จหรือล้มเหลว แล้วอัปเดต metadata สถานะของ task ตามผลที่ได้

executor หลายตัวยังทำงานร่วมกับ operating system เพื่อแยก task ออกจากกันทั้งด้านความปลอดภัยและประสิทธิภาพ YARN และ Kubernetes ใช้ Linux cgroups เป็นต้น เพื่อป้องกันไม่ให้ task เข้าถึงข้อมูลโดยไม่ได้รับอนุญาต หรือใช้ทรัพยากรมากเกินไปจนกระทบประสิทธิภาพของ task อื่นบน node เดียวกัน

Resource manager

Resource manager ของ orchestrator เก็บ metadata ของแต่ละ node เช่น ฮาร์ดแวร์ที่พร้อมใช้ (CPU, GPU, memory, disk และอื่นๆ), สถานะของ task, ตำแหน่งในเครือข่าย, สถานะของ node และข้อมูลอื่นๆ ที่เกี่ยวข้อง ด้วยเหตุนี้ manager จึงเห็นภาพรวมสถานะปัจจุบันของทั้ง cluster ความที่ resource manager เป็นศูนย์กลางอาจกลายเป็นคอขวดทั้งด้าน scalability และ availability YARN ใช้ ZooKeeper และ Kubernetes ใช้ etcd เก็บสถานะ cluster (ดู "Coordination Services" )

Scheduler

Orchestrator มักมี centralized scheduler subsystem ที่รับ request ให้เริ่ม หยุด หรือตรวจสอบสถานะของ job ตัวอย่างเช่น scheduler อาจได้รับ request ให้เริ่ม job ที่มี 10 task โดยใช้ Docker image ที่ระบุ บน node ที่มี GPU ชนิดหนึ่ง scheduler ใช้ข้อมูลจาก request และสถานะจาก resource manager เพื่อกำหนดว่า task ไหนจะรันบน node ไหน จากนั้นจะแจ้งงานที่มอบหมายให้ task executor แล้ว executor ก็เริ่มรัน

แม้แต่ละ orchestrator จะใช้ศัพท์ต่างกัน แต่คุณจะพบส่วนประกอบเหล่านี้ในระบบ orchestration เกือบทั้งหมด

Note

บางครั้งการตัดสินใจเรื่อง scheduling ต้องใช้ application-specific scheduler ที่คำนึงถึงข้อกำหนดเฉพาะได้ เช่น autoscale read replica เมื่อ query ถึงเกณฑ์ที่กำหนด centralized scheduler และ application-specific scheduler จะทำงานร่วมกันเพื่อหาวิธีรัน task ที่ดีที่สุด YARN เรียก subscheduler ของมันว่า ApplicationMasters ในขณะที่ Kubernetes เรียกพวกมันว่า operators .

Resource allocation (การจัดสรรทรัพยากร)

Scheduler มีบทบาทที่ท้าทายเป็นพิเศษใน job orchestration เพราะต้องหาวิธีแบ่งทรัพยากรที่มีจำกัดของ cluster ให้ job ที่แย่งทรัพยากรกันอยู่อย่างเหมาะสมที่สุด โดยพื้นฐานแล้ว การตัดสินใจเหล่านี้ต้องสร้างสมดุลระหว่างความยุติธรรมกับประสิทธิภาพ

ลองนึกภาพ cluster เล็กๆ ที่มีห้า node รวมกันมี CPU core พร้อมใช้งาน 160 core scheduler ของ cluster ได้รับ request สอง job และแต่ละ job ต้องใช้ 100 core จึงจะทำงานเสร็จ วิธีที่ดีที่สุดในการ schedule ภาระงานคืออะไร?

• Scheduler อาจตัดสินใจรัน 80 task ให้แต่ละ job แล้วเริ่ม 20 task ที่เหลือของแต่ละ job เมื่อ task ก่อนหน้าเสร็จ

• Scheduler อาจรัน task ทั้งหมดของ job หนึ่งก่อน แล้วค่อยเริ่มรัน task ของ job ที่สองเมื่อมี 100 core ว่างเท่านั้น (กลยุทธ์ที่เรียกว่า gang scheduling )

• ถ้า request ของ job ที่สองมาถึงช้ากว่า job แรกมาก scheduler จะมีข้อมูลไม่ครบ และต้องตัดสินใจว่าจะจัดสรร 100 core ทั้งหมดให้ job แรก หรือสงวนบางส่วนไว้เผื่อ job ในอนาคตที่อาจมาหรืออาจไม่มาเลยก็ได้

นี่เป็นตัวอย่างที่ง่ายมาก แต่ก็เห็น trade-off ที่ยากหลายอย่างแล้ว เช่น ในกรณี gang scheduling ถ้า scheduler สงวน CPU core ไว้จนกว่าจะมีครบ 100 core พร้อมกัน node จะว่างอยู่เฉยๆ การใช้ทรัพยากรของ cluster จะลดลง และอาจเกิด deadlock ได้ถ้า job อื่นพยายามสงวน CPU core เช่นกัน

ในทางกลับกัน ถ้า scheduler แค่รอให้มี 100 core ว่าง job อื่นอาจแย่ง core ไปในระหว่างนั้น cluster อาจไม่มี 100 core ว่างเป็นเวลานานมาก ซึ่งนำไปสู่ starvation หรืออีกทางเลือกหนึ่ง scheduler อาจตัดสินใจ preempt task บางตัวของ job แรก โดยฆ่า task เหล่านั้นเพื่อเปิดที่ว่างให้ job ที่สอง อย่างไรก็ตาม การ preempt task ก็ทำให้ cluster ทำงานได้ไม่เต็มประสิทธิภาพเช่นกัน เพราะ task ที่ถูกฆ่าต้องถูก restart ภายหลัง

ทีนี้ลองนึกภาพ scheduler ที่ต้องตัดสินใจจัดสรรทรัพยากรให้ request ของ job แบบนี้หลายร้อยหรือหลายล้านรายการ การหาคำตอบที่ดีที่สุดดูเหมือนจะเป็นไปไม่ได้ จริงๆ แล้วปัญหานี้ เป็น NP-hard ซึ่งหมายความว่าการคำนวณหาคำตอบที่ดีที่สุดช้าเกินกว่าจะใช้งานได้จริง ยกเว้นกับกรณีที่เล็กที่สุด [ 14 , 15 ]

ในทางปฏิบัติ scheduler จึงใช้ heuristic เพื่อตัดสินใจแบบที่อาจไม่ดีที่สุดแต่ก็สมเหตุสมผล มี algorithm หลายตัวที่ใช้กันทั่วไป เช่น first-in first-out (FIFO), dominant resource fairness (DRF), priority queue, capacity หรือ quota-based scheduling และ bin-packing algorithm ต่างๆ รายละเอียดของ algorithm เหล่านี้อยู่นอกขอบเขตของหนังสือเล่มนี้ แต่เป็นหัวข้อวิจัยที่น่าสนใจ

Scheduling workflows (การจัดตารางเวิร์กโฟลว์)

ตัวอย่าง เครื่องมือ Unix ใน "Simple Log Analysis" เกี่ยวข้องกับคำสั่งหลายคำสั่งที่ต่อกันเป็นห่วงโซ่ด้วย Unix pipe รูปแบบเดียวกันนี้พบได้ใน distributed batch process: output ของ job หนึ่งมักต้องกลายเป็น input ของอีก job หนึ่งหรือหลาย job และแต่ละ job ก็อาจมี input หลายตัวที่มาจาก job อื่นๆ สิ่งนี้เรียกว่า workflow หรือ directed acyclic graph (DAG) ของ jobs

Note

ใน "Durable Execution and Workflows" เราได้เห็น workflow engine ที่ให้ durable execution ของลำดับขั้นตอนต่างๆ ซึ่งโดยทั่วไปจะทำ RPC แต่ในบริบทของ batch processing คำว่า "workflow" มีความหมายต่างออกไป คือลำดับของ batch process ที่แต่ละ process รับ input data และสร้าง output data โดยปกติจะไม่ทำ RPC ไปยัง service ภายนอก durable execution engine โดยทั่วไปประมวลผลข้อมูลต่อ request น้อยกว่า batch processing แม้เส้นแบ่งจะค่อนข้างเลือนลางก็ตาม

Workflow ที่ประกอบด้วยหลาย job อาจจำเป็นด้วยเหตุผลหลายประการ:

• ถ้า output ของ job หนึ่งต้องเป็น input ของ job อื่นหลายตัวที่ดูแลโดยคนละทีม วิธีที่ดีที่สุดมักเป็นให้ job แรกเขียน output ไปยังตำแหน่งที่ job อื่นทั้งหมดอ่านได้ แล้วตั้งเวลาให้ job ที่ใช้ข้อมูลนั้น (consumer job) รันทุกครั้งที่ข้อมูลถูกอัปเดต หรือตามตารางเวลาอื่น

• คุณอาจต้องส่งต่อข้อมูลจากเครื่องมือประมวลผลตัวหนึ่งไปยังอีกตัวหนึ่ง ตัวอย่างเช่น Spark job อาจเขียน output ไปที่ HDFS จากนั้น Python script ก็ trigger Trino SQL query (ดู "Cloud Data Warehouses" ) ที่ประมวลผลไฟล์ HDFS ต่อ แล้วเขียน output ไปที่ S3

• data pipeline บางตัวมีขั้นตอนการประมวลผลภายในหลายขั้น ตัวอย่างเช่น ถ้าขั้นตอนหนึ่งต้อง shard ข้อมูลด้วย key หนึ่ง และขั้นตอนถัดไปต้อง shard ด้วยอีก key หนึ่ง ขั้นตอนแรกก็สามารถ shard output ไว้ในแบบที่ขั้นตอนที่สองต้องการได้เลย

ในตัวอย่างเครื่องมือ Unix pipe ที่ต่อ output ของคำสั่งหนึ่งเข้ากับ input ของอีกคำสั่งหนึ่งใช้แค่ buffer ขนาดเล็กในหน่วยความจำ และไม่ได้เขียนข้อมูลลงไฟล์ ถ้า buffer เต็ม process ที่สร้าง output ต้องรอจนกว่า process ที่อ่านจะดึงข้อมูลบางส่วนออกจาก buffer ก่อน จึงจะส่ง output เพิ่มได้ ซึ่งเป็น backpressure รูปแบบหนึ่ง Spark, Flink และ batch execution engine อื่นๆ รองรับโมเดลที่คล้ายกัน คือ output ของ task หนึ่งถูกส่งตรงไปยังอีก task หนึ่ง (ผ่านเครือข่ายถ้า task รันอยู่บนเครื่องคนละเครื่อง)

อย่างไรก็ตาม โดยทั่วไป job หนึ่งใน workflow จะเขียน output ลง distributed filesystem หรือ object store แล้ว job ถัดไปจึงอ่านจากที่นั่น วิธีนี้แยก job ออกจากกัน ทำให้รันในเวลาต่างกันได้ ถ้า job มี input หลายตัว workflow scheduler มักจะรอให้ job ทั้งหมดที่สร้าง input เหล่านั้นเสร็จก่อน จึงรัน job ที่ใช้ input เหล่านั้น

Schedulers ที่พบใน orchestration frameworks เช่น ResourceManager ของ YARN หรือ scheduler ในตัวของ Spark ไม่ได้จัดการทั้ง workflow แต่ทำ scheduling ทีละ job ดังนั้นเพื่อจัดการ dependency ระหว่างการรัน job จึงมีการพัฒนา workflow scheduler ขึ้นมาหลายตัว เช่น Airflow, Dagster และ Prefect workflow scheduler มีฟีเจอร์ด้านการบริหารจัดการที่มีประโยชน์เมื่อต้องดูแล batch job จำนวนมาก workflow ที่ประกอบด้วย job 50 ถึง 100 ตัวเป็นเรื่องปกติใน data pipeline หลายแห่ง และในองค์กรขนาดใหญ่ หลายทีมอาจรัน job หรือ workflow ที่อ่าน output ของกันและกันข้ามหลายระบบ การมีเครื่องมือช่วยจึงสำคัญสำหรับการจัดการ dataflow ที่ซับซ้อนแบบนี้

Handling faults (การจัดการข้อผิดพลาด)

Batch job มักรันนาน และ job ที่ใช้เวลานานพร้อมมี task ขนานจำนวนมากมีแนวโน้มที่จะเจอ task ล้มเหลวอย่างน้อยหนึ่งตัวระหว่างทาง ดังที่กล่าวถึงใน "Hardware and Software Faults" และ "Unreliable Networks" ซึ่งเกิดได้จากหลายสาเหตุ เช่น hardware fault (โดยเฉพาะบน commodity hardware) หรือเครือข่ายสะดุด

อีก สาเหตุอีกอย่างที่ task อาจรันไม่จบคือ scheduler อาจจงใจ preempt (ฆ่า) มัน การ preempt มีประโยชน์มากถ้ามีหลายระดับ priority เช่น task priority ต่ำที่รันถูกกว่า และ task priority สูงที่แพงกว่า task priority ต่ำรันได้ทุกครั้งที่มีกำลังคำนวณเหลือ แต่เสี่ยงถูก preempt ได้ทุกเมื่อถ้ามี task priority สูงกว่าเข้ามา virtual machine priority ต่ำที่ราคาถูกแบบนี้เรียกว่า spot instances บน Amazon EC2, spot virtual machines บน Azure, และ preemptible instances บน Google Cloud [ 16 ]

เพราะ batch processing มักใช้กับ job ที่ไม่ไวต่อเวลา จึงเหมาะกับการใช้ task priority ต่ำและ spot instance เพื่อลดต้นทุนการรัน job โดยพื้นฐานแล้ว job เหล่านี้ใช้กำลังคำนวณส่วนที่เหลือซึ่งไม่อย่างนั้นก็ว่างเปล่า จึงช่วยเพิ่ม utilization ของ cluster อย่างไรก็ตาม นั่นก็หมายความว่า task เหล่านี้มีแนวโน้มถูก scheduler ฆ่าบ่อยขึ้น เพราะ preemption เกิดบ่อยกว่า hardware fault [ 17 ]

เพราะ batch job สร้าง output ใหม่ทุกครั้งที่รัน การจัดการ task ที่ล้มเหลวจึงง่ายกว่าใน online system ระบบลบ output บางส่วนจากการรันที่ล้มเหลวทิ้ง แล้ว schedule task ให้รันใหม่บนเครื่องอื่นได้ แต่ถ้าต้อง rerun ทั้ง job เพราะ task เดียวล้มเหลวก็สิ้นเปลือง MapReduce และระบบรุ่นหลังจึงให้ parallel task ทำงานอิสระจากกัน เพื่อ retry ได้ในระดับ task แต่ละตัว [ 3 ]

Fault tolerance ซับซ้อนขึ้นเมื่อ output ของ task หนึ่งกลายเป็น input ของอีก task ใน workflow เดียวกัน MapReduce แก้ปัญหานี้ด้วยการเขียน intermediate data แบบนี้กลับลง distributed filesystem เสมอ และรอให้ task ที่เขียนเสร็จสมบูรณ์ก่อน จึงให้ task อื่นอ่านข้อมูล วิธีนี้ใช้ได้แม้ในสภาพแวดล้อมที่ preemption เกิดขึ้นเป็นปกติ แต่ก็หมายถึงการเขียนลง DFS จำนวนมาก ซึ่งอาจไม่มีประสิทธิภาพ

Spark เก็บ intermediate data ไว้ในหน่วยความจำ ("spill" ลง local disk ถ้าใส่ไม่พอ) และเขียนเฉพาะผลลัพธ์สุดท้ายลง DFS นอกจากนี้ยังติดตามว่า intermediate data ถูกคำนวณมาอย่างไร ทำให้คำนวณใหม่ได้ถ้ามันสูญหาย [ 18 ] ส่วน Flink ใช้แนวทางต่างออกไป คือ checkpoint snapshot ของ task เป็นระยะ [ 19 ] เราจะกลับมาที่หัวข้อนี้ใน "Dataflow Engines" .

Batch Processing Models (โมเดลการประมวลผลแบบแบตช์)

เราได้เห็นแล้วว่า batch job ถูก schedule ในสภาพแวดล้อมแบบกระจายอย่างไร ต่อไปมาดูว่า batch processing framework ประมวลผลข้อมูลอย่างไร โมเดลที่พบบ่อยที่สุดมีสองแบบคือ MapReduce และ dataflow engine แม้ dataflow engine จะเข้ามาแทนที่ MapReduce ในทางปฏิบัติเป็นส่วนใหญ่แล้ว แต่การเข้าใจว่า MapReduce ทำงานอย่างไรก็ยังมีประโยชน์ เพราะมันมีอิทธิพลต่อ batch processing framework สมัยใหม่หลายตัว

MapReduce และ dataflow engine พัฒนาจนรองรับ programming model ได้หลายแบบ ทั้ง low-level programmatic API, relational query language และ DataFrame API ตัวเลือกที่หลากหลายนี้ช่วยให้ application engineer, analytics engineer, business analyst และแม้แต่พนักงานที่ไม่ใช่สายเทคนิค ประมวลผลข้อมูลของบริษัทสำหรับ use case ต่างๆ ได้ ซึ่งจะกล่าวถึงใน "Batch Use Cases" .

MapReduce

รูปแบบการประมวลผลข้อมูลใน MapReduce คล้ายกับตัวอย่างการวิเคราะห์ web server log ใน "Simple Log Analysis" มาก:

1. อ่านชุดของ input file แล้วแบ่งเป็น records ในตัวอย่าง web server log แต่ละ record คือหนึ่งบรรทัดใน log (นั่นคือ \n เป็นตัวคั่น record) ใน Hadoop's MapReduce input file จะถูกเก็บใน distributed filesystem เช่น HDFS หรือ object store เช่น S3 และใช้รูปแบบไฟล์ได้หลายแบบ เช่น Apache Parquet (รูปแบบ columnar ดู "Column-Oriented Storage" ) หรือ Apache Avro (รูปแบบ row-based ดู "Avro" )

2. เรียก mapper function เพื่อ extract key และ value จาก input record แต่ละตัว ในตัวอย่างเครื่องมือ Unix mapper function คือ awk '{print $7}' ซึ่ง extract URL ( $7 ) เป็น key และปล่อยให้ value ว่าง

3. sort key-value pair ทั้งหมดตาม key ในตัวอย่าง log ขั้นตอนนี้ทำโดยคำสั่ง sort แรก

4. เรียก reducer function เพื่อ iterate ผ่าน sorted key-value pair ถ้า key เดียวกันปรากฏหลายครั้ง การ sort จะทำให้มันอยู่ติดกันใน list จึงรวมค่าเหล่านั้นได้ง่ายโดยไม่ต้องเก็บ state จำนวนมากในหน่วยความจำ ในตัวอย่างเครื่องมือ Unix reducer ถูก implement ด้วยคำสั่ง uniq -c ซึ่งนับจำนวน record ที่อยู่ติดกันและมี key เดียวกัน

MapReduce job หนึ่ง job สามารถทำสี่ขั้นตอนนั้นได้ครบ ขั้นตอนที่ 2 (map) และ 4 (reduce) คือส่วนที่คุณเขียน custom data processing code เอง ขั้นตอนที่ 1 (การแบ่งไฟล์เป็น record) ถูกจัดการโดย input format parser ส่วนขั้นตอนที่ 3 คือขั้นตอน sort , ซึ่งเป็น implicit ใน MapReduce คุณไม่ต้องเขียนเอง เพราะ output จาก mapper จะถูก sort เสมอก่อนส่งไปยัง reducer การ sort นี้เป็น algorithm พื้นฐานของ batch processing ซึ่งจะกล่าวถึงอีกครั้งใน "Shuffling Data" .

การสร้าง MapReduce job ต้อง implement callback function สองตัวคือ mapper และ reducer ซึ่งทำงานดังนี้:

Mapper

Mapper ถูกเรียกหนึ่งครั้งต่อ input record แต่ละตัว และมีหน้าที่ extract key และ value จาก record นั้น สำหรับแต่ละ input มันจะสร้าง key-value pair กี่ตัวก็ได้ (รวมถึงไม่สร้างเลย) และไม่เก็บ state ใดๆ จาก input record หนึ่งไปยังอีก record ดังนั้นแต่ละ record จึงถูกจัดการอย่างอิสระ จึงมี mapper หลายตัวรันขนานกันบนส่วนต่างๆ ของ input ได้

Reducer

MapReduce framework จะนำ key-value pair ที่ mapper สร้างขึ้น รวบรวมค่าทั้งหมดที่มี key เดียวกัน แล้วเรียก reducer พร้อม iterator ที่วนผ่านค่าเหล่านั้น reducer สร้าง output record ได้ (เช่น จำนวนครั้งที่ URL เดียวกันปรากฏ) reducer ของ key ที่ต่างกันก็รันขนานกันได้เช่นกัน

ในตัวอย่าง web server log เรามีคำสั่ง sort ที่สองใน step 5 ซึ่งจัดอันดับ URL ตามจำนวน request ใน MapReduce ถ้าต้องการการ sort รอบที่สอง ก็ทำได้โดยเขียน MapReduce job ที่สอง แล้วใช้ output ของ job แรกเป็น input ของ job ที่สอง เมื่อมองแบบนี้ mapper มีหน้าที่เตรียมข้อมูลให้อยู่ในรูปแบบที่เหมาะกับการ sort ส่วน reducer มีหน้าที่ประมวลผลข้อมูลที่ sort แล้ว

MapReduce and Functional Programming (MapReduce และการเขียนโปรแกรมเชิงฟังก์ชัน)

แม้ MapReduce จะใช้สำหรับ batch processing แต่ programming model ของมันมาจาก functional programming Lisp มี map และ reduce (หรือ fold ) เป็น higher-order function บน list และต่อมาก็แพร่เข้าสู่ภาษาหลักอย่าง Python, Rust และ Java

การประมวลผลข้อมูลทั่วไปหลายอย่าง รวมถึงที่ SQL ให้มา ก็ implement บน MapReduce ได้ หลักการของ functional programming ที่หลีกเลี่ยง mutable state ทำให้รันแบบขนานได้ เพราะการเรียก mapper และ reducer ทุกครั้งขึ้นอยู่กับข้อมูลที่ MapReduce framework ส่งให้อย่างชัดเจนเท่านั้น framework จึงรันการเรียกที่เป็นอิสระต่อกันแบบขนานบนหลาย node ได้ และถ้า task ล้มเหลว framework ก็เรียก mapper และ reducer ซ้ำด้วย input เดิมบน node อื่นได้

การ implement processing job ที่ซับซ้อนด้วย raw MapReduce API ค่อนข้างลำบาก เช่น join algorithm ที่ job ใช้ต้อง implement เองตั้งแต่ต้น [ 20 ] MapReduce ยังค่อนข้างช้าเมื่อเทียบกับ batch processor รุ่นใหม่ เหตุผลหนึ่งคือ file-based I/O ของมันทำให้ทำ job pipelining ไม่ได้ (คือการประมวลผล output data ใน downstream job ก่อนที่ upstream job จะเสร็จสมบูรณ์)

Dataflow Engines (เอนจิน Dataflow)

เพื่อแก้ปัญหาบางอย่างของ MapReduce จึงมีการพัฒนา execution engine ใหม่หลายตัวสำหรับ distributed batch computation ซึ่งรู้จักกันดีที่สุดคือ Spark [ 18 , 21 ] และ Flink [ 19 ] ทั้งสองถูกออกแบบต่างกัน แต่มีจุดร่วมอย่างหนึ่ง คือจัดการ workflow ทั้งหมดเป็น job เดียว แทนที่จะแบ่งเป็น subjob อิสระ

เพราะระบบเหล่านี้โมเดลการไหลของข้อมูลผ่านขั้นตอนการประมวลผลหลายขั้นไว้อย่างชัดเจน จึงเรียกว่า dataflow engines เช่นเดียวกับ MapReduce ระบบเหล่านี้รองรับ low-level API ที่เรียก user-defined function ซ้ำๆ เพื่อประมวลผลทีละ record แต่ยังมี higher-level operator เช่น join และ group by ระบบเหล่านี้ทำงานขนานกันโดย shard input และคัดลอก output ของ task หนึ่งผ่านเครือข่ายไปเป็น input ของอีก task หนึ่ง ต่างจาก MapReduce ตรงที่ operator ไม่ต้องสลับบทบาท map กับ reduce อย่างเคร่งครัด แต่ประกอบกันได้ยืดหยุ่นกว่า

Dataflow API เหล่านี้โดยทั่วไปใช้ building block แบบ relational เพื่อแสดงการคำนวณ เช่น การ join dataset ด้วยค่าของ field, การ group tuple ด้วย key, การ filter ตามเงื่อนไข และการ aggregate tuple ด้วยการนับ การรวม หรือฟังก์ชันอื่นๆ ภายใน การทำงานเหล่านี้ถูก implement ด้วย shuffle algorithm ที่จะกล่าวถึงในส่วนถัดไป

engine แบบนี้พัฒนามาจากระบบวิจัยอย่าง Dryad [ 22 ] และ Nephele [ 23 ] และมีข้อดีหลายประการเมื่อเทียบกับโมเดล MapReduce:

• งานที่แพงอย่าง sorting ทำเฉพาะตรงที่จำเป็นเท่านั้น แทนที่จะเกิดขึ้นเป็นค่าเริ่มต้นระหว่างทุก map และ reduce stage

• เมื่อมี operator หลายตัวต่อเนื่องกันที่ไม่เปลี่ยนการ shard ของ dataset (เช่น map หรือ filter) ก็รวมเป็น task เดียวได้ ช่วยลด overhead ของการคัดลอกข้อมูล

• เพราะ join และ data dependency ทั้งหมดใน workflow ถูกประกาศไว้อย่างชัดเจน scheduler จึงเห็นภาพรวมว่าต้องใช้ข้อมูลอะไรที่ไหน และทำ locality optimization ได้ ตัวอย่างเช่น มันอาจพยายามวาง task ที่ใช้ข้อมูลชุดหนึ่งไว้บนเครื่องเดียวกับ task ที่สร้างข้อมูลนั้น เพื่อให้แลกเปลี่ยนข้อมูลผ่าน shared memory buffer ได้ แทนที่จะต้องคัดลอกผ่านเครือข่าย

• โดยปกติ การเก็บ intermediate state ระหว่าง operator ไว้ในหน่วยความจำหรือเขียนลง local disk ก็เพียงพอ ซึ่งใช้ I/O น้อยกว่าการเขียนลง distributed filesystem หรือ object store (ที่ต้อง replicate ไปยังหลายเครื่องและเขียนลง disk บนแต่ละ replica) MapReduce ใช้ optimization นี้กับ mapper output อยู่แล้ว แต่ dataflow engine ขยายแนวคิดนี้ไปใช้กับ intermediate state ทั้งหมด

• Operator เริ่มรันได้ทันทีที่ input พร้อม ไม่ต้องรอให้ stage ก่อนหน้าทั้งหมดเสร็จก่อนจึงจะเริ่ม stage ถัดไปได้

• process ที่มีอยู่แล้วนำกลับมาใช้รัน operator ใหม่ได้ ช่วยลด overhead ในการเริ่มต้นเมื่อเทียบกับ MapReduce (ที่เริ่ม JVM ใหม่ให้ทุก task)

คุณใช้ dataflow engine implement การคำนวณแบบเดียวกับ MapReduce workflow ได้ และมักรันได้เร็วกว่ามากเพราะ optimization ที่อธิบายไว้ข้างต้น

Shuffling Data (การสับเปลี่ยนข้อมูล)

อย่างที่เห็นแล้ว ทั้งตัวอย่างเครื่องมือ Unix ตอนต้นบทและ MapReduce ต่างก็อาศัยการ sort Batch processor ต้อง sort dataset ที่มีขนาดหลาย petabyte ซึ่งใหญ่เกินกว่าจะใส่บนเครื่องเดียวได้ จึงต้องมี distributed sorting algorithm ที่ทั้ง input และ output ถูก shard algorithm แบบนี้เรียกว่า shuffle .

Shuffle is not random (Shuffle ไม่ใช่การสุ่ม)

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

Shuffling เป็น algorithm พื้นฐานของ batch processor ซึ่งใช้ทำ join และ aggregation MapReduce, Spark, Flink, Daft, Dataflow และ BigQuery [ 24 ] ล้วน implement shuffle algorithm ที่ scalable และมีประสิทธิภาพเพื่อจัดการ dataset ขนาดใหญ่ เราจะใช้ shuffle ใน Hadoop MapReduce [ 25 ] เป็นตัวอย่างประกอบ แต่แนวคิดในส่วนนี้ใช้กับระบบอื่นๆ ได้เช่นกัน

Figure 11-1 แสดง dataflow ใน MapReduce job สมมติว่า input ของ job ถูก shard และเรียก shard เหล่านั้นว่า m 1 , m 2 , และ m 3 ตัวอย่างเช่น แต่ละ shard อาจเป็นไฟล์แยกกันบน HDFS หรือเป็น object แยกกันใน object store และ shard ทั้งหมดของ dataset เดียวกันจะอยู่ใน HDFS directory เดียวกัน หรือมี key prefix เดียวกันใน object store bucket

Figure 11-1. MapReduce job ที่มี mapper สามตัวและ reducer สามตัว

Framework เริ่ม map task แยกกันให้แต่ละ input shard โดย task อ่านไฟล์ที่ได้รับมอบหมาย แล้วส่งทีละ record ไปยัง mapper callback ฝั่ง reduce ของการคำนวณก็ถูก shard เช่นกัน จำนวน map task ถูกกำหนดโดยจำนวน input shard ส่วนจำนวน reduce task ผู้เขียน job เป็นคนกำหนด (ซึ่งอาจต่างจากจำนวน map task ก็ได้)

Output ของ mapper คือ key-value pair และ framework ต้องแน่ใจว่าถ้า mapper สองตัวส่ง key เดียวกัน key-value pair เหล่านั้นจะถูกประมวลผลโดย reducer task ตัวเดียวกัน เพื่อให้ได้อย่างนั้น แต่ละ mapper จะสร้าง output file แยกบน local disk ของตัวเองให้ reducer แต่ละตัว (ตัวอย่างเช่น ไฟล์ m 1, r 2 ใน Figure 11-1 คือไฟล์ที่ mapper 1 สร้างขึ้นโดยมีข้อมูลสำหรับ reducer 2) เมื่อ mapper ส่ง key-value pair ออกมา hash ของ key มักเป็นตัวกำหนดว่าจะเขียนลงไฟล์ของ reducer ตัวไหน (คล้ายกับกระบวนการที่อธิบายใน "Sharding by Hash of Key" )

ระหว่างที่ mapper เขียนไฟล์เหล่านี้ มันจะ sort key-value pair ภายในแต่ละไฟล์ไปด้วย ซึ่งทำได้ด้วยเทคนิคที่เห็นใน "Log-Structured Storage" : key-value pair ถูกเก็บเป็น batch ใน sorted data structure ในหน่วยความจำก่อน จากนั้นเขียนออกเป็น sorted segment file และค่อยๆ merge segment file ที่เล็กกว่าเป็นไฟล์ที่ใหญ่ขึ้น

หลังจาก mapper แต่ละตัวทำงานเสร็จ reducer จะเชื่อมต่อกับ mapper นั้นและคัดลอกไฟล์ sorted key-value pair ที่เกี่ยวข้องมาไว้ใน local disk ของตัวเอง เมื่อ reduce task ได้ส่วนของ output จาก mapper ทั้งหมดครบแล้ว ก็จะ merge ไฟล์เหล่านี้เข้าด้วยกันแบบ mergesort โดยรักษาลำดับที่ sort ไว้ ทำให้ key-value pair ที่มี key เดียวกันอยู่ติดกัน แม้จะมาจากคนละ mapper ก็ตาม reducer function จะถูกเรียกหนึ่งครั้งต่อ key พร้อม iterator ที่คืนค่าทั้งหมดของ key นั้น

record ใดๆ ที่ reducer function ส่งออกมาจะถูกเขียนต่อกันลงไฟล์ โดยมีหนึ่งไฟล์ต่อ reduce task หนึ่งตัว ไฟล์เหล่านี้ ( r 1 , r 2 , และ r 3 ใน Figure 11-1 ) จะกลายเป็น shard ของ output dataset ของ job และถูกเขียนกลับลง distributed filesystem หรือ object store

แม้ MapReduce จะทำ shuffle ระหว่างขั้นตอน map กับ reduce แต่ dataflow engine และ cloud data warehouse สมัยใหม่ซับซ้อนกว่านั้น ระบบอย่าง BigQuery optimize shuffle algorithm ของตัวเองให้เก็บข้อมูลในหน่วยความจำ และเขียนข้อมูลไปยัง external sorting service [ 24 ] service เหล่านี้ช่วยให้ shuffling เร็วขึ้น และ replicate shuffled data เพื่อให้ทนต่อความล้มเหลวได้

Joins and Grouping (การ Join และการจัดกลุ่ม)

มาดูว่าข้อมูลที่ sort แล้วทำให้ distributed join และ aggregation ง่ายขึ้นได้อย่างไร เราจะใช้ MapReduce เป็นตัวอย่างต่อไป แม้แนวคิดเหล่านี้จะใช้ได้กับระบบ batch processing ส่วนใหญ่

ตัวอย่างทั่วไปของ join ใน batch job แสดงใน Figure 11-2 ด้านซ้ายคือ log ของ event ที่บันทึกสิ่งที่ผู้ใช้ซึ่ง login แล้วทำบนเว็บไซต์ (เรียกว่า activity events หรือ clickstream data ) และด้านขวาคือ database ของ user คุณอาจมองตัวอย่างนี้เป็นส่วนหนึ่งของ star schema (ดู "Stars and Snowflakes: Schemas for Analytics" ); log ของ event คือ fact table และ user database คือหนึ่งใน dimension

Figure 11-2. การ join ระหว่าง log ของ user activity events และ database ของ user profiles

ถ้าคุณต้องการวิเคราะห์ activity event โดยใช้ข้อมูลจาก user database (เช่น หาว่าหน้าไหนได้รับความนิยมมากกว่าในกลุ่มผู้ใช้อายุน้อยหรืออายุมาก โดยใช้ field วันเกิดใน user profile) คุณต้อง join สองตารางนี้เข้าด้วยกัน แล้วจะ join อย่างไร ถ้าทั้งสองตารางใหญ่จนต้อง shard?

คุณใช้ประโยชน์จากข้อเท็จจริงที่ว่าใน MapReduce shuffle จะนำ key-value pair ทั้งหมดที่มี key เดียวกันมารวมกันที่ reducer เดียวกัน ไม่ว่าเดิมจะอยู่ใน shard ไหนก็ตาม ในที่นี้ user ID ทำหน้าที่เป็น key ได้ คุณจึงเขียน mapper ที่อ่าน user activity event แล้วส่ง page view URL ออกมาโดยใช้ user ID เป็น key ดังที่แสดงใน Figure 11-3 ส่วน mapper อีกตัวอ่าน user database ทีละแถว โดย extract user ID เป็น key และวันเกิดของผู้ใช้เป็น value

จากนั้น shuffle จะทำให้แน่ใจว่า reducer function เข้าถึงวันเกิดของผู้ใช้คนหนึ่งและ event การดูหน้าเว็บทั้งหมดของผู้ใช้คนนั้นได้พร้อมกัน MapReduce job ยังจัดลำดับ record ให้ reducer เห็น record จาก user database ก่อนเสมอ ตามด้วย activity event ตามลำดับเวลาได้ด้วย เทคนิคนี้เรียกว่า secondary sort [ 25 ]

ทีนี้ reducer ก็ทำ join logic จริงได้ง่ายๆ ค่าแรกจะเป็นวันเกิด ซึ่ง reducer เก็บไว้ใน local variable จากนั้นจะ iterate ผ่าน activity event ที่มี user ID เดียวกัน แล้วส่งออกแต่ละ URL ที่ถูกดูพร้อมวันเกิดของผู้ดู เพราะ reducer ประมวลผล record ทั้งหมดของ user ID หนึ่งในครั้งเดียว มันจึงต้องเก็บ user record ไว้ในหน่วยความจำแค่รายการเดียวในแต่ละช่วงเวลา และไม่ต้องส่ง request ใดๆ ผ่านเครือข่าย algorithm นี้เรียกว่า sort-merge join เพราะ mapper output ถูก sort ด้วย key แล้ว reducer จึง merge sorted list ของ record จากทั้งสองฝั่งของ join เข้าด้วยกัน

MapReduce job ถัดไปใน workflow สามารถคำนวณการกระจายอายุของผู้ดูสำหรับแต่ละ URL ได้ โดย job จะ shuffle ข้อมูลโดยใช้ URL เป็น key ก่อน เมื่อ sort แล้ว reducer จะ iterate ผ่าน page view ทั้งหมด (ที่มีวันเกิดของผู้ดู) ของ URL เดียว เก็บ counter จำนวนการดูของแต่ละกลุ่มอายุ และเพิ่ม counter ที่เหมาะสมให้แต่ละ page view วิธีนี้ทำให้ implement group by และ aggregation ได้

Figure 11-3. Sort-merge join บน user ID ถ้า input dataset ถูก shard เป็นหลายไฟล์ ก็ประมวลผลแต่ละไฟล์ด้วย mapper หลายตัวแบบขนานได้

Query Languages (ภาษา Query)

ตลอดหลายปีที่ผ่านมา execution engine สำหรับ distributed batch processing เติบโตเต็มที่แล้ว โครงสร้างพื้นฐานตอนนี้แข็งแกร่งพอจะจัดเก็บและประมวลผลข้อมูลหลาย petabyte บน cluster ที่มีมากกว่า 10,000 เครื่อง เมื่อปัญหาในการ operate batch process ในระดับนั้นถือว่าถูกแก้ไปไม่มากก็น้อย ความสนใจจึงหันไปที่การปรับปรุง programming model

MapReduce, dataflow engine และ cloud data warehouse ล้วนหันมาใช้ SQL เป็นภาษากลางสำหรับ batch processing ซึ่งเป็นเรื่องธรรมชาติ เพราะ data warehouse แบบดั้งเดิมใช้ SQL เครื่องมือ data analytics และ ETL ก็รองรับอยู่แล้ว และนักพัฒนาทุกคนรวมถึงนักวิเคราะห์ก็รู้จัก SQL

นอกจากใช้โค้ดน้อยกว่า MapReduce job ที่เขียนด้วยมือแล้ว query language เหล่านี้ยังรองรับการใช้งานแบบ interactive คือเขียน analytical query แล้วรันจาก terminal หรือ GUI ได้เลย การ query แบบ interactive นี้เป็นวิธีที่มีประสิทธิภาพและเป็นธรรมชาติสำหรับ business analyst, product manager, ทีมขายและการเงิน และอื่นๆ ที่จะสำรวจข้อมูลในสภาพแวดล้อม batch processing การรองรับ SQL ยังทำให้ distributed batch processing system เหมาะกับ exploratory query ด้วย

Query language ระดับสูงไม่ได้แค่ช่วยให้คนที่ใช้ระบบทำงานได้มีประสิทธิภาพขึ้นเท่านั้น แต่ยังช่วยให้ job รันได้มีประสิทธิภาพขึ้นในระดับเครื่องด้วย ดังที่เห็นใน "Cloud Data Warehouses" query engine ทำหน้าที่แปลง SQL query เป็น batch job ที่จะรันบน cluster การแปลจาก query เป็น syntax tree แล้วเป็น physical operator นี้ทำให้ engine optimize query ได้ query engine อย่าง Hive, Trino, Spark และ Flink มี cost-based query optimizer ที่วิเคราะห์คุณสมบัติของ join input แล้วเลือกให้อัตโนมัติว่า algorithm ไหนเหมาะกับงานที่สุด optimizer อาจถึงขั้นสลับลำดับของ join เพื่อให้ intermediate state มีปริมาณน้อยที่สุดด้วย [ 19 , 26 , 27 , 28 ]

SQL เป็น query language สำหรับ batch processing ทั่วไปที่ได้รับความนิยมมากที่สุด แต่ภาษาอื่นๆ ก็ยังถูกใช้กับความต้องการเฉพาะทาง ตัวอย่างเช่น Apache Pig เป็นภาษาที่อิงกับ relational operator ซึ่งให้ระบุ data pipeline ทีละขั้นตอน แทนที่จะเขียนเป็น SQL query ก้อนใหญ่ก้อนเดียว DataFrame (ที่จะกล่าวถึงในส่วนถัดไป) มีลักษณะคล้ายกัน และ Morel เป็นภาษาที่ใหม่กว่าซึ่งได้รับอิทธิพลจาก Pig ส่วนผู้ใช้กลุ่มอื่นก็หันไปใช้ JSON query language เช่น jq, JMESPath หรือ JSONPath

ใน "Graph-Like Data Models" เราได้พูดถึงการใช้ graph สร้างโมเดลข้อมูล และการใช้ graph query language เพื่อ traverse edge และ vertex ใน graph graph processing framework หลายตัวยังรองรับ batch computation ผ่าน query language เช่น Gremlin ของ Apache TinkerPop เราจะดูสถานการณ์การใช้ graph processing อย่างละเอียดขึ้นใน "Batch Use Cases" .

Batch Processing and Cloud Data Warehouses Converge (การบรรจบกันของการประมวลผลแบบแบตช์และคลาวด์ Data Warehouse)

ในอดีต data warehouse รันบน specialized hardware appliance และรองรับ analytical query แบบ SQL บน relational data ส่วน batch processing framework อย่าง MapReduce ตั้งใจให้ scale ได้มากขึ้นและยืดหยุ่นขึ้น โดยรองรับ processing logic ที่เขียนด้วยภาษา programming ทั่วไป และอ่านเขียนข้อมูลได้ทุกรูปแบบ

เมื่อเวลาผ่านไป ทั้งสองแบบก็คล้ายกันมากขึ้น batch processing framework สมัยใหม่รองรับ SQL เป็นภาษาสำหรับเขียน batch job และได้ประสิทธิภาพดีกับ relational query ด้วยการใช้ columnar storage format อย่าง Parquet และ query execution engine ที่ optimize แล้ว (ดู "Query Execution: Compilation and Vectorization" ) ขณะเดียวกัน data warehouse ก็ scale ได้มากขึ้นด้วยการย้ายไปอยู่บน cloud (ดู "Cloud Data Warehouses" ) และ implement เทคนิค scheduling, fault tolerance และ shuffling แบบเดียวกับที่ distributed batch framework ใช้ หลายตัวยังใช้ distributed filesystem ด้วย

เช่นเดียวกับที่ batch processing system นำ SQL มาเป็น processing model cloud warehouse ก็นำ processing model แบบอื่นมาใช้เช่นกัน ตัวอย่างเช่น BigQuery มี DataFrames library ส่วน Snowpark library ของ Snowflake เชื่อมต่อกับ Pandas ได้ และ workflow orchestrator สำหรับ batch processing อย่าง Airflow, Prefect และ Dagster ก็ทำงานร่วมกับ cloud warehouse ได้

อย่างไรก็ตาม batch job ไม่ได้ทุกตัวที่เขียนเป็น SQL ได้ง่าย เช่น iterative graph algorithm อย่าง PageRank, งาน ML ที่ซับซ้อน และ workflow อื่นๆ อีกมากมาย การประมวลผลข้อมูลสำหรับ AI ซึ่งรวมถึงข้อมูล nonrelational และ multimodal เช่น image, video และ audio ก็อาจเขียนเป็น SQL ได้ยากเช่นกัน

cloud data warehouse เองก็ทำ workload บางประเภทได้ไม่ดีเช่นกัน การคำนวณทีละ row มีประสิทธิภาพน้อยลงเมื่อใช้ column-oriented storage format ในกรณีเช่นนี้ควรใช้ warehouse API ทางเลือกหรือ batch processing system แทน นอกจากนี้ cloud data warehouse ยังมีแนวโน้มแพงกว่า batch processing system อื่นๆ การรัน job ขนาดใหญ่บน batch processing system อย่าง Spark หรือ Flink อาจคุ้มค่ากว่า

ท้ายที่สุด การเลือกระหว่างประมวลผลข้อมูลบน batch system หรือ data warehouse มักขึ้นอยู่กับปัจจัยอย่างต้นทุน ความสะดวก ความง่ายในการ implement และความพร้อมใช้งาน องค์กรขนาดใหญ่ส่วนใหญ่มีระบบประมวลผลข้อมูลหลายระบบ จึงมีความยืดหยุ่นในการตัดสินใจนี้ ส่วนบริษัทขนาดเล็กมักอยู่ได้ด้วยระบบเดียว

DataFrames

data scientist และนักสถิติโดยทั่วไปคุ้นเคยกับ data model แบบ DataFrame ที่พบใน R และ Pandas (ดู "DataFrames, Matrices, and Arrays" ) DataFrame คล้ายกับตารางใน relational database คือเป็น collection ของ row และค่าทั้งหมดใน column เดียวกันมี type เดียวกัน แทนที่จะเขียน SQL query ก้อนใหญ่ ผู้ใช้จะเรียกฟังก์ชันที่ตรงกับ relational operator เพื่อทำ filter, join, sorting, aggregation และ operation อื่นๆ

เดิมที การจัดการ DataFrame มักเกิดขึ้นในเครื่องและในหน่วยความจำ DataFrame จึงถูกจำกัดอยู่ที่ dataset ที่พอดีบนเครื่องเดียว data scientist ต้องการใช้ DataFrame API ที่คุ้นเคยกับ dataset ขนาดใหญ่ในสภาพแวดล้อม batch processing เพราะ SQL และ MapReduce ไม่ตอบโจทย์ distributed data processing framework อย่าง Spark, Flink และ Daft จึงนำ DataFrame API มาใช้เพื่อตอบสนองความต้องการนี้ อย่างไรก็ตาม implementation ทำงานต่างออกไปบ้าง DataFrame แบบ local มักมี index และเรียงลำดับ ส่วน distributed DataFrame โดยทั่วไปไม่เป็นเช่นนั้น [ 29 ] ซึ่งอาจทำให้เจอเรื่องไม่คาดคิดด้านประสิทธิภาพเมื่อย้ายไปใช้ batch framework

DataFrame API ดูคล้ายกับ dataflow API แต่ implementation ต่างกัน Pandas จะรัน operation ทันทีที่เรียก DataFrame method ส่วน Spark จะแปลงการเรียก DataFrame API ทั้งหมดเป็น query plan และรัน query optimization ก่อน แล้วจึงรัน workflow บน distributed dataflow engine ของตัวเอง

framework อย่าง Daft ถึงขั้นรองรับทั้ง client-side และ server-side computation operation ขนาดเล็กในหน่วยความจำจะรันบน client ส่วน dataset ที่ใหญ่กว่าจะประมวลผลบน server columnar storage format อย่าง Apache Arrow มี unified data model ที่ทั้ง client-side และ server-side execution engine ใช้ร่วมกันได้

Batch Use Cases (กรณีการใช้งาน Batch)

เมื่อเห็นแล้วว่า batch processing ทำงานอย่างไร มาดูว่ามันถูกนำไปใช้กับงานต่างๆ อย่างไร batch job เหมาะมากกับการประมวลผล dataset ขนาดใหญ่ทีเดียวเป็นก้อน แต่ไม่เหมาะกับ use case ที่ต้องการ latency ต่ำ ดังนั้นจึงมักพบ batch job ในที่ที่มีข้อมูลจำนวนมากและความสดใหม่ของข้อมูลไม่สำคัญ ฟังดูอาจจำกัด แต่ปรากฏว่างานประมวลผลข้อมูลจำนวนมากเข้ากับโมเดลนี้ ตัวอย่างเช่น:

• การทำบัญชีและการกระทบยอดสินค้าคงคลัง ซึ่งบริษัทตรวจสอบว่าธุรกรรมตรงกับบัญชีธนาคารและสินค้าคงคลัง มักดำเนินการเป็น batch jobs [ 30 ]

• ในภาคการผลิต การพยากรณ์ความต้องการมักรันเป็น periodic batch job [ 31 ]

• บริษัท Ecommerce, media, และ social media ฝึกโมเดลแนะนำของพวกเขาด้วย batch jobs [ 32 , 33 ]

• ระบบการเงินหลายระบบเป็น batch-based; ตัวอย่างเช่น เครือข่ายธนาคารของสหรัฐฯ รันบน batch jobs เกือบทั้งหมด [ 34 ]

ในส่วนต่อไปนี้ เราจะพูดถึงกรณีการใช้งาน batch processing ที่คุณจะพบในแทบทุกอุตสาหกรรม

Extract–Transform–Load (การสกัด-แปลง-โหลด)

"Data Warehousing" แนะนำ ETL และ ELT ซึ่ง data processing pipeline จะ extract ข้อมูลจาก production database, transform แล้วโหลดผลลัพธ์ไปยัง downstream system (ในส่วนนี้เราจะใช้คำว่า "ETL" แทนทั้ง ETL และ ELT workload) batch job มักถูกใช้กับ workload แบบนี้ โดยเฉพาะเมื่อ downstream system คือ data warehouse

batch job ทำงานแบบขนานได้ จึงเหมาะมากกับการแปลงข้อมูล ซึ่งส่วนใหญ่เป็น workload แบบ "embarrassingly parallel" การ filter ข้อมูล การ project field และการแปลงข้อมูลใน data warehouse ทั่วไปอื่นๆ ล้วนทำแบบขนานได้ทั้งหมด

สภาพแวดล้อม batch processing ยังมาพร้อม workflow scheduler ที่แข็งแกร่ง ซึ่งช่วยให้ schedule, orchestrate และ debug job ใน ETL data pipeline ได้ง่าย เมื่อเกิดความล้มเหลว scheduler มักจะ retry job เพื่อรับมือกับปัญหาชั่วคราวที่อาจเกิดขึ้น job ที่ล้มเหลวซ้ำๆ จะถูกทำเครื่องหมายว่าล้มเหลว ทำให้นักพัฒนาเห็นได้ง่ายว่า job ไหนใน data pipeline หยุดทำงาน scheduler อย่าง Airflow ยังมี source, sink และ query operator ในตัวสำหรับ MySQL, PostgreSQL, Snowflake, Spark, Flink และระบบยอดนิยมอื่นๆ อีกหลายสิบระบบ การที่ scheduler เชื่อมกับระบบประมวลผลข้อมูลได้แน่นแฟ้นช่วยให้ data integration ง่ายขึ้น

เรายังเห็นด้วยว่า batch job troubleshoot และแก้ไขได้ง่ายเมื่อมีอะไรผิดพลาด ซึ่งมีค่ามากเมื่อต้อง debug data pipeline เพราะตรวจไฟล์ที่ล้มเหลวเพื่อดูว่าเกิดอะไรขึ้นได้ง่าย และแก้ ETL batch job แล้ว rerun ได้ ตัวอย่างเช่น ถ้า input file ไม่มี field ที่ transformation batch job ตั้งใจจะใช้ data engineer ก็เห็นได้ง่ายว่า field นั้นหายไป แล้วอัปเดต transformation logic หรือ job ที่สร้าง input นั้น

ในอดีต data pipeline ถูกจัดการโดยทีม data engineering ทีมเดียว เพราะถือว่าไม่ยุติธรรมที่จะให้ทีมอื่นซึ่งทำ product feature อยู่มาเขียนและดูแล batch data pipeline ที่ซับซ้อน ล่าสุด การพัฒนา batch processing model และการจัดการ metadata ทำให้วิศวกรทั่วทั้งองค์กรมีส่วนร่วมและดูแล data pipeline ของตัวเองได้ง่ายขึ้นมาก Data mesh [ 35 , 36 ], data contract [ 37 ], และ data fabric [ 38 ] practice เหล่านี้ให้มาตรฐานและเครื่องมือที่ช่วยให้ทีมเผยแพร่ข้อมูลของตัวเองได้อย่างปลอดภัย เพื่อให้ใครก็ตามในองค์กรนำไปใช้ได้

data pipeline และ analytical query เริ่มใช้ร่วมกันไม่เพียงแต่ processing model แต่รวมถึง execution engine ด้วย ETL batch job จำนวนมากตอนนี้รันบนระบบเดียวกับ analytical query ที่อ่าน output ของมัน และไม่ใช่เรื่องแปลกที่จะเห็นทั้ง data pipeline transformation และ analytical query รันเป็น SparkSQL, Trino หรือ DuckDB query architecture แบบนี้ยิ่งทำให้เส้นแบ่งระหว่าง application engineering, data engineering, analytics engineering และ business analysis เลือนลางลงไปอีก

Analytics (การวิเคราะห์)

ใน "Operational Versus Analytical Systems" เราเห็นว่า analytical query (OLAP) มัก scan record จำนวนมาก แล้วทำ grouping และ aggregation เราสามารถรัน workload แบบนี้บน batch processing system ควบคู่ไปกับ batch processing workload อื่นๆ ได้ นักวิเคราะห์เขียน SQL query ที่รันบน query engine ซึ่งอ่านและเขียนข้อมูลบน distributed filesystem หรือ object store ส่วน table metadata เช่น การ map ระหว่าง table กับไฟล์, ชื่อ และ type ถูกจัดการด้วย table format เช่น Apache Iceberg และ catalogs เช่น Unity (ดู "Cloud Data Warehouses" ) สถาปัตยกรรมนี้เรียกว่า data lakehouse [ 39 ]

เช่นเดียวกับ ETL เมื่อ SQL query interface ดีขึ้น หลายองค์กรจึงใช้ batch framework อย่าง Spark ทำ analytics รูปแบบ query แบบนี้มีสองแบบ:

Pre-aggregation queries

ข้อมูลถูก roll up เป็น OLAP cube หรือ data mart เพื่อเร่ง query (ดู "Materialized Views and Data Cubes" ) ข้อมูลที่ pre-aggregate แล้วจะถูก query ใน warehouse หรือ push ไปยัง purpose-built real-time OLAP system เช่น Apache Druid หรือ Apache Pinot โดยปกติการ pre-aggregation จะเกิดขึ้นตามช่วงเวลาที่กำหนด และใช้ workflow scheduler ที่กล่าวถึงใน "Scheduling workflows" จัดการ workload เหล่านี้

Ad hoc queries

ผู้ใช้รัน query เหล่านี้เพื่อตอบคำถามทางธุรกิจเฉพาะเรื่อง ตรวจสอบพฤติกรรมของผู้ใช้ debug ปัญหาด้านการ operate และอื่นๆ อีกมากมาย response time สำคัญสำหรับ use case นี้ นักวิเคราะห์จะรัน query ซ้ำๆ ตามคำตอบที่ได้และสิ่งที่เรียนรู้เพิ่มเติมเกี่ยวกับข้อมูลที่กำลังตรวจสอบ batch processing framework ที่รัน query ได้เร็วจึงช่วยลดเวลารอของนักวิเคราะห์

การรองรับ SQL ทำให้ batch processing framework เชื่อมต่อกับ spreadsheet และเครื่องมือ data visualization เช่น Tableau, Power BI, Looker และ Apache Superset ได้ ตัวอย่างเช่น Tableau มี SparkSQL และ Presto connector ส่วน Apache Superset รองรับ Trino, Hive, Spark SQL, Presto และระบบอื่นๆ อีกมากมายที่สุดท้ายแล้วก็รัน batch job เพื่อ query ข้อมูล

Machine Learning (การเรียนรู้ของเครื่อง)

Machine learning (ML) ใช้ batch processing บ่อย data scientist, ML engineer และ AI engineer ใช้ batch processing framework เพื่อสำรวจรูปแบบในข้อมูล แปลงข้อมูล และ train ML model การใช้งานทั่วไปได้แก่:

Feature engineering

ข้อมูลดิบถูก filter และแปลงเป็นข้อมูลที่ใช้ train model ได้ predictive model มักต้องใช้ข้อมูลเชิงตัวเลข วิศวกรจึงต้องแปลงข้อมูลรูปแบบอื่น (เช่น text หรือ discrete value) ให้เป็นรูปแบบที่ต้องการ

Model training

ข้อมูล training คือ input ของ batch process และ weights ของโมเดลที่ train แล้วคือ output

Batch inference

ถ้า dataset มีขนาดใหญ่และไม่ต้องการผลลัพธ์แบบ real-time ก็ใช้โมเดลที่ train แล้วทำนายเป็นกลุ่มได้ ซึ่งรวมถึงการประเมินผลทำนายของโมเดลบน test dataset ด้วย

batch processing framework มีเครื่องมือสำหรับ use case เหล่านี้โดยเฉพาะ ตัวอย่างเช่น MLlib ของ Apache Spark และ FlinkML ของ Apache Flink มาพร้อมเครื่องมือ feature engineering, ฟังก์ชันทางสถิติ และ classifier ที่หลากหลาย

แอปพลิเคชัน ML อย่าง recommendation engine และ ranking system ก็ใช้ graph processing อย่างหนักเช่นกัน (ดู "Graph-Like Data Models" ) graph algorithm หลายตัวเขียนโดย traverse ทีละ edge, join vertex หนึ่งกับ vertex ที่อยู่ติดกันเพื่อ propagate ข้อมูลบางอย่าง และทำซ้ำจนกว่าจะเข้าเงื่อนไขที่กำหนด เช่น จนกว่าจะไม่มี edge ให้ตามต่อ หรือจนกว่า metric จะ converge

โมเดลการคำนวณ bulk synchronous parallel (BSP) [ 40 ] ได้รับความนิยมสำหรับการประมวลผล graph แบบ batch และถูก implement ไว้ใน Apache Giraph [ 20 ], GraphX API ของ Spark, และ Gelly API ของ Flink [ 41 ] เป็นต้น model นี้ยังรู้จัก ในชื่อ Pregel model เพราะ paper Pregel ของ Google ทำให้แนวทางนี้เป็นที่นิยมสำหรับการประมวลผล graph [ 42 ]

Batch processing ยังเป็นส่วนสำคัญของการเตรียมข้อมูลและการ train large language model (LLM) ข้อมูล input ที่เป็นข้อความดิบ เช่น เนื้อหาของเว็บไซต์ มักอยู่ใน DFS หรือ object store ข้อมูลนี้ต้องผ่านการ preprocess ให้เหมาะกับการ train ขั้นตอน preprocessing ที่เหมาะกับ batch processing framework ได้แก่:

• การ extract plain text จาก HTML และแก้ข้อความที่ผิดรูป

• การตรวจจับและลบเอกสารที่คุณภาพต่ำ ไม่เกี่ยวข้อง และซ้ำซ้อน

• การ tokenize ข้อความ (แบ่งเป็นคำ) และแปลงเป็น embedding หรือค่าตัวเลขที่แทนแต่ละคำ

batch processing framework อย่าง Kubeflow, Flyte และ Ray ถูกสร้างมาเพื่อ workload แบบนี้โดยเฉพาะ ตัวอย่างเช่น OpenAI ใช้ Ray เป็นส่วนหนึ่งของกระบวนการ train ChatGPT [ 43 ] framework เหล่านี้มี integration ในตัวกับ LLM และ AI library เช่น PyTorch, TensorFlow, XGBoost และอื่นๆ อีกมากมาย และรองรับ feature engineering, model training, batch inference และ fine-tuning (การปรับ foundation model ให้เหมาะกับ use case เฉพาะ) ในตัวด้วย

สุดท้าย data scientist มักทดลองกับข้อมูลใน interactive notebook เช่น Jupyter หรือ Hex โดย notebook ประกอบด้วย cells ซึ่งเป็นชิ้นส่วนเล็กๆ ของ Markdown, Python หรือ SQL cell จะถูก execute ตามลำดับเพื่อสร้าง spreadsheet, graph หรือข้อมูล notebook จำนวนมากใช้ batch processing ผ่าน DataFrame API หรือ query ระบบเหล่านั้นด้วย SQL

Serving Derived Data (การให้บริการข้อมูลที่ได้มาจากการประมวลผล)

batch job มักถูกใช้สร้าง dataset ที่ precompute ไว้หรือ derived เช่น คำแนะนำสินค้า รายงานที่ผู้ใช้เห็น และ feature สำหรับ ML model dataset เหล่านี้มักถูก serve จาก production database, key-value store หรือ search engine ไม่ว่าจะใช้ระบบใด ข้อมูลที่ precompute ไว้ต้องถูกส่งจาก distributed filesystem หรือ object store ของ batch processor กลับไปยัง database ที่รับ traffic จริงอยู่

คุณอาจอยากใช้ client library ของ database ที่ชอบตรงๆ ภายใน batch job แล้วเขียนไปยัง database server ทีละ record วิธีนี้ใช้ได้ (สมมติว่ากฎ firewall อนุญาตให้สภาพแวดล้อม batch processing เข้าถึง production database ได้โดยตรง) แต่ไม่ใช่ความคิดที่ดี ด้วยเหตุผลหลายประการ:

• การส่ง network request ต่อทุกๆ record ช้ากว่า throughput ปกติของ batch task หลายเท่า แม้ client library จะรองรับ batching ก็มีแนวโน้มว่าประสิทธิภาพจะยังแย่

• batch processing framework มักรันหลาย task พร้อมกัน ถ้า task ทั้งหมดเขียนไปยัง output database เดียวกันพร้อมกันในอัตราที่ batch process ทำได้ database นั้นอาจถูกถล่มจนล้นได้ง่าย และประสิทธิภาพในการตอบ query มีแนวโน้มลดลง ซึ่งอาจก่อให้เกิดปัญหาด้านการ operate ในส่วนอื่นๆ ของระบบ [ 44 ]

• โดยปกติ batch job ให้การรับประกันแบบ all-or-nothing ที่ชัดเจนสำหรับ output ของ job ถ้า job สำเร็จ ผลลัพธ์คือ output ที่ได้จากการรันทุก task หนึ่งครั้งพอดี แม้บาง task จะล้มเหลวและต้อง retry ระหว่างทาง ส่วนถ้า job ล้มเหลวทั้งหมด ก็จะไม่มี output เกิดขึ้น อย่างไรก็ตาม การเขียนไปยังระบบภายนอกจากภายใน job ทำให้เกิด side effect ที่มองเห็นได้จากภายนอก ซึ่งซ่อนแบบนี้ไม่ได้ ดังนั้นคุณต้องกังวลว่าระบบอื่นอาจเห็นผลลัพธ์จาก job ที่ทำเสร็จเพียงบางส่วน และถ้า task ล้มเหลวแล้วถูก restart ก็อาจเขียน output ซ้ำกับที่เกิดจากการรันที่ล้มเหลว

ทางออกที่ดีกว่าคือให้ batch job push dataset ที่ precompute ไว้ไปยัง stream เช่น Kafka topic ซึ่งจะกล่าวถึงเพิ่มเติมใน Chapter 12 search engine อย่าง Elasticsearch, real-time OLAP system อย่าง Apache Pinot และ Apache Druid, derived datastore อย่าง Venice [ 45 ] และ cloud data warehouse อย่าง ClickHouse ล้วนมีความสามารถในตัวในการ ingest ข้อมูลจาก Kafka เข้าสู่ระบบ การ push ข้อมูลผ่าน streaming system แก้ปัญหาที่กล่าวข้างต้นได้บางส่วน:

• streaming system ถูก optimize สำหรับ sequential write จึงเหมาะกับ workload การเขียนข้อมูลเป็นก้อนใหญ่ของ batch job มากกว่า

• streaming system ทำหน้าที่เป็น buffer ระหว่าง batch job กับ production database ได้ ระบบ downstream สามารถ throttle read rate ของตัวเองเพื่อให้ยังรับ production traffic ได้สบายๆ ต่อไป

• output ของ batch job เดียวสามารถถูกใช้โดย downstream system หลายระบบ

• Streaming system ยังทำหน้าที่เป็นขอบเขตด้านความปลอดภัยระหว่างสภาพแวดล้อม batch processing กับ production network ได้ โดย deploy ไว้ในเครือข่ายที่เรียกว่า demilitarized zone (DMZ) ซึ่งอยู่ระหว่าง batch processing network กับ production network

ปัญหาหนึ่งที่ streaming ไม่ได้แก้ให้โดยธรรมชาติคือการรับประกันแบบ all-or-nothing เพื่อให้ได้อย่างนั้น เมื่อ batch job เสร็จสมบูรณ์ต้องส่ง notification ไปแจ้ง downstream system ว่า job เสร็จแล้วและข้อมูลพร้อมให้บริการ ฝั่ง consumer ของ stream ต้องเก็บข้อมูลที่ได้รับไว้โดยไม่ให้ query มองเห็น เหมือน transaction ที่ยังไม่ commit ภายใต้ read-committed isolation (ดู "Read Committed" ) จนกว่าจะได้รับแจ้งว่า job เสร็จสมบูรณ์

อีก pattern หนึ่งที่พบบ่อยกว่าเมื่อ bootstrap database คือการสร้าง database ใหม่เอี่ยม ภายใน batch job แล้ว bulk-load ไฟล์เหล่านั้นเข้า database โดยตรงจาก distributed filesystem, object store หรือ local filesystem ระบบข้อมูลหลายระบบมีเครื่องมือ bulk import เช่น Lightning ของ TiDB และ Hadoop import job ของ Apache Pinot ส่วน RocksDB ก็มี API สำหรับ bulk-import ไฟล์ Sorted String Table (SST) จาก batch job

การสร้าง database แบบ batch แล้ว bulk-import ข้อมูลเข้าไปนั้นเร็วมาก และทำให้ระบบสลับระหว่าง dataset คนละเวอร์ชันแบบ atomic ได้ง่ายขึ้น ในทางกลับกัน การอัปเดต dataset แบบ incremental อาจทำได้ยากถ้าใช้ batch job ที่สร้าง database ใหม่ทั้งหมด จึงมักใช้แนวทางผสมเมื่อต้องการทั้ง bootstrap และ incremental load ตัวอย่างเช่น Venice รองรับ hybrid store ที่ให้ทั้ง batch row-based update และการสลับ dataset ทั้งชุด

Summary (สรุป)

ในบทนี้เราได้สำรวจการออกแบบและ implementation ของ batch processing system เราเริ่มด้วยชุดเครื่องมือ Unix คลาสสิก ( awk , sort , uniq , ฯลฯ) เพื่อแสดง primitive พื้นฐานของ batch processing เช่น sorting และ counting

จากนั้นเราขยายไปยัง distributed batch processing system โดย batch framework ประมวลผล input dataset ที่ immutable และ bounded เพื่อสร้าง output data ทำให้ rerun และ debug ได้โดยไม่มี side effect การประมวลผลนี้ประกอบด้วยสามส่วนหลัก คือ orchestration layer ที่กำหนดว่า job จะรันที่ไหนและเมื่อไร, storage layer ที่เก็บข้อมูล และ computation layer ที่ประมวลผลข้อมูลจริง

เราดูว่า distributed filesystem และ object store จัดการไฟล์ขนาดใหญ่ด้วย block-based replication, caching และ metadata service อย่างไร และ batch framework สมัยใหม่ทำงานกับระบบเหล่านี้ผ่าน pluggable API อย่างไร เรายังพูดถึงว่า job orchestrator schedule task, จัดสรรทรัพยากร และจัดการ fault ใน cluster ขนาดใหญ่อย่างไร และเปรียบเทียบกับ workflow orchestrator ที่ดูแล lifecycle ของ job หลายตัวที่รันอยู่ใน dependency graph

เราสำรวจ batch processing model โดยเริ่มจาก MapReduce และฟังก์ชัน map กับ reduce มาตรฐานของมัน จากนั้นหันไปดู dataflow engine อย่าง Spark และ Flink ซึ่งมี dataflow API ที่ใช้ง่ายกว่าและประสิทธิภาพดีกว่า เพื่อทำความเข้าใจว่า batch job scale ได้อย่างไร เรากล่าวถึง shuffle algorithm ซึ่งเป็นการทำงานพื้นฐานที่ทำให้ทำ grouping, joining และ aggregation ได้

เราเห็นว่าเมื่อ batch system เติบโตเต็มที่ ความสนใจก็เปลี่ยนไปที่ความง่ายในการใช้งาน จึงมีการเพิ่มการรองรับ high-level query language เช่น SQL และ DataFrame API ทำให้ batch job เข้าถึงได้ง่ายขึ้นและ optimize ได้ง่ายขึ้น batch framework จะรับ job ที่เขียนด้วยภาษาเหล่านี้ แล้วกำหนดโดยอัตโนมัติว่าจะรันอย่างมีประสิทธิภาพบน cluster ของเครื่องได้อย่างไร

เราปิดท้ายบทด้วยการสำรวจ use case ทั่วไปของ batch processing ซึ่งรวมถึง:

• ETL pipeline ที่ extract, transform และ load ข้อมูลระหว่างระบบโดยใช้ workflow ที่ตั้งเวลาไว้

• Analytics ที่ batch job รองรับทั้ง pre-aggregated query และ ad hoc query

• Machine learning ที่ใช้ batch job เตรียมและประมวลผล training dataset ขนาดใหญ่

• การ populate ระบบที่ให้บริการผู้ใช้จริงจาก batch output ซึ่งมักผ่าน stream หรือเครื่องมือ bulk-loading เพื่อให้บริการข้อมูล derived แก่ผู้ใช้

ในบทถัดไปเราจะหันไปที่ stream processing ซึ่ง input เป็น unbounded —นั่นคือ input ของ job เป็น stream ของข้อมูลที่ไม่มีวันจบ ซึ่งหมายความว่า job ไม่เคยเสร็จสมบูรณ์ เพราะงานเพิ่มเติมอาจเข้ามาได้ทุกเมื่อ เราจะเห็นว่า stream processing และ batch processing คล้ายกันในบางแง่ แต่สมมติฐานเรื่อง unbounded stream ก็มีผลอย่างมากต่อวิธีที่เราสร้างระบบ