Workshop PySpark: ประมวลผล Big Data ด้วย Python ตั้งแต่พื้นฐานถึง Data Pipeline

• 1. ผลลัพธ์การเรียนรู้
• 2. PySpark คืออะไร และเหมาะกับงานแบบไหน?
  • สถาปัตยกรรมแบบย่อ
  • Transformation, Action และ Lazy Evaluation
• 3. เตรียมสภาพแวดล้อม
  • 3.1 ตรวจสอบเครื่อง
  • 3.2 ติดตั้งบน macOS / Linux
  • 3.3 ติดตั้งบน Windows PowerShell
  • 3.4 ตรวจสอบ SparkSession
• 4. โจทย์ Workshop: Sales Analytics Pipeline
  • 4.1 โครงสร้างไฟล์
  • 4.2 สร้างชุดข้อมูล data/orders.csv
• 5. สร้างแอปพลิเคชัน ETL แบบสมบูรณ์
  • 5.1 อธิบายขั้นตอนสำคัญ
• 6. รัน Workshop และตรวจสอบผลลัพธ์
• 7. ทดลอง Structured Streaming (เสริม)
• 8. การวัดและปรับประสิทธิภาพ (Performance Tuning)
  • ตัวอย่างดู Physical Plan
• 9. ปัญหาที่พบบ่อย (Troubleshooting)
• 10. แบบฝึกหัดท้าย Workshop
  • แนวคิดสำหรับการขยายเป็นโปรเจกต์จริง
• 11. สรุป
• เอกสารอ้างอิง

Workshop ภาษาไทย | ระดับเริ่มต้น–กลาง | เวลา 90–120 นาที
Stack: Python 3.10+, Java 17+, PySpark 4.2.0
ตัวอย่าง: วิเคราะห์ข้อมูลยอดขายด้วย ETL Pipeline บน Spark DataFrame และ Spark SQL

การประมวลผลข้อมูลนับล้านแถวจากหลายแหล่งเป็นปัญหาที่เกิดขึ้นบ่อยในงาน Data Engineering โดยเฉพาะเมื่อข้อมูลมีขนาดเกินกว่าที่จะจัดการด้วย Python บนคอมพิวเตอร์เครื่องเดียวได้อย่างมีประสิทธิภาพ PySpark ช่วยให้เราเขียนโค้ด Python เพื่อใช้งานพลังประมวลผลแบบกระจายของ Apache Spark โดยไม่ต้องจัดการการแจกงานให้แต่ละเครื่องด้วยตนเอง

บทความนี้จะพัฒนาระบบวิเคราะห์ยอดขายจำลองตั้งแต่การอ่านไฟล์ CSV การตรวจสอบคุณภาพข้อมูล การคำนวณยอดขาย การสรุปผลด้วย DataFrame/SQL การเชื่อมข้อมูลลูกค้า การจัดอันดับ และการบันทึกเป็น Parquet พร้อมตัวอย่าง Streaming สำหรับศึกษาต่อยอด ไม่จำเป็นต้องติดตั้ง Cluster เพราะใช้โหมด Local ได้

#1. ผลลัพธ์การเรียนรู้

เมื่อทำ Workshop นี้จบ ผู้อ่านจะสามารถ:

1. อธิบายโครงสร้างการทำงานของ Spark Driver, Executors, SparkSession และ DataFrame
2. แยกความแตกต่างระหว่าง Transformation กับ Action และเข้าใจ Lazy Evaluation
3. ติดตั้ง PySpark และรัน Application บนเครื่องของตนเอง
4. อ่าน CSV ด้วย Schema ที่กำหนดชัดเจน และทำ Data Cleaning / ETL
5. ใช้ groupBy(), agg(), join(), Spark SQL และ Window Function วิเคราะห์ข้อมูล
6. เขียน/อ่านผลลัพธ์ด้วย Apache Parquet และตรวจสอบความถูกต้องของข้อมูล
7. เริ่มต้นใช้งาน Structured Streaming และดู Execution Plan เพื่อวิเคราะห์ประสิทธิภาพ

#2. PySpark คืออะไร และเหมาะกับงานแบบไหน?

