เกิดอะไรขึ้นจริง ๆ เมื่อเรียกใช้งาน Spark Job?

บทนำ

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

Apache Spark ช่วยจัดการกับความท้าทายเหล่านี้ด้วยการเป็นเอนจินประมวลผลข้อมูลแบบกระจาย ซึ่งแบ่งงานออกเป็นพาร์ทิชันและประมวลผลงานหลายรายการพร้อมกันผ่านคอร์ CPU และเครื่องหลายเครื่อง

อย่างไรก็ตาม Spark ไม่ได้หมายถึงเพียงการเพิ่มจำนวนเครื่องเท่านั้น

การทำความเข้าใจว่าองค์ประกอบต่าง ๆ ทำงานร่วมกันอย่างไร ทั้งไดรเวอร์ เอ็กเซคิวเตอร์ RDD, DataFrame, พาร์ทิชัน งาน ขั้นตอน และทาสก์ เป็นสิ่งสำคัญต่อการออกแบบไปป์ไลน์ข้อมูลที่มีประสิทธิภาพและเชื่อถือได้

ในบทความนี้ ผมจะอธิบายสถาปัตยกรรม กลไกการทำงาน และปัจจัยสำคัญที่ส่งผลต่อประสิทธิภาพของ Spark โดยใช้ตัวอย่างการรวมยอดขายอีคอมเมิร์ซที่นำไปประยุกต์ใช้ได้จริง

1. เหตุใดจึงใช้ Apache Spark — การประมวลผลข้อมูลแบบขนาน

Apache Spark คือเอนจินประมวลผลแบบกระจายที่ออกแบบมาสำหรับการประมวลผลข้อมูลและวิศวกรรมข้อมูลขนาดใหญ่

Spark ประมวลผลข้อมูลขนาดใหญ่ด้วยการประมวลผลข้อมูลแบบขนาน:

ภาพ A. สถาปัตยกรรมคลัสเตอร์แบบกระจายของ Apache Spark ในภาพรวม (คลัสเตอร์ ไดรเวอร์ และโหนดผู้ปฏิบัติงาน) โดยเปรียบเทียบกับการประมวลผลบนโหนดเดียว (จัดทำโดย Kuriko IWAI)

เมื่อเทียบกับการประมวลผลบนโหนดเดียว (ด้านขวาใน ภาพ A) Spark จะแบ่งงานออกเป็นทาสก์และประมวลผลบนหลายโหนด โดยใช้ทรัพยากรที่มีอยู่ข้ามเครื่องต่าง ๆ

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

2. สถาปัตยกรรมหลักของ Apache Spark

Spark ประกอบด้วยองค์ประกอบหลักหลายส่วน ดังนี้

• ไดรเวอร์ Spark (ไดรเวอร์): กระบวนการหลักที่ทำหน้าที่ประสานงานแอปพลิเคชัน Spark
• ตัวจัดการคลัสเตอร์: บริการภายนอกที่รับผิดชอบจัดสรรทรัพยากรฮาร์ดแวร์ เช่น คอร์ CPU และหน่วยความจำ
• โหนดผู้ปฏิบัติงาน (โหนดผู้ปฏิบัติงาน): เครื่องจริงหรือเครื่องเสมือน หรือโหนดประมวลผลในคลัสเตอร์
• เอ็กเซคิวเตอร์: กระบวนการผู้ปฏิบัติงานแบบกระจายที่เริ่มทำงานบนโหนดผู้ปฏิบัติงาน
• Java Virtual Machine (JVM): สภาพแวดล้อมขณะทำงานที่จำเป็นต่อการเรียกใช้ไบต์โค้ดของ Spark (เนื่องจาก Spark เขียนด้วย Scala/Java)

ขั้นแรก ไดรเวอร์ จะประสานงานแอปพลิเคชัน Spark และแปลงแอปพลิเคชันนั้นเป็นกราฟแบบมีทิศทางและไม่มีวงจร (DAG) ซึ่งประกอบด้วยทาสก์สำหรับการประมวลผลจริง จากนั้นจึงจัดตารางทาสก์เหล่านั้นให้ทำงานบนเอ็กเซคิวเตอร์

จากนั้น ตัวจัดการคลัสเตอร์ จะจัดสรรทรัพยากรตามระบบจัดการทรัพยากรของการติดตั้งใช้งาน

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

