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

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

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

เบราว์เซอร์ที่ขอหน้าเว็บ, บริการที่เรียก API ระยะไกล, ฐานข้อมูล, cache, search indexes, และระบบอื่นๆ อีกมากมายทำงานในลักษณะนี้ เราเรียกสิ่งเหล่านี้ว่า 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 ใหม่ได้

การปฏิบัติต่อ inputs เป็น immutable และหลีกเลี่ยง side effects (เช่นการเขียนไปยัง external databases) ทำให้ batch jobs มีประสิทธิภาพที่ดี รวมถึงข้อดีอื่นๆ ดังนี้:

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

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

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

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

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

batch job อาจใช้เวลานานในการรัน: นาที ชั่วโมง หรือแม้แต่วัน Jobs อาจถูก scheduled ให้รันเป็นระยะ (เช่น วันละครั้ง) ตัววัดประสิทธิภาพหลักมักจะเป็น 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 engines ที่พบ ตัวอย่างเช่น ใน data warehouses [ 4 , 5 ] เมื่อมันถูกสร้างขึ้นมาใหม่ MapReduce เป็นก้าวสำคัญในแง่ของขนาดการประมวลผลที่สามารถทำได้บน commodity hardware แต่ปัจจุบันมันล้าสมัยเป็นส่วนใหญ่และไม่ได้ถูกใช้ที่ Google อีกต่อไป [ 6 , 7 ]

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

