Dagster Pipeline กับ Team Productivity — วิธีใช้ Dagster เพิ่ม Productivity ของทีม Data

Dagster Data Orchestrator

Dagster เป็น Open-source Data Orchestrator เน้น Software-defined Assets ใช้ Python สร้าง Pipelines มี Type System ตรวจ Data Quality มี UI แสดง Asset Lineage

เพิ่ม Team Productivity ด้วย Asset-based Approach ทีมเห็น Data Lineage ชัดเจน แบ่ง Code Locations ตามทีม Reusable Asset Factories

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง ZFS on Linux Team Productivity

Dagster Assets และ Jobs

# === Dagster Software-defined Assets ===

# pip install dagster dagster-webserver dagster-postgres dagster-duckdb



import dagster as dg

from dagster import (

 asset, AssetExecutionContext, MaterializeResult,

 MetadataValue, AssetKey, Definitions,

 ScheduleDefinition, SensorDefinition,

)

from typing import Dict, List

import json



# 1. Software-defined Assets

@asset(

 description="ดึงข้อมูล Orders จาก API",

 group_name="raw",

 metadata={"source": "REST API", "format": "JSON"},

)

def raw_orders(context: AssetExecutionContext) -> Dict:

 """Extract Orders from API"""

 context.log.info("Fetching orders from API...")

 # response = requests.get("https://api.example.com/orders")

 # orders = response.json()

 orders = [

 {"id": i, "amount": 100 + i * 10, "status": "completed"}

 for i in range(100)

 ]

 context.log.info(f"Fetched {len(orders)} orders")

 return {"orders": orders, "count": len(orders)}



@asset(

 description="ดึงข้อมูล Customers จาก Database",

 group_name="raw",

 metadata={"source": "PostgreSQL", "table": "customers"},

)

def raw_customers(context: AssetExecutionContext) -> Dict:

 """Extract Customers from Database"""

 customers = [

 {"id": i, "name": f"Customer {i}", "tier": "gold" if i % 3 == 0 else "silver"}

 for i in range(50)

 ]

 return {"customers": customers, "count": len(customers)}



@asset(

 description="รวม Orders กับ Customers",

 group_name="transformed",

 deps=[AssetKey("raw_orders"), AssetKey("raw_customers")],

)

def enriched_orders(

 context: AssetExecutionContext,

 raw_orders: Dict,

 raw_customers: Dict,

) -> Dict:

 """Join Orders with Customers"""

 orders = raw_orders["orders"]

 customers = {c["id"]: c for c in raw_customers["customers"]}



 enriched = []

 for order in orders:

 customer = customers.get(order["id"] % 50, {})

 enriched.append({

 **order,

 "customer_name": customer.get("name", "Unknown"),

 "customer_tier": customer.get("tier", "standard"),

 })



 context.log.info(f"Enriched {len(enriched)} orders")

 return {"orders": enriched, "count": len(enriched)}



@asset(

 description="สรุปยอดขายรายวัน",

 group_name="analytics",

 deps=[AssetKey("enriched_orders")],

)

def daily_sales_summary(

 context: AssetExecutionContext,

 enriched_orders: Dict,

) -> MaterializeResult:

 """Aggregate Daily Sales"""

 orders = enriched_orders["orders"]

 total_revenue = sum(o["amount"] for o in orders)

 avg_order = total_revenue / len(orders) if orders else 0



 by_tier = {}

 for o in orders:

 tier = o["customer_tier"]

 by_tier[tier] = by_tier.get(tier, 0) + o["amount"]



 context.log.info(f"Total Revenue: {total_revenue}")



 return MaterializeResult(

 metadata={

 "total_revenue": MetadataValue.float(total_revenue),

 "total_orders": MetadataValue.int(len(orders)),

 "avg_order_value": MetadataValue.float(avg_order),

 "revenue_by_tier": MetadataValue.json(by_tier),

 }

 )



# 2. Schedule

daily_schedule = ScheduleDefinition(

 name="daily_pipeline",

 target="*", # All assets

 cron_schedule="0 6 * * *", # ทุกวัน 6:00

)



# 3. Definitions