สภาพแวดล้อมที่มีผู้ให้บริการจัดการบางแห่ง เช่น Databricks ใช้เอ็กเซคิวเตอร์หนึ่งตัวต่อโหนดผู้ปฏิบัติงานหนึ่งโหนด แต่รูปแบบนี้ไม่ได้เป็นกฎที่ใช้กับ Spark ทุกกรณี โหนดผู้ปฏิบัติงานหนึ่งโหนดอาจรองรับเอ็กเซคิวเตอร์หลายตัวได้ ขึ้นอยู่กับการกำหนดค่าคลัสเตอร์

2-1. กรณีศึกษาอีคอมเมิร์ซ — โครงสร้างข้อมูลขาเข้า

ลองนึกถึงบริษัทอีคอมเมิร์ซบูติกแห่งหนึ่ง

บริษัทมีระเบียนการขาย 10,000 รายการ โดยแต่ละรายการมีรหัสสินค้า (SKU) และจำนวนสินค้าที่ขาย สมมติว่ามี SKU ที่เป็นไปได้สามรายการ: A, B, และ C.

ตาราง 1. ตัวอย่างการแจกแจงระเบียนการขายอีคอมเมิร์ซและยอดรวมที่คาดว่าจะได้

เป้าหมายของเราคือคำนวณจำนวนหน่วยสินค้าที่ขายได้ทั้งหมดของแต่ละ SKU:

1. กรองระเบียนที่ qty <= 0, จากนั้น
2. รวมข้อมูลตัวอย่างตาม SKU

ในชุดข้อมูลเต็มที่ใช้ประกอบตัวอย่าง สมมติว่าหลังการกรองแล้วเหลือระเบียน 9,800 รายการจากทั้งหมด 10,000 รายการ ยอดรวมสุดท้ายที่คาดว่าจะได้จาก Spark คือ:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. คลัสเตอร์ Spark

ผมจะกำหนดคลัสเตอร์แบบง่ายสำหรับกรณีศึกษานี้

โหนดผู้ปฏิบัติงาน

ผมจะกำหนดให้มีโหนดผู้ปฏิบัติงาน 3 โหนด โดยแต่ละโหนดเป็นอินสแตนซ์ที่มี 8 vCPU และ RAM 32 GiB:

ภาพ B. การจัดสรรทรัพยากรระหว่างโหนดผู้ปฏิบัติงาน (จัดทำโดย Kuriko IWAI)

โดยรวมแล้ว ระบบมี vCPU ให้ใช้งาน 24 คอร์

การจัดสรร CPU/หน่วยความจำให้ Spark Executors

สมมติว่าฉันจะจัดสรรดังนี้:

• เอ็กซีคิวเตอร์ 7 ตัว
• vCPU 2 คอร์ต่อเอ็กซีคิวเตอร์หนึ่งตัว
• หน่วยความจำฮีป JVM 3 GiB ต่อเอ็กซีคิวเตอร์หนึ่งตัว

แอปพลิเคชัน Spark ขอให้ตัวจัดการคลัสเตอร์จัดสรรทรัพยากรให้เอ็กซีคิวเตอร์ทั้ง 7 ตัวนี้

จากนั้น ตัวจัดการคลัสเตอร์จะตัดสินใจจัดสรร CPU และหน่วยความจำตามการใช้ทรัพยากรของ worker แต่ละตัว

ตัวอย่างเช่น ตัวจัดการคลัสเตอร์ตัดสินใจสร้างช่องสำหรับงาน 14 ช่อง โดยคำนวณจาก vCPU 2 คอร์ต่อเอ็กซีคิวเตอร์ × เอ็กซีคิวเตอร์ 7 ตัว:

ภาพ C-1 การจัดสรรคอร์ vCPU หน่วยความจำ และ Executor ระหว่าง Worker โดยมีทรัพยากรไม่สมดุลใน Worker A

โปรดทราบว่าหน่วยความจำ 3 GiB ที่จัดสรรให้ Executor ไม่ใช่หน่วยความจำทั้งหมดที่ Executor ใช้ได้ เพราะต้องสำรองหน่วยความจำส่วนเกินไว้สำหรับ Worker และการทำงานอื่น ๆ ด้วย

สมมติว่างานหนึ่งงานต้องใช้ 1 vCPU (spark.task.cpus = 1) worker แต่ละตัวจะจัดสรร vCPU ดังนี้:

ภาพ C-2 การจัดสรร vCPU แบบขนานระหว่าง Worker