Cloud computing กลายเป็นสิ่งที่แพร่หลาย ชั้นเก็บข้อมูลแบบ batch กำลังเปลี่ยนจาก 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 ได้อย่างไร เราจะเห็นว่า คล้ายกับ operating system มาก distributed batch processing frameworks มี scheduler และ filesystem จากนั้นเราจะสำรวจโมเดลการประมวลผลต่างๆ ที่เราใช้ในการเขียน batch jobs สุดท้าย เราจะพูดถึงกรณีการใช้งาน 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 การตอบสนองคือ 200 (หมายถึง request สำเร็จ) และการตอบสนองมีขนาด 3,377 bytes เบราว์เซอร์คือ Chrome 137 และมันโหลดไฟล์เพราะถูกอ้างอิงในหน้าที่ URL [_https://martin.kleppmann.com/](https://martin.kleppmann.com/)_ .

แม้ว่าการแยกวิเคราะห์ log อาจดูเหมือนเป็นเรื่องสมมติ แต่มันเป็นส่วนสำคัญของการดำเนินงานของบริษัทเทคโนโลยีสมัยใหม่หลายแห่ง และถูกใช้ในทุกอย่างตั้งแต่ ad pipelines ไปจนถึงการประมวลผลการชำระเงิน อันที่จริง มันเป็นแรงผลักดันเบื้องหลังการยอมรับ 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 ของ URLs ไว้ในหน่วยความจำ โดยที่แต่ละ URL จะถูก map ไปยังจำนวนครั้งที่มันถูกพบ ตัวอย่าง Unix pipeline ไม่มี hash table แบบนั้น แต่อาศัยการ sort รายการ URLs ที่การเกิดขึ้นซ้ำของ URL เดียวกันถูกทำซ้ำๆ กัน

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

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

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

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

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 frameworks อันที่จริง คุณสามารถคิดถึง frameworks เหล่านี้ในฐานะ distributed operating systems พวกมันมี filesystems, job schedulers, และโปรแกรมที่ส่งข้อมูลถึงกันผ่าน filesystem หรือช่องทางการสื่อสารอื่นๆ

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

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

• ในระดับต่ำสุด block device drivers จะสื่อสารกับ disk โดยตรงและอนุญาตให้ชั้นที่อยู่เหนืออ่านและเขียน raw blocks ได้

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

• block API ถูกห่อหุ้มในชั้น filesystem ที่แบ่งไฟล์ขนาดใหญ่ออกเป็น blocks และติดตาม metadata ของไฟล์ เช่น inodes, directories, และ files สอง implementation ทั่วไปบน Linux คือ ext4 และ XFS เป็นต้น

• สุดท้าย operating system จะเปิดเผย filesystems ที่แตกต่างกันให้กับ applications ผ่าน API ทั่วไปที่เรียกว่า virtual filesystem (VFS) VFS คือสิ่งที่ทำให้ applications สามารถอ่านและเขียนในรูปแบบมาตรฐานได้โดยไม่ขึ้นกับ filesystem ที่อยู่เบื้องล่าง

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

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

DFS blocks ถูกอ่านโดยการส่ง network requests ไปยังเครื่องใน cluster ที่เก็บ block นั้น แต่ละเครื่องจะรัน daemon ที่เปิดเผย API ซึ่งอนุญาตให้ process ระยะไกลอ่านและเขียน blocks เป็นไฟล์บน local filesystem ของมัน HDFS เรียก daemons เหล่านี้ว่า DataNodes ในขณะที่ GlusterFS เรียก process ว่า glusterfsd เราจะเรียกพวกมันว่า data nodes ในหนังสือเล่มนี้

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

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

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

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

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

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

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

การ replication อาจหมายถึงการเก็บสำเนาของข้อมูลเดียวกันหลายชุดบนหลายเครื่อง ตามที่อธิบายใน Chapter 6 หรือการใช้ erasure coding scheme เช่น Reed–Solomon codes ซึ่งช่วยให้ข้อมูลที่สูญหายสามารถกู้คืนได้ด้วย overhead ในการจัดเก็บที่ต่ำกว่าการ replication แบบเต็ม [ 10 , 11 , 12 ] เทคนิคเหล่านี้คล้ายกับ RAID ซึ่งให้ความซ้ำซ้อนข้าม disks หลายตัวที่เชื่อมต่อกับเครื่องเดียวกัน ความแตกต่างคือใน distributed filesystem การเข้าถึงไฟล์และการ replication ทำผ่านเครือข่าย 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 drivers อนุญาตให้ผู้ใช้ปฏิบัติต่อ object stores เช่น S3 เป็น filesystem DFS implementations บางอย่าง เช่น JuiceFS และ Ceph นำเสนอทั้ง object storage และ filesystem APIs อย่างไรก็ตาม 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 ของมัน

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

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

• การดำเนินการ list แบบ prefix จะทำงานเหมือนการเรียก ls -R แบบเรียกซ้ำบนระบบ Unix มันจะคืนค่า objects ทั้งหมดที่ขึ้นต้นด้วย prefix และ objects ใน subpaths ก็จะถูกรวมด้วย

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

DFS implementations มักรองรับการดำเนินการ filesystem ทั่วไป เช่น hard links, symbolic links, file locking, และ atomic renames คุณสมบัติดังกล่าวหายไปจาก object stores การ linking และ locking มักไม่ได้รับการสนับสนุน ในขณะที่ renames ไม่ใช่ atomic; มันทำได้โดยการ copy object ไปยัง key ใหม่แล้วลบ object เก่า หากคุณต้องการ rename directory คุณต้อง rename ทุก object ภายในมันทีละรายการ เนื่องจากชื่อ directory เป็นส่วนหนึ่งของ key

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

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

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

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

Batch processing frameworks ส่ง request ไปยัง scheduler ของ orchestrator เพื่อรัน job Requests เพื่อเริ่ม job จะมี metadata เช่น:

• จำนวน tasks ที่จะ execute

• ปริมาณหน่วยความจำ, CPU, และ disk ที่ต้องการสำหรับแต่ละ task

• Job identifier

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

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

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

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

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

Task executors

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

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

Resource manager

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

Scheduler

Orchestrators มักมี centralized scheduler subsystem ซึ่งรับ requests เพื่อเริ่ม, หยุด, หรือตรวจสอบสถานะของ job ตัวอย่างเช่น scheduler อาจได้รับ request เพื่อเริ่ม job ที่มี 10 tasks โดยใช้ Docker image เฉพาะบน nodes ที่มี GPU ชนิดหนึ่ง Scheduler ใช้ข้อมูลจาก request และสถานะของ resource manager เพื่อกำหนดว่า tasks ใดจะรันบน nodes ใด จากนั้น task executors จะได้รับแจ้งงานที่ได้รับมอบหมายและเริ่ม execution

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

Note

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

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

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

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

• Scheduler อาจตัดสินใจรัน 80 tasks สำหรับแต่ละ job โดยเริ่ม 20 tasks ที่เหลือของแต่ละ job เมื่อ tasks ก่อนหน้าเสร็จสมบูรณ์

• Scheduler อาจรัน tasks ทั้งหมดของ job หนึ่งก่อน จากนั้นจึงเริ่มรัน tasks ของ job ที่สองเมื่อมี 100 cores พร้อมใช้งานเท่านั้น (กลยุทธ์ที่เรียกว่า gang scheduling )

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

นี่เป็นตัวอย่างที่ง่ายมาก แต่เราเห็น trade-offs ที่ยากหลายอย่างแล้ว ในสถานการณ์ gang scheduling เป็นต้น หาก scheduler สงวน CPU cores ไว้จนกว่าทั้ง 100 cores จะพร้อมใช้งานในเวลาเดียวกัน nodes จะว่างเปล่า การใช้ทรัพยากรของ cluster จะลดลง และ deadlock อาจเกิดขึ้นหาก jobs อื่นพยายามสงวน CPU cores เช่นกัน

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

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

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

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

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

Note

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

Workflow ของหลาย jobs อาจจำเป็นด้วยเหตุผลหลายประการ:

• หาก output ของ job หนึ่งต้องกลายเป็น input ให้กับ jobs อื่นๆ หลายตัว ซึ่งดูแลโดยทีมต่างๆ วิธีที่ดีที่สุดคือให้ job แรกเขียน output ไปยังตำแหน่งที่ jobs อื่นๆ ทั้งหมดสามารถอ่านได้ Jobs ที่เป็นผู้บริโภคเหล่านั้นสามารถถูก scheduled ให้รันทุกครั้งที่ข้อมูลนั้นถูกอัปเดตหรือตามตารางเวลาอื่น

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

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

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

อย่างไรก็ตาม โดยทั่วไปแล้ว job หนึ่งใน workflow จะเขียน output ไปยัง distributed filesystem หรือ object store และ job ถัดไปจะอ่านมันจากที่นั่น สิ่งนี้แยก jobs ออกจากกัน ทำให้พวกมันสามารถรันในเวลาที่แตกต่างกัน ถ้า job มี inputs หลายตัว workflow scheduler โดยทั่วไปจะรอจนกว่า jobs ทั้งหมดที่สร้าง inputs ของมันจะเสร็จสมบูรณ์ก่อนจึงจะรัน job ที่บริโภค inputs เหล่านั้น

Schedulers ที่พบใน orchestration frameworks เช่น ResourceManager ของ YARN หรือ scheduler ในตัวของ Spark ไม่ได้จัดการ workflows ทั้งหมด; พวกมันทำ scheduling แบบต่อ job เพื่อจัดการ dependencies เหล่านี้ระหว่างการ executions ของ job workflow schedulers ต่างๆ ได้ถูกพัฒนาขึ้น รวมถึง Airflow, Dagster, และ Prefect Workflow schedulers มีฟีเจอร์การจัดการที่มีประโยชน์เมื่อดูแลรักษา collection ขนาดใหญ่ของ batch jobs Workflows ที่ประกอบด้วย 50 ถึง 100 jobs เป็นเรื่องปกติใน data pipelines หลายแห่ง และในองค์กรขนาดใหญ่หลายทีมอาจรัน jobs หรือ workflows ที่อ่าน output ของกันและกันข้ามหลายระบบ การสนับสนุนเครื่องมือเป็นสิ่งสำคัญสำหรับการจัดการ dataflows ที่ซับซ้อนเช่นนี้

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

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

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

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

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

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

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

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

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

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

MapReduce

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

1. อ่านชุดของ input files และแบ่งมันเป็น 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 pairs ทั้งหมดตาม key ในตัวอย่าง log สิ่งนี้ทำโดยคำสั่ง sort แรก

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

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

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

Mapper

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

Reducer

MapReduce framework จะนำ key-value pairs ที่ผลิตโดย mappers รวบรวมค่าทั้งหมดที่อยู่ใน key เดียวกัน และเรียก reducer ด้วย iterator ที่วนผ่าน collection ของค่าดังกล่าว Reducer สามารถสร้าง output records (เช่นจำนวนครั้งที่ URL เดียวกันปรากฏ) Reducers สำหรับ keys ที่แตกต่างกันก็สามารถรันแบบขนานได้เช่นกัน

ในตัวอย่าง web server log เรามีคำสั่ง sort ที่สองใน step 5 ซึ่งจัดอันดับ URLs ตามจำนวน requests ใน MapReduce ถ้าคุณต้องการขั้นตอนการ sort ที่สอง คุณสามารถ implement ได้โดยการเขียน 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 functions บน lists และพวกมันได้เข้าสู่ภาษาหลักเช่น Python, Rust, และ Java

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

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

Dataflow Engines (เอนจิน Dataflow)

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

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

Dataflow APIs เหล่านี้โดยทั่วไปใช้ building blocks แบบ relational เพื่อแสดงการคำนวณ: การ join datasets ด้วยค่าของ field; การ grouping tuples ด้วย key; การ filtering ตามเงื่อนไข; และการ aggregating tuples โดยการนับ การรวม หรือฟังก์ชันอื่นๆ ภายใน การดำเนินการเหล่านี้ถูก implement โดยใช้ shuffle algorithms ที่เราจะกล่าวถึงในส่วนถัดไป

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

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

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

• เนื่องจาก joins และ data dependencies ทั้งหมดใน workflow ถูก declared อย่างชัดเจน scheduler จึงมีภาพรวมว่าต้องการข้อมูลอะไรที่ไหน ดังนั้นมันจึงสามารถทำ locality optimizations ตัวอย่างเช่น มันสามารถพยายามวาง task ที่บริโภคข้อมูลบางอย่างบนเครื่องเดียวกับ task ที่สร้างมัน เพื่อให้ข้อมูลสามารถถูกแลกเปลี่ยนผ่าน shared memory buffer แทนที่จะต้องคัดลอกผ่านเครือข่าย

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

• Operators สามารถเริ่ม execution ทันทีที่ input ของพวกมันพร้อม; ไม่จำเป็นต้องรอให้ stage ก่อนหน้าทั้งหมดเสร็จก่อนที่ stage ถัดไปจะเริ่ม

• Processes ที่มีอยู่สามารถถูกนำกลับมาใช้ใหม่เพื่อรัน operators ใหม่ ลด overhead การเริ่มต้นเมื่อเทียบกับ MapReduce (ซึ่งเริ่ม JVM ใหม่สำหรับแต่ละ task)

คุณสามารถใช้ dataflow engines เพื่อ implement การคำนวณเดียวกันกับ MapReduce workflows และพวกมันมักจะ execute ได้เร็วกว่าอย่างมีนัยสำคัญเนื่องจาก optimizations ที่อธิบายไว้ที่นี่

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

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

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

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

Shuffling เป็น foundational algorithm สำหรับ batch processors ซึ่งใช้สำหรับ joins และ aggregations MapReduce, Spark, Flink, Daft, Dataflow, และ BigQuery [ 24 ] ทั้งหมด implement shuffle algorithms ที่ scalable และมีประสิทธิภาพเพื่อจัดการ datasets ขนาดใหญ่ เราจะใช้ shuffle ใน Hadoop MapReduce [ 25 ] เพื่อเป็นตัวอย่าง illustration แต่แนวคิดในส่วนนี้สามารถประยุกต์ใช้กับระบบอื่นๆ ได้เช่นกัน

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

Figure 11-1. MapReduce job ที่มีสาม mappers และสาม reducers

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

Output ของ mapper ประกอบด้วย key-value pairs และ framework ต้องแน่ใจว่าถ้า mappers สองตัวส่ง key เดียวกัน key-value pairs เหล่านั้นจะถูกประมวลผลโดย 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 pairs ภายในแต่ละไฟล์ด้วย สิ่งนี้สามารถทำได้โดยใช้เทคนิคที่เราเห็นใน "Log-Structured Storage" : batches ของ key-value pairs ถูกเก็บครั้งแรกใน sorted data structure ในหน่วยความจำ จากนั้นเขียนออกเป็น sorted segment files และ segment files ที่เล็กกว่าจะถูก progressive merge เป็นไฟล์ที่ใหญ่ขึ้น

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

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

แม้ว่า MapReduce จะ execute shuffle step ระหว่าง map และ reduce steps ของมัน แต่ modern dataflow engines และ cloud data warehouses มีความซับซ้อนมากกว่า ระบบเช่น BigQuery ได้ optimize shuffle algorithms ของพวกมันเพื่อเก็บข้อมูลในหน่วยความจำและเขียนข้อมูลไปยัง external sorting services [ 24 ] บริการดังกล่าวเร่งความเร็ว shuffling และ replicate shuffled data เพื่อให้ความยืดหยุ่น

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

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

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

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

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

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

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

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

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

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

Query Languages (ภาษา Query)

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

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

นอกเหนือจากการใช้โค้ดน้อยกว่า MapReduce jobs ที่เขียนด้วยมือแล้ว อินเทอร์เฟซ query language เหล่านี้ยังอนุญาตให้การใช้งานแบบโต้ตอบ ซึ่งคุณเขียน analytical queries และรันมันจาก terminal หรือ GUI รูปแบบการ query แบบโต้ตอบนี้เป็นวิธีที่มีประสิทธิภาพและเป็นธรรมชาติสำหรับ business analysts, product managers, ทีมขายและการเงิน และอื่นๆ ในการสำรวจข้อมูลในสภาพแวดล้อม batch processing การรองรับ SQL ยังทำให้ distributed batch processing systems เหมาะสมสำหรับ exploratory queries

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

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

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

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

ในอดีต data warehouses รันบน specialized hardware appliances และรองรับ SQL-based analytical queries บน relational data Batch processing frameworks เช่น MapReduce ตั้งใจที่จะให้ความ scalable และ flexibility ที่มากขึ้นโดยรองรับ processing logic ที่เขียนในภาษา programming ทั่วไป อนุญาตให้อ่านและเขียนรูปแบบข้อมูลตามอำเภอใจ

เมื่อเวลาผ่านไป ทั้งสองได้กลายเป็นสิ่งที่คล้ายกันมากขึ้น Batch processing frameworks สมัยใหม่ตอนนี้รองรับ SQL เป็นภาษาสำหรับเขียน batch jobs และพวกมันทำงานได้ดีกับ relational queries โดยใช้ columnar storage formats เช่น Parquet และ optimized query execution engines (ดู "Query Execution: Compilation and Vectorization" ) ในขณะเดียวกัน data warehouses ได้เติบโต scalable มากขึ้นโดยการย้ายไปยัง cloud (ดู "Cloud Data Warehouses" ) และ implement เทคนิค scheduling, fault tolerance, และ shuffling หลายอย่างที่ distributed batch frameworks ทำ หลายตัวใช้ distributed filesystems เช่นกัน

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

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

Cloud data warehouses ก็ประสบปัญหาเกี่ยวกับ workloads บางประเภทเช่นกัน การคำนวณแบบ row-by-row จะมีประสิทธิภาพน้อยกว่าเมื่อใช้ column-oriented storage formats; alternative warehouse APIs หรือ batch processing system เป็นที่นิยมกว่าในกรณีเช่นนี้ Cloud data warehouses ยังมีแนวโน้มที่จะแพงกว่าระบบ batch processing อื่นๆ การรัน jobs ขนาดใหญ่ใน batch processing systems เช่น Spark หรือ Flink อาจคุ้มค่ากว่า

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

DataFrames

นักวิทยาศาสตร์ข้อมูลและนักสถิติโดยทั่วไปคุ้นเคยกับการทำงานกับโมเดลข้อมูล DataFrame ที่พบใน R และ Pandas (ดู "DataFrames, Matrices, and Arrays" ) DataFrame คล้ายกับตารางใน relational database: มันคือ collection ของ rows และค่าทั้งหมดใน column เดียวกันมี type เดียวกัน แทนที่จะเขียน query SQL ขนาดใหญ่หนึ่งชุด ผู้ใช้เรียกฟังก์ชันที่สอดคล้องกับ relational operators เพื่อดำเนินการ filters, joins, sorting, aggregations, และ operations อื่นๆ

เดิมที การจัดการ DataFrame มักเกิดขึ้นในเครื่อง ในหน่วยความจำ ดังนั้น DataFrames จึงถูกจำกัดให้อยู่กับ datasets ที่พอดีบนเครื่องเดียว นักวิทยาศาสตร์ข้อมูลต้องการโต้ตอบกับ datasets ขนาดใหญ่ที่พบในสภาพแวดล้อม batch processing โดยใช้ DataFrame APIs ที่พวกเขาคุ้นเคย เนื่องจาก SQL และ MapReduce ไม่เหมาะกับความต้องการของพวกเขา Frameworks การประมวลผลข้อมูลแบบกระจายเช่น Spark, Flink, และ Daft ได้นำ DataFrame APIs มาใช้เพื่อตอบสนองความต้องการนี้ อย่างไรก็ตาม implementation ของพวกมันทำงานแตกต่างออกไปบ้าง; DataFrames ในเครื่องมักจะถูก indexed และ ordered ในขณะที่ distributed DataFrames โดยทั่วไปไม่เป็นเช่นนั้น [ 29 ] สิ่งนี้สามารถนำไปสู่ความประหลาดใจด้านประสิทธิภาพเมื่อย้ายไปยัง batch frameworks

DataFrame APIs ดูคล้ายกับ dataflow APIs แต่ implementations แตกต่างกัน ในขณะที่ Pandas ดำเนินการ operations ทันทีเมื่อเรียก DataFrame methods Spark จะแปลงการเรียก DataFrame API ทั้งหมดเป็น query plan ก่อนและรัน query optimization ก่อนที่จะ execute workflow บน distributed dataflow engine ของมัน

Frameworks เช่น Daft ยังรองรับทั้ง client- และ server-side computation Operations ในหน่วยความจำขนาดเล็กจะถูก execute บน client และ datasets ที่ใหญ่กว่าจะถูกประมวลผลบน server รูปแบบ columnar storage เช่น Apache Arrow นำเสนอ unified data model ที่ทั้ง client- และ server-side execution engines สามารถแชร์กันได้

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

ตอนนี้เราได้เห็นแล้วว่า batch processing ทำงานอย่างไร มาดูกันว่ามันถูกประยุกต์ใช้กับงานต่างๆ อย่างไร Batch jobs ยอดเยี่ยมสำหรับการประมวลผล datasets ขนาดใหญ่เป็นกลุ่ม แต่มันไม่เหมาะสำหรับกรณีการใช้งานที่ต้องการ latency ต่ำ ดังนั้นคุณจะพบ batch jobs ในที่ที่มีข้อมูลจำนวนมากและความสดใหม่ของข้อมูลไม่สำคัญ สิ่งนี้อาจฟังดูจำกัด แต่ปรากฏว่างานประมวลผลข้อมูลจำนวนมากเข้ากับโมเดลนี้ ตัวอย่างเช่น:

• การทำบัญชีและการกระทบยอดสินค้าคงคลัง ซึ่งบริษัทตรวจสอบว่าธุรกรรมตรงกับบัญชีธนาคารและสินค้าคงคลัง มักดำเนินการเป็น 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 workloads) Batch jobs มักถูกใช้สำหรับ workloads ดังกล่าว โดยเฉพาะเมื่อ downstream system คือ data warehouse

