การสร้าง Data Pipeline ด้วย Apache Spark Declarative Pipelines (Delta Live Tables) บน Databricks เป็นแนวทางที่เน้นการกำหนดผลลัพธ์ปลายทางที่ต้องการ (What) แทนการเขียนคำสั่งควบคุมทีละขั้นตอน (How) ระบบจะบริหารจัดการโครงสร้างพื้นฐาน ลำดับการประมวลผล และการตรวจสอบคุณภาพข้อมูลให้อัตโนมัติ แนวทางนี้ช่วยลดความซับซ้อนของโค้ด รองรับทั้งข้อมูลแบบ Batch และ Streaming ทำให้วิศวกรข้อมูลสามารถพัฒนาระบบและส่งมอบข้อมูลที่พร้อมใช้งานทางธุรกิจได้อย่างรวดเร็ว ถูกต้อง และมีเสถียรภาพ

1. Databricks คืออะไร?

Databricks คือแพลตฟอร์มวิเคราะห์ข้อมูลแบบรวมศูนย์ (Unified Data Analytics Platform) ที่พัฒนาขึ้นโดยทีมผู้สร้าง Apache Spark โดยมีสถาปัตยกรรมหลักที่เรียกว่า Lakehouse ซึ่งเป็นการนำข้อดีของ Data Lake (เก็บข้อมูลได้ทุกประเภท ขยายขนาดได้ง่าย คุ้มค่า) มารวมกับ Data Warehouse (การจัดการข้อมูลอย่างมีโครงสร้าง มีประสิทธิภาพในการคิวรี และรองรับ ACID Transactions)

แพลตฟอร์มนี้ทำงานอยู่บน Cloud (เช่น AWS, Azure, GCP) มีหน้าที่หลักในการจัดการโครงสร้างพื้นฐาน (Infrastructure) และคลัสเตอร์สำหรับการประมวลผล Big Data โดยอัตโนมัติ ทำให้นักพัฒนา (Data Engineer, Data Scientist, Data Analyst) สามารถเข้ามาทำงานร่วมกันในพื้นที่เดียว (Workspace) ได้อย่างมีประสิทธิภาพ โดยไม่ต้องกังวลเรื่องการตั้งค่าเซิร์ฟเวอร์หรือการจูนระบบที่ซับซ้อน

2. Apache Spark Declarative Pipelines (ใน Databricks คือ Delta Live Tables)

เมื่อพูดถึงหัวข้อ "Build Data Pipelines with Apache Spark Declarative Pipelines" บน Databricks เรากำลังพูดถึงเครื่องมือที่เรียกว่า Delta Live Tables (DLT)

ในอดีต การเขียน Data Pipeline จะเป็นแบบ Imperative คือนักพัฒนาต้องเขียนโค้ดระบุ "วิธีการ (How)" ทีละขั้นตอน เช่น ดึงข้อมูลมาอย่างไร จัดการคลัสเตอร์อย่างไร ถ้าพังจะรันใหม่ตรงไหน และเขียนข้อมูลลงตารางอย่างไร ซึ่งมีความซับซ้อนและดูแลรักษายาก

แนวคิดแบบ Declarative Pipelines คือการเปลี่ยนมาบอกระบบแค่ว่า "เราต้องการอะไร (What)" เท่านั้น โดยนักพัฒนาเพียงแค่ "ประกาศ" (Declare) ว่าตารางปลายทางควรมีหน้าตาแบบไหน และดึงข้อมูลมาจากไหน ส่วนงานที่เหลือ ระบบจัดการให้ทั้งหมด ได้แก่:

• Orchestration & DAG: สร้างลำดับการประมวลผล (Dependency Graph) ให้อัตโนมัติ ว่าต้องรันตารางไหนก่อนหลัง

• Infrastructure Management: จัดการขนาดของคลัสเตอร์ และสเกลระบบตามปริมาณข้อมูลให้อัตโนมัติ

• Data Quality (Expectations): ฝังการตรวจสอบคุณภาพข้อมูล (Data Quality Rules) ไว้ใน Pipeline ได้เลย หากข้อมูลผิดปกติสามารถเลือกได้ว่าจะ ทิ้ง (Drop), แจ้งเตือน (Warn), หรือหยุดการทำงาน (Fail)

• Batch & Streaming Unification: จัดการข้อมูลทั้งแบบ Batch และ Streaming ได้ด้วยโครงสร้างโค้ดเดียวกัน

3. ตัวอย่างการทำงาน (Code Example)

สถาปัตยกรรมที่นิยมใช้ร่วมกับ Declarative Pipelines คือ Medallion Architecture (Bronze: ข้อมูลดิบ, Silver: ข้อมูลที่ทำความสะอาดแล้ว, Gold: ข้อมูลพร้อมใช้สำหรับ Business/BI)

ด้านล่างนี้คือตัวอย่างการเขียน Delta Live Tables (DLT) ด้วย PySpark โดยใช้แนวคิด Declarative

import dlt
from pyspark.sql.functions import *

# ---------------------------------------------------------
# Bronze Layer: Ingest Raw Data
# ---------------------------------------------------------
@dlt.table(
    name="bronze_sales",
    comment="Raw sales data loaded from cloud storage in real-time."
)
def bronze_sales():
    # Read raw JSON data streams using Databricks Auto Loader
    return (spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/mnt/raw-data/sales/"))

# ---------------------------------------------------------
# Silver Layer: Clean Data & Apply Data Quality Rules
# ---------------------------------------------------------
@dlt.table(
    name="silver_sales",
    comment="Cleaned sales data with Data Quality expectations applied."
)
# Expectation: Drop the record if order_id is null
@dlt.expect_or_drop("valid_order_id", "order_id IS NOT NULL")
# Expectation: Fail the entire pipeline if amount is less than or equal to 0
@dlt.expect_or_fail("valid_amount", "amount > 0")
def silver_sales():
    # Read as a stream from the Bronze table and add a processing timestamp
    return (dlt.read_stream("bronze_sales")
            .withColumn("processed_time", current_timestamp()))

# ---------------------------------------------------------
# Gold Layer: Aggregate Data for Business / Dashboard
# ---------------------------------------------------------
@dlt.table(
    name="gold_daily_sales_summary",
    comment="Aggregated daily sales ready for BI dashboards."
)
def gold_daily_sales():
    # Read from Silver table and aggregate total sales by date
    return (dlt.read("silver_sales")
            .groupBy("order_date")
            .agg(sum("amount").alias("total_daily_sales")))

อธิบายจากตัวอย่าง: จากโค้ดด้านบน เราจะเห็นว่าเราไม่ต้องเขียนโค้ดเพื่อตั้งเวลาการรัน (Cron Job) ไม่ต้องเขียนโค้ดจัดการ Checkpoint สำหรับข้อมูล Streaming และไม่ต้องเขียนโค้ดเพื่อเช็คว่าต้องรัน bronze ให้เสร็จก่อนจึงจะมารัน silver ได้ เพราะ dlt.read() จะทำหน้าที่สร้างความสัมพันธ์ (Dependencies) ให้โดยอัตโนมัติ การพัฒนา Pipeline จึงสั้นลง อ่านเข้าใจง่าย และมีความน่าเชื่อถือมากขึ้นอย่างมาก