# defs = Definitions(

# assets=[raw_orders, raw_customers, enriched_orders, daily_sales_summary],

# schedules=[daily_schedule],

# )



print("Dagster Assets:")

print(" raw_orders -> enriched_orders -> daily_sales_summary")

print(" raw_customers -> enriched_orders")

print(" Schedule: Daily 6:00 AM")

Team Productivity Patterns

# team_productivity.py — Dagster Team Productivity Patterns

from dataclasses import dataclass, field

from typing import List, Dict

from datetime import datetime



@dataclass

class TeamMember:

 name: str

 role: str

 code_location: str

 assets_owned: List[str]



@dataclass

class ProductivityMetric:

 metric: str

 before_dagster: str

 after_dagster: str

 improvement: str



class TeamProductivityDashboard:

 """Team Productivity Dashboard"""



 def __init__(self):

 self.team: List[TeamMember] = []

 self.metrics: List[ProductivityMetric] = []



 def add_member(self, member: TeamMember):

 self.team.append(member)



 def add_metric(self, metric: ProductivityMetric):

 self.metrics.append(metric)



 def team_overview(self):

 """Team Overview"""

 print(f"\n{'='*55}")

 print(f"Data Team Overview")

 print(f"{'='*55}")



 locations = {}

 for member in self.team:

 loc = member.code_location

 if loc not in locations:

 locations[loc] = []

 locations[loc].append(member)



 for loc, members in locations.items():

 print(f"\n [{loc}]")

 for m in members:

 print(f" {m.name} ({m.role})")

 print(f" Assets: {', '.join(m.assets_owned)}")



 def productivity_report(self):

 """Productivity Improvement Report"""

 print(f"\n Productivity Improvements:")

 for m in self.metrics:

 print(f" {m.metric}:")

 print(f" Before: {m.before_dagster}")

 print(f" After: {m.after_dagster}")

 print(f" Improvement: {m.improvement}")



 def best_practices(self):

 """Best Practices สำหรับทีม"""

 practices = [

 ("Asset Naming Convention",

 "ใช้ prefix ตาม layer: raw_, clean_, agg_, ml_"),

 ("Code Locations",

 "แบ่ง Code Location ตามทีม/Domain"),

 ("Asset Factories",

 "สร้าง Factory Functions สำหรับ Assets ที่คล้ายกัน"),

 ("Testing",

 "เขียน Unit Test สำหรับทุก Asset ด้วย materialize_to_memory"),

 ("Documentation",

 "ใส่ description และ metadata ทุก Asset"),

 ("Alerting",

 "ตั้ง Slack/PagerDuty alerts เมื่อ Asset fail"),

 ("Partitioning",

 "ใช้ Daily/Monthly Partitions สำหรับ Time-series Data"),

 ("CI/CD",

 "ใช้ dagster-cloud CLI deploy อัตโนมัติ"),

 ]



 print(f"\n Best Practices:")

 for name, desc in practices:

 print(f" {name}: {desc}")



# ตัวอย่าง

dashboard = TeamProductivityDashboard()



team = [

 TeamMember("Alice", "Data Engineer", "ingestion",

 ["raw_orders", "raw_customers", "raw_products"]),

 TeamMember("Bob", "Data Engineer", "transformation",

 ["enriched_orders", "clean_customers"]),

 TeamMember("Charlie", "Analytics Engineer", "analytics",

 ["daily_sales", "monthly_report", "customer_segments"]),

 TeamMember("Diana", "ML Engineer", "ml_pipeline",

 ["feature_store", "model_training", "predictions"]),

]



metrics = [

 ProductivityMetric("Pipeline Development Time",

 "2-3 weeks", "3-5 days", "70% faster"),

 ProductivityMetric("Debugging Time",

 "4-8 hours", "30 min-1 hour", "80% reduction"),

 ProductivityMetric("Data Quality Issues",

 "5-10 per week", "1-2 per week", "75% reduction"),

 ProductivityMetric("Onboarding New Member",

 "2-3 weeks", "3-5 days", "70% faster"),

 ProductivityMetric("Deployment Frequency",

 "Weekly", "Daily (CI/CD)", "5x more frequent"),

]