ลักษณะขนานของ batch jobs ทำให้พวกมันเหมาะสมอย่างยิ่งสำหรับการแปลงข้อมูล ซึ่งส่วนใหญ่เกี่ยวข้องกับ workloads ที่ "embarrassingly parallel" การกรองข้อมูล การ projection fields และการแปลง data warehouse ทั่วไปอื่นๆ สามารถทำได้แบบขนานทั้งหมด

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

เรายังได้เห็นว่า batch jobs นั้นง่ายต่อการ troubleshoot และแก้ไขเมื่อสิ่งต่างๆ ผิดพลาด คุณสมบัตินี้มีค่ามากเมื่อ debugging data pipelines ไฟล์ที่ล้มเหลวสามารถตรวจสอบได้ง่ายเพื่อดูว่าเกิดอะไรขึ้น และ ETL batch jobs สามารถแก้ไขและ rerun ได้ ตัวอย่างเช่น ถ้า input file ไม่มีฟิลด์ที่ transformation batch job ตั้งใจจะใช้ วิศวกรข้อมูลสามารถสังเกตได้ง่ายว่าฟิลด์นั้นหายไปและอัปเดต transformation logic หรือ job ที่สร้าง input

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

Data pipelines และ analytical queries ได้เริ่มแชร์ไม่เพียงแต่ processing models แต่ยังรวมถึง execution engines ด้วย ETL batch jobs จำนวนมากตอนนี้รันบนระบบเดียวกับ analytical queries ที่อ่าน output ของพวกมัน ไม่ใช่เรื่องแปลกที่จะเห็นทั้ง data pipeline transformations และ analytical queries รันเป็น SparkSQL, Trino, หรือ DuckDB queries สถาปัตยกรรมดังกล่าวทำให้เส้นแบ่งระหว่าง application engineering, data engineering, analytics engineering, และ business analysis เลือนลางยิ่งขึ้น

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