ประเด็นสำคัญมี 2 ข้อดังนี้:

1. worker แต่ละตัวมีอัตราการใช้งาน CPU แตกต่างกัน (ใน ภาพ C-2 Worker A ใช้งาน 75% ขณะที่ Worker B และ C ยังคงอยู่ที่ 50%)
2. แม้ระบบจะมี vCPU ทั้งหมด 24 คอร์ (8 vCPU/worker × 3 Workers) แต่ใช้ได้เพียง 14 คอร์ ส่วนที่เหลืออีก 10 คอร์ไม่ได้ถูกใช้กับงานเหล่านี้

2-3. Java Virtual Machine (JVM) เบื้องหลัง Spark

Apache Spark สร้างขึ้นบน Java Virtual Machine (JVM)

JVM คือสภาพแวดล้อมขณะทำงานที่จำเป็นต่อการเรียกใช้ไบต์โค้ดของ Spark

เนื่องจาก Spark เขียนด้วย Scala/Java:

• เมื่อใช้ Scala หรือ Java ในแอปพลิเคชัน: โค้ดจะทำงานภายใน JVM โดยตรง
• เมื่อใช้ PySpark (Python): โพรเซส Python จะเรียกใช้โค้ด แต่สื่อสารกับ driver ที่ทำงานบน JVM ผ่านบริดจ์ เช่น Py4J ส่วน Spark executors บน worker nodes ยังคงทำงานภายใน JVM ซึ่งจะจัดการ worker ของ Python เพื่อเรียกใช้โค้ด Python และ ฟังก์ชันที่ผู้ใช้กำหนดใน Python (Python UDFs)

ไม่ว่าจะเป็นกรณีใด JVM จะจัดการเธรดการทำงาน การจัดสรรหน่วยความจำ และการเก็บกวาดขยะให้ทั้ง driver และ executors

JVM ของ Driver และ Executor ในแอปพลิเคชัน PySpark

ในแอปพลิเคชัน PySpark ตัวขับเคลื่อน Python จะสื่อสารกับ Spark driver ที่ทำงานบน JVM

• ตัวขับเคลื่อน Python: ส่วนที่ใช้เขียนและเรียกใช้โค้ด PySpark ระดับแอปพลิเคชัน
• Driver JVM: ประสานงานการทำงานและการจัดตารางเวลาของ Spark
• Executor JVM: เรียกใช้งาน Spark และจัดการทรัพยากรสำหรับการทำงาน โดยจะใช้โพรเซส worker ของ Python เมื่องานต้องประมวลผลด้วย Python
• Client JVM (โพรเซสไคลเอนต์หรือเครื่องที่ใช้ส่งงาน): JVM ที่ทำงานบนเครื่องในระบบหรือโหนด edge/gateway

ดังที่แสดงในภาพ client JVM จะเชื่อมต่อกับ driver JVM ก่อน จากนั้นจึงประสานงานกับ executor JVMs:

ภาพ D ความสัมพันธ์โดยรวมระหว่าง client, driver และ executor JVM

จากจุดนี้ driver JVM จะโฮสต์ Spark Driver เพื่อจัดการแอปพลิเคชัน ขณะที่ executor JVMs จะโฮสต์เอ็กซีคิวเตอร์แต่ละตัวซึ่งมีหน้าที่เรียกใช้งานเหล่านั้น

JVM สำหรับ driver แต่ละตัวมีหน่วยความจำฮีป JVM เพื่อจัดการการประมวลผลข้อมูล การแคช และออบเจ็กต์ของผู้ใช้ได้อย่างมีประสิทธิภาพ

2-4. เจาะลึก Resilient Distributed Datasets (RDDs)

Resilient Distributed Dataset (RDD) หรือชุดข้อมูลแบบกระจายที่ทนทานต่อความผิดพลาด คือ ชุดองค์ประกอบที่แบ่งเป็นพาร์ทิชันและไม่เปลี่ยนแปลง ซึ่ง Spark สามารถประมวลผลแบบขนานได้

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

ชื่อนี้สื่อถึงคุณสมบัติ 3 ประการ:

