ระบบจัดการเวิร์กโฟลว์ข้อมูล (Airflow, Dagster & Prefect)

ทฤษฎีและกลไกการทำงาน (How it works): Apache Airflow ประกอบด้วยแกนทำงานหลัก: - **Scheduler**: สแกนไดเรกทอรีไฟล์รหัส DAG เพื่อถอดสูตร และทำการบันทึกจัดเก็บประวัติสถานะลงใน **Metadata Database (เช่น PostgreSQL)** เพื่อค้นหา Task ที่พร้อมรัน จากนั้นยิงคำสั่งไปต่อคิวในระบบข้อความ (Redis/RabbitMQ) - **Executor (เช่น Celery, Kubernetes)**: ดึงคำสั่งจากคิวข้อความส่งต่อกระจายให้ **Workers** ที่มีหน้าที่เปิดรันโพรเซสทำงานจริง

# Conceptual architecture of Airflow execution loop
[DAG Definitions] -> [Scheduler Engine] <-> [PostgreSQL Metadata]
                               |
                        (Pushes to Queue)
                               |
                               v
                       [Celery/K8s Queue] -> [Worker Node 1 / Pod 1]
                    

Use Case ในชีวิตจริง (Real-world Scenario): สายงานการสกัดข้อมูลบัตรเครดิตลูกค้าที่ต้องคอยรอบรันทุกคืน ตัว Scheduler จะคอยเฝ้าเช็คประวัติใน PostgreSQL และเมื่อถึงเวลารันตีหนึ่งตรง จะสั่งจอง Workers เพื่อเปิดเครื่องประมวลผลทันที

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): สภาวะ **Database Bottleneck** เมื่อ Scheduler สแกนโฟลเดอร์ DAG ถี่เกินไปและยิงล็อกฐานข้อมูล PostgreSQL พังเสียหาย แก้ไขโดยตั้งค่าขยายเวลารอบสแกนไฟล์ `min_file_process_interval` ในไฟล์ config เพื่อลดภาระการเขียนอ่านลงคลังสเตท

ทฤษฎีและกลไกการทำงาน (How it works): ท่อส่งที่ดีต้องมีคุณสมบัติ **Idempotent** หมายความว่ารันโค้ดประมวลผลข้อมูลกี่สิบรอบด้วยพารามิเตอร์วันเดิม จะต้องได้ผลลัพธ์ปลายทางคงเดิมเสมอ ห้ามเกิดปัญหายอดข้อมูลซ้อนหรือผลลัพธ์งอกเพิ่ม นอกจากนี้ Airflow รองรับระบบ **Backfill** โดยเราสามารถระบุสั่งรันย้อนเวลาตามประวัติวันเวลารัน (Execution Date) ย้อนหลังแบบอัตโนมัติได้เมื่อต้องการเก็บข้อมูลย้อนหลังแก้โครงสร้างตาราง

# Idempotent Airflow DAG with logical execution date filters
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    "idempotent_etl_pipeline",
    start_date=datetime(2026, 7, 1),
    schedule_interval="@daily",
    catchup=False # Prevents automatic run of unexecuted past DAG runs
) as dag:
    
    # Process only partition corresponding to the execution date using Jinja templates
    process_daily_data = BashOperator(
        task_id="run_pipeline_partition",
        bash_command="python3 /opt/etl.py --date {{ ds }}"
    )
                    

Use Case ในชีวิตจริง (Real-world Scenario): ระบบวิเคราะห์ยอดขายพบคิวรีล่มในประวัติ 5 วันที่แล้ว วิศวกรข้อมูลสามารถสั่งคำสั่งรันแก้สเปคสคริปต์ และทำ Backfill เฉพาะข้อมูลพาร์ทิชัน 5 วันนั้นใหม่ได้อย่างปลอดภัยโดยไม่ทำลายโครงสร้างตารางวันอื่น

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การกำหนดวันเวลารันโดยดึงค่าโมดูลเวลาปัจจุบัน `datetime.now()` ในเนื้อโค้ดประมวลผลโดยตรง ซึ่งจะทำให้คุณสมบัติ Idempotency และระบบ Backfill ใช้งานไม่ได้ แก้ไขโดยใช้ตัวแปรแฝงระบบของ Airflow `{{ ds }}` (Logical Date) เสมอ