ใน "Operational Versus Analytical Systems" เราเห็นว่า analytical queries (OLAP) มัก scan ผ่าน records จำนวนมาก ทำ groupings และ aggregations มันเป็นไปได้ที่จะรัน workloads ดังกล่าวใน batch processing system ควบคู่ไปกับ workloads batch processing อื่นๆ นักวิเคราะห์เขียน SQL queries ที่ execute บน query engine ซึ่งอ่านจากและเขียนไปยัง distributed filesystem หรือ object store Table metadata เช่น table-to-file mappings, ชื่อ, และ types ถูกจัดการด้วย table formats เช่น Apache Iceberg และ catalogs เช่น Unity (ดู "Cloud Data Warehouses" ) สถาปัตยกรรมนี้เรียกว่า data lakehouse [ 39 ]

เช่นเดียวกับ ETL การปรับปรุงใน SQL query interfaces หมายถึงหลายองค์กรตอนนี้ใช้ batch frameworks เช่น Spark สำหรับ analytics รูปแบบ query ดังกล่าวมีสองรูปแบบ:

Pre-aggregation queries

ข้อมูลถูก roll up เป็น OLAP cubes หรือ data marts เพื่อเร่ง queries (ดู "Materialized Views and Data Cubes" ) ข้อมูลที่ pre-aggregated ถูก query ใน warehouse หรือ push ไปยัง purpose-built real-time OLAP systems เช่น Apache Druid หรือ Apache Pinot การ pre-aggregation ปกติจะเกิดขึ้นตามช่วงเวลาที่กำหนด Workflow schedulers ที่กล่าวถึงใน "Scheduling workflows" ถูกใช้เพื่อจัดการ workloads เหล่านี้