for m in team:

 dashboard.add_member(m)

for m in metrics:

 dashboard.add_metric(m)



dashboard.team_overview()

dashboard.productivity_report()

dashboard.best_practices()

Testing และ CI/CD

# === Dagster Testing และ CI/CD ===



# 1. Unit Testing Assets

# tests/test_assets.py

# from dagster import materialize_to_memory

# from my_project.assets import raw_orders, enriched_orders

#

# def test_raw_orders():

# result = materialize_to_memory([raw_orders])

# assert result.success

# output = result.output_for_node("raw_orders")

# assert output["count"] > 0

# assert "orders" in output

#

# def test_enriched_orders():

# result = materialize_to_memory(

# [raw_orders, raw_customers, enriched_orders]

# )

# assert result.success

# output = result.output_for_node("enriched_orders")

# assert output["count"] > 0

# for order in output["orders"]:

# assert "customer_name" in order

# assert "customer_tier" in order



# 2. GitHub Actions CI/CD

# .github/workflows/dagster.yml

# name: Dagster CI/CD

# on:

# push:

# branches: [main]

# pull_request:

# branches: [main]

#

# jobs:

# test:

# runs-on: ubuntu-latest

# steps:

# - uses: actions/checkout@v4

# - uses: actions/setup-python@v5

# with:

# python-version: '3.11'

# - run: pip install -e ".[dev]"

# - run: pytest tests/ -v

# - run: dagster asset list # Verify assets

#

# deploy:

# needs: test

# if: github.ref == 'refs/heads/main'

# runs-on: ubuntu-latest

# steps:

# - uses: actions/checkout@v4

# - run: pip install dagster-cloud

# - run: dagster-cloud workspace sync

# env:

# DAGSTER_CLOUD_API_TOKEN: }



# 3. Asset Factory Pattern

# def create_table_asset(table_name, schema, source_query):

# @asset(

# name=f"raw_{table_name}",

# group_name="raw",

# metadata={"table": table_name, "schema": schema},

# )

# def _asset(context):

# # db.execute(source_query)

# context.log.info(f"Loaded {table_name}")

# return {"table": table_name, "rows": 1000}

# return _asset

#

# # สร้าง Assets จาก Factory

# tables = ["orders", "customers", "products", "payments"]

# raw_assets = [create_table_asset(t, "public", f"SELECT * FROM {t}") for t in tables]



testing_ci = {

 "Unit Tests": "materialize_to_memory() ทดสอบ Asset ใน Memory",

 "Integration Tests": "ทดสอบกับ Database จริงใน CI",

 "Asset Validation": "dagster asset list ตรวจสอบ Assets",

 "CI Pipeline": "GitHub Actions: test -> deploy",

 "CD": "dagster-cloud workspace sync",

}



print("Testing & CI/CD:")

for key, value in testing_ci.items():

 print(f" {key}: {value}")

Best Practices

• Asset-first: คิดเป็น Assets (Data Products) ไม่ใช่ Tasks
• Code Locations: แบ่ง Code Location ตามทีมหรือ Domain
• Asset Factories: สร้าง Factory Functions สำหรับ Assets ที่คล้ายกัน
• Type System: ใช้ Dagster Types ตรวจสอบ Data Quality
• Partitioning: ใช้ Partitions สำหรับ Time-series Data ลด Reprocessing
• Testing: เขียน Unit Test ทุก Asset ด้วย materialize_to_memory

Dagster คืออะไร

Open-source Data Orchestrator เน้น Software-defined Assets Python Data Pipelines Type System Data Quality UI Asset Lineage Partitioning Scheduling Sensors Modern Data Stack

สรุป

Dagster เป็น Data Orchestrator เน้น Software-defined Assets ทีมเห็น Data Lineage ชัดเจน แบ่ง Code Locations ตามทีม Asset Factories สร้าง Assets ซ้ำได้ Type System ตรวจ Quality Unit Test ด้วย materialize_to_memory CI/CD ด้วย dagster-cloud

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Java GraalVM Service Mesh Setup — คู่มือฉบับสมบูรณ์ 2026