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

ทฤษฎีและกลไกการทำงาน (How it works): Apache Airflow มีสถาปัตยกรรมแบบแยกส่วน: - **Webserver**: ให้บริการหน้า UI สำหรับผู้ใช้งานเพื่อดูสถานะ DAGs และจัดการระบบ - **Scheduler**: สแกนไฟล์ DAG อย่างต่อเนื่อง ตัดสินใจว่า Task ไหนถึงเวลารัน และบันทึกสถานะลง **Metadata Database (เช่น PostgreSQL)** - **Executors**: กลไกการรัน Task ที่รับคำสั่งจาก Scheduler - **CeleryExecutor**: ใช้ Message Broker (Redis/RabbitMQ) กระจายงานให้ Worker nodes ที่รันรออยู่ตลอดเวลา เหมาะกับงานที่คงที่ - **KubernetesExecutor**: สร้าง Pod ใหม่ขึ้นมาทำงานแต่ละ Task แยกกันโดยเฉพาะ (Dynamic provisioning) คืนทรัพยากรทันทีเมื่อจบงาน เหมาะกับงานที่มีไลบรารีต่างกันและต้องการแยก Resource ชัดเจน

# airflow.cfg - Executor configuration
[core]
executor = KubernetesExecutor
# executor = CeleryExecutor
                    

Use Case ในชีวิตจริง (Real-world Scenario): บริษัทอีคอมเมิร์ซที่มี Data Pipeline แตกต่างกัน 500 เส้น เลือกใช้ KubernetesExecutor เพื่อแยกการจัดการ Library (เช่น เส้น A ใช้ Pandas 1.0, เส้น B ใช้ Pandas 2.0) โดยที่ Pod จะถูกสร้างและทำลายทิ้งทันทีเมื่อ Task เสร็จสิ้น ทำให้ไม่เปลืองทรัพยากร

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หากสเกล Worker ไม่พอใน CeleryExecutor งานจะไปกองที่คิว (Queue buildup) จนเกิดคอขวด แก้ไขโดยการตั้งค่าการสเกล Worker อัตโนมัติด้วย KEDA หรือย้ายไปใช้ KubernetesExecutor ที่รองรับการสเกลตามโควต้าของคลัสเตอร์

ทฤษฎีและกลไกการทำงาน (How it works): ท่อส่งที่ดีต้องมีคุณสมบัติ **Idempotent** หมายความว่ารันโค้ดประมวลผลข้อมูลกี่สิบรอบด้วยพารามิเตอร์วันเดิม จะต้องได้ผลลัพธ์ปลายทางคงเดิมเสมอ ห้ามเกิดปัญหายอดข้อมูลซ้อนหรือผลลัพธ์งอกเพิ่ม นอกจากนี้ Airflow รองรับระบบ **Backfill** โดยเราสามารถระบุสั่งรันย้อนเวลาตามประวัติวันเวลารัน (Logical Date / 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 on deployment
) as dag:
    
    # Process only partition corresponding to the execution date using Jinja templates ({{ ds }})
    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 วันนั้นใหม่ได้อย่างปลอดภัย โดยสคริปต์จะใช้ตัวแปร `{{ ds }}` ในการสั่ง Delete และ Insert ข้อมูลเก่าทิ้งโดยไม่ทำลายโครงสร้างตารางวันอื่น

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

ทฤษฎีและกลไกการทำงาน (How it works): Dynamic Task Mapping เป็นฟีเจอร์ (ใน Airflow 2.3+) ที่อนุญาตให้สร้าง Task ย่อยแบบไดนามิกตอนรันไทม์ (Runtime) ตามข้อมูลหรืออาร์เรย์ที่ถูกส่งมาจาก Task ก่อนหน้า โดยใช้ฟังก์ชัน `.expand()` และ `.partial()` ซึ่งแตกต่างจากการใช้ `for loop` ธรรมดาตอนนิยาม DAG ที่จะคงที่ (Static) ไปตลอด

# Dynamic Task Mapping Example
from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2026, 1, 1), schedule_interval=None)
def dynamic_mapping_dag():
    @task
    def get_files():
        return ["file1.csv", "file2.csv", "file3.csv"] # Returns list at runtime

    @task
    def process_file(file_name: str):
        print(f"Processing {file_name}")

    # Dynamically maps a task for each file in the list returned by get_files
    process_file.expand(file_name=get_files())

dynamic_mapping_dag()
                    

Use Case ในชีวิตจริง (Real-world Scenario): สคริปต์สแกนตรวจสอบว่าวันนี้มีไฟล์ลูกค้าอัปโหลดเข้ามาใน S3 กี่ไฟล์ หากมี 10 ไฟล์ Airflow จะสร้าง 10 Tasks คู่ขนานกันไปรันให้เองทันที หากมี 0 ไฟล์ก็จะข้ามไป ทำให้มีความยืดหยุ่นสูงมากโดยไม่ต้องแก้โค้ด DAG

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หาก Task ก่อนหน้าส่งลิสต์ที่มีสมาชิกจำนวนมหาศาล (เช่น 10,000 ไฟล์) จะทำให้ Scheduler และ Metadata Database สร้าง Task ล้นจนช้า (Overload) แก้ไขโดยตั้งค่า `max_map_length` หรือจับกลุ่ม (Chunk) ลิสต์ให้มีขนาดพอเหมาะ (เช่น กลุ่มละ 500 ไฟล์)

ทฤษฎีและกลไกการทำงาน (How it works): - **Custom Operators**: หากตรรกะในธุรกิจมีความเฉพาะตัว การสืบทอด (Inherit) คลาส `BaseOperator` มาเขียนพฤติกรรมเองจะช่วยลดโค้ดซ้ำซ้อน - **KubernetesPodOperator (KPOs)**: คือ Operator มาตรฐานสำหรับการสั่งสร้างกล่องคอนเทนเนอร์ Pod แยกต่างหากบนคลัสเตอร์ Kubernetes คัดแยกความเสี่ยงของรหัสหลุดล่มออกไปโดยสิ้นเชิง

# Creating a Custom Operator
from airflow.models import BaseOperator

class MyAPIETLOperator(BaseOperator):
    def __init__(self, api_endpoint, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.api_endpoint = api_endpoint

    def execute(self, context):
        # Specific business logic to extract and load
        print(f"Extracting data from {self.api_endpoint} for date {context['ds']}")
                    

Use Case ในชีวิตจริง (Real-world Scenario): ทีมวิศวกรสร้าง `CompanySecurityScanOperator` เป็น Custom Operator เพื่อบังคับให้ทุกท่อส่งข้อมูลต้องเรียกใช้เช็คความปลอดภัยก่อนดึงข้อมูล ทำให้การใช้งานง่ายและเป็นมาตรฐานเดียวกันทั้งบริษัท

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