Ad hoc queries

ผู้ใช้รัน queries เหล่านี้เพื่อตอบคำถามทางธุรกิจเฉพาะ สืบสวนพฤติกรรมผู้ใช้ แก้ไขปัญหาการดำเนินงาน และอื่นๆ อีกมากมาย เวลาตอบสนองมีความสำคัญสำหรับกรณีการใช้งานนี้ นักวิเคราะห์รัน queries ซ้ำๆ เมื่อพวกเขาได้รับคำตอบและเรียนรู้เพิ่มเติมเกี่ยวกับข้อมูลที่กำลังสืบสวน Batch processing frameworks ที่มีการ execution query ที่รวดเร็วช่วยลดเวลารอสำหรับนักวิเคราะห์

การรองรับ SQL ทำให้ batch processing frameworks สามารถรวมเข้ากับ spreadsheets และเครื่องมือแสดงข้อมูล เช่น Tableau, Power BI, Looker, และ Apache Superset ตัวอย่างเช่น Tableau มี SparkSQL และ Presto connectors ในขณะที่ Apache Superset รองรับ Trino, Hive, Spark SQL, Presto, และระบบอื่นๆ อีกมากมายที่ท้ายที่สุดแล้ว execute batch jobs เพื่อ query ข้อมูล

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

Machine learning (ML) ใช้ batch processing บ่อยครั้ง นักวิทยาศาสตร์ข้อมูล, ML engineers, และ AI engineers ใช้ batch processing frameworks เพื่อสืบสวนรูปแบบข้อมูล, แปลงข้อมูล, และ train ML models การใช้งานทั่วไปรวมถึง:

Feature engineering

ข้อมูลดิบถูกกรองและแปลงเป็นข้อมูลที่ models สามารถ train บนได้ โมเดลการทำนายมักต้องการข้อมูลเชิงตัวเลข ดังนั้นวิศวกรต้องแปลงข้อมูลรูปแบบอื่นๆ (เช่น text หรือค่าที่ไม่ต่อเนื่อง) เป็นรูปแบบที่ต้องการ

Model training

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

Batch inference

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

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

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

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

Batch processing ยังเป็นส่วนสำคัญของการเตรียมข้อมูลและการ training large language model ข้อมูล input ข้อความดิบ เช่น เนื้อหาของเว็บไซต์ มักอยู่ใน DFS หรือ object store ข้อมูลนี้ต้องถูก preprocessed เพื่อทำให้มันเหมาะสมสำหรับการ training ขั้นตอน preprocessing ที่เหมาะสมสำหรับ batch processing frameworks รวมถึง:

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

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

• การ tokenizing ข้อความ (แบ่งเป็นคำ) และแปลงเป็น embeddings หรือการแสดงตัวเลขของแต่ละคำ

Batch processing frameworks เช่น Kubeflow, Flyte, และ Ray ถูกสร้างขึ้นมาโดยเฉพาะสำหรับ workloads ดังกล่าว OpenAI ใช้ Ray เป็นส่วนหนึ่งของกระบวนการ training ChatGPT ตัวอย่างเช่น [ 43 ] Frameworks เหล่านี้มีการรวมในตัวสำหรับ LLM และ AI libraries เช่น PyTorch, TensorFlow, XGBoost, และอื่นๆ อีกมากมาย พวกมันยังรองรับในตัวสำหรับ feature engineering, model training, batch inference, และ fine-tuning (การปรับโมเดลพื้นฐานสำหรับกรณีการใช้งานเฉพาะ)