• ทนทานต่อความผิดพลาด: Spark บันทึก สายที่มาของข้อมูล ซึ่งระบุว่า RDD ใดขึ้นอยู่กับ RDD ก่อนหน้าใด และมักสร้างพาร์ทิชันที่หายไปขึ้นใหม่ได้ โดยไม่ต้องเก็บสำเนาของผลลัพธ์ระหว่างทางทุกชุด
• แบบกระจาย: ชุดข้อมูลจะแสดงเป็นพาร์ทิชัน ซึ่ง Spark อาจประมวลผลบน executors คนละตัวกันได้ แต่ ไม่จำเป็นต้อง ใช้เครื่องจริงหลายเครื่อง เพราะโหมดภายในเครื่องก็รองรับ RDD เช่นกัน
• ชุดข้อมูล: RDD ประกอบด้วยคู่คีย์-ค่า จำนวนเต็ม สตริง ทูเพิล และองค์ประกอบประเภทอื่น ๆ ที่รองรับได้ ไม่ได้จำกัดเฉพาะแถวข้อมูลเชิงสัมพันธ์ API ของ RDD มีการดำเนินการต่าง ๆ เช่น map, filter, reduceByKey, joinและ aggregateByKey.

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

พาร์ทิชัน

พาร์ทิชันคือส่วนย่อยของข้อมูลในเชิงตรรกะ และโดยปกติจะสอดคล้องกับหนึ่งทาสก์ภายในสเตจ

สมมติว่าเราสร้างพาร์ทิชัน 100 พาร์ทิชัน (การกำหนดพาร์ทิชันเริ่มต้น 100 พาร์ทิชันนี้เป็นเพียงตัวเลือกเพื่อใช้ในการอธิบาย)

เมื่อมีองค์ประกอบต้นทาง 10,000 รายการ แต่ละพาร์ทิชันจะมีประมาณ 100 รายการ หากกระจายข้อมูลขาเข้าอย่างทั่วถึงโดยประมาณ:

ภาพ E. การจัดสรรพาร์ทิชันแบบขนานจากการแบ่งข้อมูลเริ่มต้น 100 ส่วน

3. API ของ RDD เทียบกับ API ของ DataFrame: การปรับแต่งประสิทธิภาพภายในด้วย Catalyst

ตัวอย่าง RDD ดั้งเดิมช่วยให้เข้าใจลำดับสายงาน พาร์ทิชัน และการชัฟเฟิล

อย่างไรก็ตาม สำหรับงาน ETL แบบมีโครงสร้างทั่วไป API ของ DataFrame เป็นจุดเริ่มต้นที่แนะนำ เพราะมีคอลัมน์ ข้อมูลสคีมา นิพจน์คล้าย SQL และการดำเนินการที่ตัวปรับแต่งประสิทธิภาพสามารถนำไปใช้ได้

แผนภาพด้านล่างแสดงความสัมพันธ์ดังกล่าว:

ภาพ F. ความสัมพันธ์ระหว่างแนวคิดกราฟของ Spark ได้แก่ ลำดับสายงานของ RDD, DataFrame และ DAG สำหรับการประมวลผลของ Spark

เช่นเดียวกับที่ RDD มีลำดับสายงาน RDD เพื่อกำหนดการแปลงข้อมูล แผนคิวรีของ DataFrame ก็ใช้แทนคิวรีที่มีโครงสร้าง

Spark สร้างงานสำหรับการประมวลผลจากการพึ่งพากันและตัวดำเนินการทางกายภาพ

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

3-1. แผนคิวรีของ DataFrame

แผนคิวรีของ DataFrameคือลำดับขั้นตอนเชิงตรรกะและเชิงกายภาพที่เอนจินประมวลผลสร้างขึ้น (กรอบสีม่วงในภาพ L)

คิวรีของ DataFrame จะผ่านการแทนค่าในหลายรูปแบบ:

1. แผนเชิงตรรกะ: คำอธิบายการดำเนินการที่ร้องขอ (สแกน กรอง จัดกลุ่ม รวมผล)
2. แผนเชิงตรรกะที่ผ่านการวิเคราะห์และปรับแต่งแล้ว: Spark แก้ไขการอ้างอิงและใช้การปรับแต่งเชิงตรรกะที่เกี่ยวข้อง เช่น การตัดคอลัมน์ที่ไม่จำเป็นออก หรือการดันเงื่อนไขตัวกรองลงไป
3. แผนเชิงกายภาพ: Spark เลือกตัวดำเนินการที่เรียกใช้ได้ เช่น การสแกน การรวมผลบางส่วน/ขั้นสุดท้าย การแลกเปลี่ยนข้อมูล การเชื่อมตาราง และการเรียงลำดับ
4. การประมวลผล: แผนเชิงกายภาพจะทำงานโดยใช้กลไกการประมวลผลของ Spark ซึ่งท้ายที่สุดจะเกี่ยวข้องกับทาสก์บนตัวดำเนินการ