ทฤษฎีและกลไกการทำงาน (How it works): ในการเรียกใช้งาน Task ย่อยของสคริปต์ประมวลผลข้อมูล: - **BashOperator**: รันสคริปต์ในสภาพแวดล้อมระบบปฏิบัติการเดียวกับเซิร์ฟเวอร์ Airflow Worker ตรงๆ มีความยืดหยุ่นต่ำ เสี่ยงต่อปัญหาไลบรารีโค้ดขัดแย้งกันข้าม Task - **KubernetesPodOperator (KPOs)**: สั่งสร้างกล่องคอนเทนเนอร์ Pod แยกต่างหากบนคลัสเตอร์ Kubernetes มั่นคงปลอดภัย แยกสิทธิ์ใช้งานแรมและ CPU อย่างเป็นสัดส่วน และช่วยคัดแยกความเสี่ยงของรหัสหลุดล่มออกไปโดยสิ้นเชิง

# Declaring a secure task using KubernetesPodOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

run_spark_container = KubernetesPodOperator(
    task_id="secure_data_extraction",
    name="data-extract-pod",
    namespace="airflow-tasks",
    image="pipecraft/etl-extractor:v2.1",
    cmds=["python", "extract.py"],
    arguments=["--target-partition", "{{ ds }}"],
    get_logs=True
)
                    

Use Case ในชีวิตจริง (Real-world Scenario): ท่อส่งหนึ่งใช้ Python 3.9 ประมวลผลข้อมูล ขณะที่อีกท่อใช้ Python 3.11 รันประมวลผล MLOps การใช้งาน KPOs ช่วยให้ทั้งคู่สามารถรันงานพร้อมกันบนโหนดเดียวกันได้โดยไร้ปัญหา Dependency ขัดแย้งกัน

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การเริ่มต้นเปิดคอนเทนเนอร์ Pod ใหม่ในทุก Task ย่อยสร้างความล่าช้า (Pod startup latency overhead) แก้ไขโดยหลีกเลี่ยง KPOs กับงานประเภทประมวลผลขนาดสั้นต่ำกว่าวิ (Sub-second tasks)

ทฤษฎีและกลไกการทำงาน (How it works): ในแพลตฟอร์มระดับองค์กร ท่อส่งข้อมูลมักไม่ได้รันจบอยู่ในไฟล์เดียว แต่ต้องแยกสายงานออกจากกัน เช่น ท่อโหลดข้อมูลลูกค้ารันเสร็จก่อน จึงค่อยสั่งรันสายงานคำนวณเงินปันผล Airflow สนับสนุนการเชื่อมโยงข้ามไฟล์ผ่านเครื่องมือ **TriggerDagRunOperator** หรือเฝ้ารอคอยความสมบูรณ์ผ่าน **ExternalTaskSensor**

# Triggering another pipeline file using TriggerDagRunOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

trigger_reporting_pipeline = TriggerDagRunOperator(
    task_id="trigger_downstream_report",
    trigger_dag_id="financial_reporting_dashboard_pipeline",
    execution_date="{{ execution_date }}",
    reset_dag_run=True,
    wait_for_completion=False
)
                    

Use Case ในชีวิตจริง (Real-world Scenario): ทีมวิจัยและทีมการตลาดแยกกันดูแลจ๊อบการทำงานของระบบตนเอง เมื่อทีมข้อมูลหลักโหลดประวัติลูกค้าเข้ามาเสร็จสิ้น จึงจะยิงสัญญาณปลุกท่อวิจัยวิเคราะห์ MLOps ให้ตื่นขึ้นมารันคำนวณแคมเปญโฆษณา

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ปัญหาวงจรรันระบบรอสายค้างคา (Deadlocks) เมื่อ Sensor เฝ้ารอคอยสายงานอื่นนานเกินไปแย่งโควต้า Workers รันค้างเต็มระบบ แก้ไขโดยตั้งค่าโหมดประหยัดทรัพยากร `mode='reschedule'` เพื่อให้ Sensor ปล่อยแรม Workers ไปทำจ๊อบอื่นระหว่างคอยตรวจเช็ครอบใหม่