Dagster Pipeline กับ 12-Factor App — วิธีออกแบบ

Dagster คืออะไร

Dagster เป็น Data Orchestration Platform ที่ออกแบบมาสำหรับ Data Engineering สมัยใหม่ จุดเด่นคือแนวคิด Software-defined Assets ที่ให้คิดเรื่อง Data Pipeline ในรูปแบบของ Asset (ข้อมูลที่ต้องการสร้าง) แทนที่จะคิดเป็น Task (งานที่ต้องทำ) วิธีคิดนี้ทำให้เห็น Data Lineage ได้ชัดเจนและ Debug ปัญหาได้ง่ายกว่า

Dagster มีคุณสมบัติเด่นได้แก่ Type System สำหรับตรวจสอบ Data Quality, IO Managers สำหรับจัดการ Storage แบบ Abstract, Built-in Testing Framework, Partitions สำหรับจัดการ Incremental Processing และ Sensors/Schedules สำหรับ Trigger Pipeline อัตโนมัติ

เนื้อหาเกี่ยวข้อง — C# Minimal API Community Building — คู่มือฉบับสมบูรณ์ 2026

12-Factor App Methodology กับ Data Pipeline

12-Factor App เป็นแนวทาง 12 ข้อสำหรับออกแบบ Application ที่ Deploy ได้ง่าย Scale ได้ดี และ Maintain ได้ในระยะยาว เมื่อนำมาใช้กับ Data Pipeline จะได้ Pipeline ที่ Portable ข้าม Environment และ Reliable มากขึ้น

แนะนำเพิ่มเติม — ดูสัญญาณเทรดที่ XM Signal

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน React Server Components Testing Strategy QA

Factor	หลักการ	การนำไปใช้กับ Dagster
I. Codebase	One codebase, many deploys	เก็บ Pipeline Code ใน Git Repository เดียว Deploy ได้ทุก Environment
II. Dependencies	Explicitly declare dependencies	ใช้ requirements.txt หรือ pyproject.toml จัดการ Python Dependencies
III. Config	Store config in environment	ใช้ EnvVar ใน Dagster Resources และ dagster.yaml
IV. Backing Services	Treat as attached resources	ใช้ IO Managers สำหรับ Database, S3, GCS แบบ Abstract
V. Build, Release, Run	Strictly separate stages	Build Docker Image, Tag Version, Deploy ด้วย Helm
VI. Processes	Stateless processes	Dagster Run แต่ละตัว Stateless ใช้ External Storage สำหรับ State
VII. Port Binding	Export services via port	Dagster Webserver Bind Port สำหรับ UI และ GraphQL API
VIII. Concurrency	Scale via processes	ใช้ K8sRunLauncher สร้าง Pod แยกสำหรับแต่ละ Run
IX. Disposability	Fast startup, graceful shutdown	Pipeline Run สามารถ Cancel ได้ทันทีและ Resume จาก Checkpoint
X. Dev/Prod Parity	Keep environments similar	ใช้ Resource Configuration แยกตาม Environment
XI. Logs	Treat logs as event streams	Dagster ส่ง Log ไปยัง stdout ใช้ Structured Logging
XII. Admin	Run admin as one-off	ใช้ Dagster CLI สำหรับ One-off Tasks เช่น Backfill

การตั้งค่า Dagster Project ตาม 12-Factor

# โครงสร้าง Project

# dagster-pipeline/

# ├── dagster_pipeline/

# │   ├── __init__.py

# │   ├── assets/

# │   │   ├── __init__.py

# │   │   ├── raw_data.py

# │   │   ├── transformed.py

# │   │   └── analytics.py

# │   ├── resources/

# │   │   ├── __init__.py

# │   │   ├── database.py

# │   │   └── storage.py

# │   ├── sensors/

# │   │   └── file_sensor.py

# │   └── schedules/

# │       └── daily_schedule.py

# ├── dagster.yaml

# ├── workspace.yaml

# ├── pyproject.toml

# ├── Dockerfile

# └── helm/

#     └── values.yaml



# pyproject.toml

cat > pyproject.toml << 'TOML'

[build-system]

requires = ["setuptools"]

build-backend = "setuptools.build_meta"



[project]

name = "dagster-pipeline"

version = "1.0.0"

dependencies = [

    "dagster>=1.6.0",

    "dagster-postgres>=0.22.0",

    "dagster-k8s>=0.22.0",

    "dagster-aws>=0.22.0",

    "pandas>=2.0",

    "sqlalchemy>=2.0",

    "boto3>=1.28",

]



[project.optional-dependencies]

dev = [

    "dagster-webserver>=1.6.0",

    "pytest>=7.0",

]

TOML

Docker และ Kubernetes Deployment

# Dockerfile — Factor V: Build, Release, Run

FROM python:3.11-slim AS builder

WORKDIR /app

COPY pyproject.toml .

RUN pip install --no-cache-dir ".[dev]"

COPY . .

RUN pip install --no-cache-dir -e .



FROM python:3.11-slim

WORKDIR /app

COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages

COPY --from=builder /usr/local/bin /usr/local/bin

COPY --from=builder /app /app

EXPOSE 3000

CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "3000", "-m", "dagster_pipeline"]



---

# helm/values.yaml — Kubernetes Deployment

dagsterWebserver:

  replicaCount: 2

  image:

    repository: registry.company.com/dagster-pipeline

    tag: v1.0.0

  env:

    - name: POSTGRES_HOST

      valueFrom:

        secretKeyRef:

          name: dagster-secrets

          key: postgres-host

    - name: POSTGRES_DB

      value: dagster

    - name: POSTGRES_USER

      valueFrom:

        secretKeyRef:

          name: dagster-secrets

          key: postgres-user

    - name: POSTGRES_PASSWORD

      valueFrom:

        secretKeyRef:

          name: dagster-secrets

          key: postgres-password

    - name: S3_BUCKET

      value: company-data-lake

    - name: AWS_REGION

      value: ap-southeast-1

    - name: S3_ENDPOINT_URL

      value: ""



dagsterDaemon:

  enabled: true



runLauncher:

  type: K8sRunLauncher

  config:

    k8sRunLauncher:

      envVars:

        - POSTGRES_HOST

        - POSTGRES_DB

        - POSTGRES_USER

        - POSTGRES_PASSWORD

        - S3_BUCKET

        - AWS_REGION

      resources:

        requests:

          cpu: 250m

          memory: 512Mi

        limits:

          cpu: 1000m

          memory: 2Gi



---

# Deploy ด้วย Helm

helm repo add dagster https://dagster-io.github.io/helm

helm upgrade --install dagster dagster/dagster \

  -f helm/values.yaml \

  -n dagster --create-namespace

Testing ตาม 12-Factor

Dagster คืออะไรและต่างจาก Airflow อย่างไร

Dagster เป็น Data Orchestrator ที่เน้น Software-defined Assets คือคิดเรื่อง Data ที่ต้องการสร้างก่อนแล้วค่อยกำหนดวิธีสร้าง ต่างจาก Airflow ที่เน้น Task-based DAGs Dagster มี Type System, Built-in Testing, IO Managers และ Asset Lineage ทำให้ Debug ง่ายกว่าและ Test ได้สะดวกกว่า

แนะนำเพิ่มเติม — เรียนเทรดกับ iCafeForex

เนื้อหาเกี่ยวข้อง — แพทเทรนกราฟ — วิธีตั้งค่าและใช้งานจริงพร้อมตัวอย่าง