ในกรณีใช้งานของเรา แผนเชิงกายภาพอาจประกอบด้วยขั้นตอนต่อไปนี้:

• การรวมผลแฮชบางส่วน,
• การแลกเปลี่ยนข้อมูลตาม SKU และ
• การรวมผลแฮชขั้นสุดท้าย

3-2. เปรียบเทียบการดำเนินการ: API ของ RDD เทียบกับ API ของ DataFrame

ลองเปรียบเทียบ API ของ RDD กับ API ของ DataFrame โดยใช้การรวมผลตาม SKU เป็นตัวอย่าง

API ของ RDD

องค์ประกอบต่าง ๆ เป็น (sku, qty) คู่ข้อมูล ตรรกะการรวมผลใน Python ของเราระบุไว้อย่างชัดเจน และ Spark จะทำตามการพึ่งพากันของ RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

API ของ DataFrame

สมมติว่า sales_df มีคอลัมน์ sku (ชนิดสตริง) และ qty (ชนิดตัวเลข):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

ในที่นี้ ตัวอย่างทั้งสองแบบแก้ปัญหาทางธุรกิจเดียวกัน:

กรองรายการขายที่ไม่ถูกต้อง และรวมปริมาณที่ถูกต้องตาม SKU

และในทั้งสองกรณี collect() จะเริ่มการประมวลผลของ Spark และส่งระเบียนผลลัพธ์ทั้งหมดกลับไปยังไดรเวอร์

ความแตกต่างสำคัญในที่นี้อยู่ที่รูปแบบการแทนผลลัพธ์:

• API ของ RDD: ส่งคืนองค์ประกอบ Python เช่น (sku, total_qty) ทูเพิล
• API ของ DataFrame: ส่งคืนออบเจ็กต์แบบมีโครงสร้าง Row ที่มีฟิลด์พร้อมชื่อ เช่น sku และ total_qty.

ตารางด้านล่างเปรียบเทียบ API ของ RDD และ DataFrame โดยรวม:

ตารางที่ 2. เปรียบเทียบคุณลักษณะและระดับนามธรรมของ RDD API กับ DataFrame API (จัดทำโดย Kuriko IWAI)

4. สรุป

เป้าหมายของการปรับแต่ง Spark ไม่ใช่การเพิ่มจำนวนพาร์ทิชันหรือการใช้งาน CPU ให้สูงสุดโดยแยกจากกัน

แต่คือการลดต้นทุนและเวลาแฝงของการประมวลผลตั้งแต่ต้นจนจบ โดยไม่ทำให้ระบบขาดเสถียรภาพ

ในบทความนี้ เราได้เรียนรู้ความสัมพันธ์หลัก ๆ ภายในระบบดังนี้

• ทรัพยากร: Worker เป็นโฮสต์ให้กับ executor; executor เรียกใช้ task; การกำหนดค่า CPU และหน่วยความจำจะจำกัดการทำงานพร้อมกัน
• การเขียนโปรแกรมด้วย RDD: การแปลงข้อมูลจะสร้างคำจำกัดความของ RDD ที่ไม่เปลี่ยนแปลงและลำดับการสืบทอดที่มา; การดำเนินการจะเริ่มการคำนวณ
• การเขียนโปรแกรมด้วย DataFrame: การแปลงข้อมูลจะสร้างการดำเนินการของคิวรีเชิงตรรกะ; การปรับแต่งจะเลือกตัวดำเนินการเชิงกายภาพที่นำไปใช้ได้; การดำเนินการจะเริ่มการคำนวณ
• การประมวลผล: Spark จัดงานออกเป็น job, stage และ task โดย task จะประมวลผลพาร์ทิชันบน executor การชัฟเฟิลและการพึ่งพาอื่น ๆ ทำให้เกิดจุดแบ่งที่ต้องประสานงานกันระหว่าง stage

การทำความเข้าใจวิธีที่ Spark ประมวลผลเวิร์กโหลดแบบกระจายเป็นเพียงพื้นฐานเท่านั้น

คุณสามารถเรียนรู้ลำดับการคำนวณ 7 ขั้นตอนโดยใช้ RDD APIได้