Apache Spark คือเอนจินประมวลผลข้อมูลทั่วไปที่ทำงานแบบกระจาย (Distributed Computing Engine) ส่วน PySpark คือ Python API สำหรับใช้งาน Spark จึงสามารถเขียน Data Pipeline ด้วย Python และนำงานไปขยายจาก Local Mode สู่ Cluster ได้

ข้อควรรู้: PySpark ไม่ได้เร็วกว่าการใช้ Pandas ในทุกงาน หากข้อมูลเล็กและอยู่ในหน่วยความจำได้สบาย Pandas หรือ DuckDB อาจเหมาะสมกว่า ควรตัดสินใจจากขนาดข้อมูล ปริมาณงาน และผลการ Benchmark

#สถาปัตยกรรมแบบย่อ

flowchart LR
    A[Python Application] --> B[Driver / SparkSession]
    B --> C[Cluster Manager]
    C --> D[Executor 1]
    C --> E[Executor 2]
    C --> F[Executor N]
    D <--> G[(Data Sources / Storage)]
    E <--> G
    F <--> G

• Driver: ควบคุม Spark Application วางแผนการทำงาน และประสานงาน Tasks
• Cluster Manager: จัดสรรทรัพยากรเมื่อติดตั้งแบบ Cluster เช่น Standalone, Kubernetes หรือ YARN
• Executor: ประมวลผล Tasks และจัดเก็บข้อมูลระหว่างการคำนวณตามแผนงาน
• SparkSession: Entry Point สำหรับ DataFrame, SQL และการเข้าถึงแหล่งข้อมูล
• DataFrame: ข้อมูลเชิงตารางที่มี Schema และ Spark สามารถแบ่งงานประมวลผลได้

ใน Workshop นี้ตั้งค่า .master("local[*]") เพื่อรันบนเครื่องเดียว โดยไม่ต้องตั้ง Cluster Manager แยก

#Transformation, Action และ Lazy Evaluation

filtered = df.filter(df.status == "completed")  # Transformation
result = filtered.groupBy("category").count()    # Transformation
result.show()                                      # Action

คำสั่ง Transformation เช่น select, filter, join และ groupBy โดยทั่วไปเป็นการประกาศแผนประมวลผล ส่วน Action เช่น show, count, collect และ write กระตุ้นให้ Spark ดำเนินงาน Spark ใช้ Catalyst Optimizer ช่วยวางแผน Query ก่อนทำงานจริง

#3. เตรียมสภาพแวดล้อม

เอกสาร PySpark 4.2.0 ระบุข้อกำหนดเป็น Python 3.10 ขึ้นไป และ Java 17 ขึ้นไป สำหรับการรัน PySpark แบบปกติ

#3.1 ตรวจสอบเครื่อง

python --version
java -version

หากตรวจไม่พบ Java ให้ติดตั้ง OpenJDK 17 หรือ 21 ก่อน แล้วตั้ง JAVA_HOME ให้ชี้ไปยังโฟลเดอร์ติดตั้ง Java เมื่อจำเป็น

#3.2 ติดตั้งบน macOS / Linux

mkdir pyspark-web-blog-workshop
cd pyspark-web-blog-workshop
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install pyspark==4.2.0

#3.3 ติดตั้งบน Windows PowerShell

mkdir pyspark-web-blog-workshop
cd pyspark-web-blog-workshop
py -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install pyspark==4.2.0

หาก PowerShell ไม่อนุญาตให้ Activate สามารถใช้ Command Prompt และคำสั่ง .venv\Scripts\activate.bat โดยไม่ต้องเปลี่ยน Execution Policy ทั้งระบบ

ตรวจสอบแพ็กเกจ:

python -c "import pyspark; print(pyspark.__version__)"

#3.4 ตรวจสอบ SparkSession

บันทึกไฟล์ hello_spark.py:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("HelloPySpark")
    .getOrCreate()
)

spark.createDataFrame(
    [("Python", 120), ("Java", 90), ("Go", 75)],
    ["language", "students"]
).show()

spark.stop()

รัน python hello_spark.py หากเห็นตาราง 3 แถว แสดงว่าสภาพแวดล้อมพร้อมใช้งาน