สุดท้าย นักวิทยาศาสตร์ข้อมูลมักทดลองกับข้อมูลใน interactive notebooks เช่น Jupyter หรือ Hex Notebooks ประกอบด้วย cells ซึ่งเป็นชิ้นส่วนเล็กๆ ของ Markdown, Python, หรือ SQL Cells ถูก execute ตามลำดับเพื่อสร้าง spreadsheets, graphs, หรือข้อมูล Notebooks จำนวนมากใช้ batch processing ผ่าน DataFrame APIs หรือ query ระบบดังกล่าวโดยใช้ SQL

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

Batch jobs มักถูกใช้เพื่อสร้าง datasets ที่ precomputed หรือ derived เช่น คำแนะนำสินค้า, รายงานที่ผู้ใช้เห็น, และ features สำหรับ ML models Datasets เหล่านี้มักถูกให้บริการจาก production database, key-value store, หรือ search engine ไม่ว่าจะใช้ระบบใด ข้อมูลที่ precomputed ต้องเดินทางจาก distributed filesystem หรือ object store ของ batch processor กลับไปยังฐานข้อมูลที่ให้บริการ traffic จริง

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

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

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

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

ทางออกที่ดีกว่าคือให้ batch jobs push datasets ที่ precomputed ไปยัง streams เช่น Kafka topics ซึ่งเราจะกล่าวถึงเพิ่มเติมใน Chapter 12 Search engines เช่น Elasticsearch, real-time OLAP systems เช่น Apache Pinot และ Apache Druid, derived datastores เช่น Venice [ 45 ], และ cloud data warehouses เช่น ClickHouse ทั้งหมดมีความสามารถในตัวในการรับข้อมูลจาก Kafka เข้าสู่ระบบของพวกมัน การ push ข้อมูลผ่าน streaming system แก้ไขปัญหาบางอย่างที่กล่าวถึงข้างต้น:

• Streaming systems ถูกปรับให้เหมาะสมสำหรับ sequential writes ซึ่งทำให้พวกมันเหมาะสมกว่าสำหรับ workload การเขียนเป็นกลุ่มของ batch job

• Streaming systems สามารถทำหน้าที่เป็น buffer ระหว่าง batch job และ production databases ระบบ downstream สามารถ throttle อ่าน rate ของพวกมันเพื่อให้แน่ใจว่าพวกมันสามารถให้บริการ traffic การผลิตได้อย่างสบายๆ ต่อไป

• Output ของ batch job เดียวสามารถถูกบริโภคโดยหลาย downstream systems

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

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

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

การสร้าง databases ใน batch และ bulk-import ข้อมูลนั้นเร็วมากและทำให้ระบบสลับระหว่างเวอร์ชันของ dataset ได้ง่ายขึ้น ในทางกลับกัน มันอาจท้าทายในการอัปเดต datasets แบบ incremental จาก batch jobs ที่สร้าง databases ใหม่ทั้งหมด เป็นเรื่องปกติที่จะใช้แนวทางแบบผสมเมื่อต้องการทั้ง bootstrapping และ incremental loads ตัวอย่างเช่น Venice รองรับ hybrid stores ที่อนุญาตให้ทั้ง batch row-based updates และ full dataset swaps

Summary (สรุป)

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

จากนั้นเราขยายไปยัง distributed batch processing systems Batch frameworks ประมวลผล datasets input ที่ immutable และ bounded เพื่อสร้าง output data ทำให้สามารถ rerun และ debugging ได้โดยไม่มี side effects การประมวลผลนี้เกี่ยวข้องกับสามส่วนประกอบหลัก: orchestration layer ที่กำหนดว่า job จะรันที่ไหนและเมื่อไหร่, storage layer สำหรับเก็บข้อมูล, และ computation layer ที่ประมวลผลข้อมูลจริง

เราดูว่า distributed filesystems และ object stores จัดการไฟล์ขนาดใหญ่ผ่าน block-based replication, caching, และ metadata services อย่างไร และ modern batch frameworks โต้ตอบกับระบบเหล่านี้ผ่าน pluggable APIs อย่างไร เรายังพูดถึงว่า job orchestrators schedule tasks, จัดสรรทรัพยากร, และจัดการ faults ใน clusters ขนาดใหญ่ได้อย่างไร และเราเปรียบเทียบพวกมันกับ workflow orchestrators ที่จัดการ lifecycle ของ collection ของ jobs ที่รันใน dependency graph

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

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

เราจบบทด้วยการสำรวจกรณีการใช้งาน batch processing ทั่วไป รวมถึง:

• ETL pipelines ซึ่ง extract, transform, และ load ข้อมูลระหว่างระบบโดยใช้ scheduled workflows

• Analytics ซึ่ง batch jobs รองรับทั้ง pre-aggregated และ ad hoc queries

• Machine learning ซึ่ง batch jobs ใช้เพื่อเตรียมและประมวลผล datasets การฝึกขนาดใหญ่

• การ populate ระบบที่ให้บริการการผลิตจาก batch outputs มักผ่าน streams หรือ bulk-loading tools เพื่อให้บริการข้อมูลที่ derived แก่ผู้ใช้

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