#4. โจทย์ Workshop: Sales Analytics Pipeline

สมมติว่าระบบร้านค้าออนไลน์บันทึกคำสั่งซื้อเป็น CSV ต้องการสร้าง Pipeline เพื่อ:

1. Extract: อ่านคำสั่งซื้อและกำหนดชนิดข้อมูลอย่างชัดเจน
2. Transform: กรองเฉพาะออร์เดอร์สำเร็จ ตรวจสอบข้อมูล และคำนวณยอดขาย
3. Analyze: สรุปยอดขายตามหมวดสินค้าและภูมิภาค
4. Enrich: Join ข้อมูลลูกค้า และหา Top 2 ออร์เดอร์ในแต่ละหมวด
5. Load: ส่งออกข้อมูลที่ผ่านการแปลงเป็น Parquet พร้อม Validate ผลลัพธ์

flowchart LR
    A[(orders.csv)] --> B[Read with Schema]
    B --> C[Validate and Clean]
    C --> D[Revenue Calculation]
    D --> E[Aggregate / Join / SQL]
    E --> F[(Parquet)]
    F --> G[Verification]

#4.1 โครงสร้างไฟล์

pyspark-web-blog-workshop/
├── data/
│   └── orders.csv
├── sales_pipeline.py
├── streaming_demo.py
└── output/                 # สร้างเมื่อรัน Pipeline
    └── sales_parquet/

สำหรับผู้ดาวน์โหลดชุด Workshop: ไฟล์ orders.csv, sales_pipeline.py และ streaming_demo.py เตรียมไว้แล้ว สามารถข้ามขั้นตอนพิมพ์ไฟล์และรันตามหัวข้อที่ 5 ได้

#4.2 สร้างชุดข้อมูล data/orders.csv

ใช้ Text Editor สร้างไฟล์ CSV (UTF-8) ดังนี้:

order_id,customer_id,category,region,unit_price,quantity,status
1,101,Laptop,Bangkok,32000.00,1,completed
2,102,Accessories,Chiang Mai,900.00,2,completed
3,101,Accessories,Bangkok,1500.00,3,completed
4,103,Laptop,Bangkok,26000.00,1,cancelled
5,104,Monitor,Khon Kaen,7200.00,2,completed
6,105,Monitor,Bangkok,5500.00,1,completed
7,102,Laptop,Chiang Mai,28000.00,1,completed
8,104,Accessories,Khon Kaen,650.00,4,completed

รายละเอียดคอลัมน์:

เพื่อหลีกเลี่ยงปัญหาเลขทศนิยมแบบ Floating Point ในการคำนวณยอดเงิน Workshop นี้ใช้ DecimalType ไม่ใช่ DoubleType

#5. สร้างแอปพลิเคชัน ETL แบบสมบูรณ์

บันทึกโค้ดต่อไปนี้เป็น sales_pipeline.py ที่ระดับเดียวกับโฟลเดอร์ data โค้ดแบ่งเป็นฟังก์ชัน build_sales() เพื่อให้แยกทดสอบและนำกลับมาใช้ใหม่ได้

"""Hands-on PySpark 4.2 sales ETL workshop. Run from this folder."""

from decimal import Decimal
from pathlib import Path

from pyspark.sql import SparkSession, Window, functions as F
from pyspark.sql.types import (
    DecimalType,
    IntegerType,
    StringType,
    StructField,
    StructType,
)

BASE_DIR = Path(__file__).resolve().parent
SOURCE = BASE_DIR / "data" / "orders.csv"
DESTINATION = BASE_DIR / "output" / "sales_parquet"

ORDER_SCHEMA = StructType([
    StructField("order_id", IntegerType(), True),
    StructField("customer_id", IntegerType(), True),
    StructField("category", StringType(), True),
    StructField("region", StringType(), True),
    StructField("unit_price", DecimalType(12, 2), True),
    StructField("quantity", IntegerType(), True),
    StructField("status", StringType(), True),
])

CUSTOMER_SCHEMA = StructType([
    StructField("customer_id", IntegerType(), False),
    StructField("customer_name", StringType(), False),
])


def build_sales(orders):
    """Keep valid completed orders and calculate line totals."""
    return (
        orders
        .filter(F.col("status") == "completed")
        .filter(F.col("order_id").isNotNull())
        .filter(F.col("customer_id").isNotNull())
        .filter(F.col("category").isNotNull())
        .filter(F.col("region").isNotNull())
        .filter(F.col("unit_price").isNotNull() & (F.col("unit_price") >= 0))
        .filter(F.col("quantity").isNotNull() & (F.col("quantity") > 0))
        .dropDuplicates(["order_id"])
        .withColumn("line_total", F.col("unit_price") * F.col("quantity"))
    )


def main():
    spark = (
        SparkSession.builder
        .appName("PySparkSalesWorkshop")
        .master("local[*]")
        .config("spark.sql.shuffle.partitions", "4")  # For this tiny local lab only
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("WARN")

    try:
        orders = (
            spark.read
            .option("header", True)
            .option("mode", "FAILFAST")
            .schema(ORDER_SCHEMA)
            .csv(str(SOURCE))
        )
        print("=== Input ===")
        orders.printSchema()
        orders.show(truncate=False)

        sales = build_sales(orders)
        print("=== Valid completed sales ===")
        sales.select("order_id", "category", "region", "line_total").show()

        by_category = (
            sales.groupBy("category")
            .agg(
                F.sum("line_total").alias("revenue"),
                F.count("order_id").alias("order_count"),
            )
            .orderBy(F.desc("revenue"))
        )
        print("=== Revenue by category ===")
        by_category.show()

        customers = spark.createDataFrame(
            [(101, "Arun"), (102, "Ben"), (103, "Dao"),
             (104, "Fah"), (105, "Nok")],
            schema=CUSTOMER_SCHEMA,
        )
        enriched = sales.join(F.broadcast(customers), on="customer_id", how="left")
        print("=== Enriched orders ===")
        enriched.select("order_id", "customer_name", "line_total").show()

        window = Window.partitionBy("category").orderBy(
            F.desc("line_total"), F.asc("order_id")
        )
        top_2_by_category = (
            sales.withColumn("row_num", F.row_number().over(window))
            .filter(F.col("row_num") <= 2)
            .select("category", "order_id", "line_total", "row_num")
            .orderBy("category", "row_num")
        )
        print("=== Top 2 orders in each category ===")
        top_2_by_category.show()

        sales.createOrReplaceTempView("sales")
        by_region = spark.sql("""
            SELECT region, SUM(line_total) AS revenue
            FROM sales
            GROUP BY region
            ORDER BY revenue DESC
        """)
        print("=== Revenue by region (Spark SQL) ===")
        by_region.show()

        sales.write.mode("overwrite").parquet(str(DESTINATION))
        restored = spark.read.parquet(str(DESTINATION))
        print("=== Parquet verification ===")
        print("Read rows:", restored.count())

        # The collect() below is safe only because there are 3 grouped rows.
        actual = {
            row["category"]: row["revenue"]
            for row in by_category.collect()
        }
        expected = {
            "Laptop": Decimal("60000.00"),
            "Monitor": Decimal("19900.00"),
            "Accessories": Decimal("8900.00"),
        }
        assert actual == expected, f"Unexpected totals: {actual}"
        assert restored.count() == 7, "Expected 7 completed orders"
        print("PASS: totals and Parquet row count are correct")

        print("=== Execution plan ===")
        by_category.explain(mode="formatted")
    finally:
        spark.stop()


if __name__ == "__main__":
    main()

#5.1 อธิบายขั้นตอนสำคัญ

(1) อ่าน CSV โดยกำหนด Schema

orders = (
    spark.read.option("header", True)
    .option("mode", "FAILFAST")
    .schema(ORDER_SCHEMA)
    .csv(str(SOURCE))
)

การกำหนด Schema ลดความคลาดเคลื่อนของชนิดข้อมูลเมื่ออ่านหลายไฟล์ ส่วน FAILFAST ทำให้หยุดเมื่อเจอระเบียนที่ผิดรูปแบบตามเงื่อนไขการอ่าน แทนการเงียบแล้วปล่อยข้อมูลเสียผ่านไป

(2) ทำ Data Cleaning และคำนวณยอดขาย

sales = build_sales(orders)
sales.select("order_id", "category", "line_total").show()

ฟังก์ชันนี้กรอง completed, ตรวจสอบค่าว่างที่สำคัญ ตรวจสอบ quantity > 0 และ unit_price >= 0 ตัดรายการซ้ำจาก order_id แล้วคำนวณ line_total = unit_price × quantity

(3) Aggregate และ Join

by_category = sales.groupBy("category").agg(
    F.sum("line_total").alias("revenue"),
    F.count("order_id").alias("order_count")
)

joined = sales.join(F.broadcast(customers), "customer_id", "left")

groupBy มักเกี่ยวข้องกับ Shuffle ส่วน broadcast เหมาะเมื่ออีก DataFrame มีขนาดเล็กเพียงพอ ไม่ควร Broadcast ตารางขนาดใหญ่โดยไม่ดูแผนการประมวลผล

(4) Window Function และ Spark SQL

row_number() ภายใต้ Window.partitionBy("category") ใช้จัดอันดับออร์เดอร์แยกตามหมวดสินค้า ขณะที่ createOrReplaceTempView("sales") ทำให้เขียน SQL เช่น SUM และ GROUP BY กับข้อมูลเดียวกันได้

(5) บันทึก Parquet

sales.write.mode("overwrite").parquet(str(DESTINATION))
restored = spark.read.parquet(str(DESTINATION))

Spark มักเขียนผลลัพธ์เป็น โฟลเดอร์ที่มีไฟล์ part-...parquet หลายไฟล์ ตามจำนวน Partitions ซึ่งเป็นพฤติกรรมปกติ

#6. รัน Workshop และตรวจสอบผลลัพธ์

เมื่ออยู่ในโฟลเดอร์โปรเจกต์และเปิด Virtual Environment แล้ว:

python sales_pipeline.py

ผลสรุปยอดขายตามหมวดสินค้าที่ ควรได้จากข้อมูลตัวอย่าง (หน่วย: บาท):

ผลสรุปตามภูมิภาค:

ตรวจสอบว่าสคริปต์แสดงข้อความ:

PASS: totals and Parquet row count are correct

สาเหตุที่ได้ 7 รายการ: ออร์เดอร์หมายเลข 4 ถูกยกเลิก (cancelled) และไม่ถูกนับในยอดขาย

หากต้องการดู Spark Web UI ขณะโปรแกรมกำลังทำงาน ให้เปิด http://localhost:4040 (หากพอร์ตนี้ว่าง) ซึ่งใช้ดูรายละเอียด Jobs, Stages, Executors และ SQL ได้ โดย UI มักปิดเมื่อ Spark Application สิ้นสุด

#7. ทดลอง Structured Streaming (เสริม)

Spark Structured Streaming รองรับการเขียน Pipeline สำหรับข้อมูลที่ไหลเข้ามาต่อเนื่อง โดยสามารถใช้ DataFrame API ใกล้เคียงกับงาน Batch การทดลองนี้ใช้แหล่งข้อมูลจำลอง rate ไม่ต้องเชื่อม Kafka หรือสร้าง Server ภายนอก

สร้างไฟล์ streaming_demo.py:

"""Optional PySpark Structured Streaming lab with built-in rate source."""
from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder
    .appName("PySparkStreamingDemo")
    .master("local[*]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

query = None
try:
    events = (
        spark.readStream
        .format("rate")
        .option("rowsPerSecond", 2)
        .load()
        .withColumn("group_id", F.col("value") % 3)
    )
    query = (
        events.writeStream
        .format("console")
        .outputMode("append")
        .option("truncate", False)
        .start()
    )
    query.awaitTermination(15)  # Demonstration duration, not an exact row count
finally:
    if query is not None:
        query.stop()
    spark.stop()

รัน:

python streaming_demo.py

คาดว่าจะเห็นข้อมูล timestamp, value และ group_id แสดงออกมาบน Console เป็นรอบ ๆ ตาม Micro-batch ที่ Spark ประมวลผล จำนวนแถวจริงขึ้นอยู่กับการเริ่มต้นและเวลาที่ใช้รัน จึงไม่ควรคาดหวังจำนวนแถวที่ตายตัว

#8. การวัดและปรับประสิทธิภาพ (Performance Tuning)

ข้อควรระวัง: ค่า spark.sql.shuffle.partitions=4 ใน Workshop ตั้งไว้เพื่อทดลองด้วยข้อมูลขนาดเล็กเท่านั้น ไม่ใช่ค่าแนะนำสำหรับ Production และหลีกเลี่ยง collect()/toPandas() กับ DataFrame ขนาดใหญ่มาก เพราะข้อมูลจะถูกนำกลับมาที่ Driver

#ตัวอย่างดู Physical Plan

by_category.explain(mode="formatted")

หากทดลองตัด F.broadcast(customers) ออกแล้วดู enriched.explain(mode="formatted") อาจเห็น Join Strategy ต่างกัน ขึ้นกับ Spark Optimizer ขนาดข้อมูล และการตั้งค่า Adaptive Query Execution (AQE) จึงควรอธิบายผลจาก Plan จริง ไม่ใช่คาดเดาจากโค้ดเพียงอย่างเดียว

#9. ปัญหาที่พบบ่อย (Troubleshooting)

#10. แบบฝึกหัดท้าย Workshop

พื้นฐาน: เพิ่มสรุปค่าเฉลี่ยยอดขายต่อคำสั่งซื้อ (avg) แยกตามหมวด และกรองคำสั่งซื้อที่มี quantity >= 2

ระดับกลาง: สรุปลูกค้า 3 อันดับแรกตามยอดซื้อสะสม โดยเริ่มจาก DataFrame enriched และเพิ่ม Window Function เพื่อคำนวณอันดับ

ระดับสูง: เพิ่มคอลัมน์ order_date ในไฟล์ CSV แล้วสรุปยอดขายรายวัน/รายเดือนด้วย Spark SQL พร้อมเขียน Parquet แยก Partition ตามเดือน

โจทย์คุณภาพข้อมูล: เพิ่มรายการที่มี quantity = -1 หรือราคาว่าง แล้วตรวจว่าระบบคัดออกได้ถูกต้อง และเพิ่ม Assertions ให้ครอบคลุมกรณีเหล่านี้

#แนวคิดสำหรับการขยายเป็นโปรเจกต์จริง

• เปลี่ยนแหล่งข้อมูลจาก CSV เป็น Object Storage เช่น S3 หรือ Data Lake ที่เหมาะกับสถาปัตยกรรม
• ใช้ Airflow หรือ Orchestrator อื่นเพื่อกำหนดเวลารัน ETL
• เพิ่ม Monitoring, Metrics, Data Quality Checks และ Logging
• พิจารณา Delta Lake หรือ Apache Iceberg สำหรับความสามารถด้านตาราง เช่น Transaction และ Schema Evolution (ต้องมีการตั้งค่าเพิ่มเติม)
• ทดสอบ Performance ด้วยข้อมูลจำนวนมากและแยก Dev/Production Configuration

#11. สรุป

PySpark เป็นเครื่องมือสำคัญของ Data Engineering เพราะเชื่อม Python เข้ากับ Distributed Processing ของ Apache Spark การฝึกครั้งนี้เริ่มจาก SparkSession และ DataFrame ก่อนต่อยอดสู่ Data Quality, Aggregation, Join, Window Function, Spark SQL และ Parquet รวมทั้งได้เห็นวิธีตรวจสอบผลลัพธ์เชิงปริมาณ เมื่อเข้าใจพื้นฐานแล้วสามารถขยับไปทำ Pipeline สำหรับ Data Lake, Structured Streaming หรือระบบบน Cloud ได้

#เอกสารอ้างอิง

1. Apache Spark, PySpark 4.2.0 Installation
2. Apache Spark, PySpark DataFrame User Guide
3. Apache Spark, PySpark User Guide
4. Apache Spark, SparkSession API Reference
5. Apache Spark, Structured Streaming Programming Guide
6. Apache Spark, Apache Spark 4.2.0 Documentation

เผยแพร่: 14 ตุลาคม 2026 | บทความนี้ใช้ข้อมูลสมมติเพื่อการศึกษา ไม่ใช่ข้อมูลยอดขายจริง