Dagster Pipeline Performance Tuning

Dagster Performance Tuning

Dagster Data Pipeline Performance Tuning Asset Optimization Partitioning Concurrency Caching IO Manager Profiling Monitoring Python Orchestrator

Asset Optimization

=== Dagster Asset Performance ===

pip install dagster dagster-webserver dagster-postgres

Before: Slow — Process all data every run

@asset

df = pd.read_sql("SELECT * FROM orders", engine) # ALL data

result = df.groupby('date').agg({'amount': 'sum'})

result.to_sql('daily_sales', engine, if_exists='replace')

return result

After: Fast — Partitioned by date

from dagster import asset, DailyPartitionsDefinition

daily_partitions = DailyPartitionsDefinition(start_date="2024-01-01")

@asset(partitions_def=daily_partitions)

partition_date = context.partition_key

df = pd.read_sql(

"SELECT * FROM orders WHERE date = %s",

engine, params=[partition_date]

) # Only 1 day

result = df.groupby('date').agg({'amount': 'sum'})

return result

Concurrency — Run independent assets in parallel

from dagster import define_asset_job, Definitions

@asset

@asset

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง XDR Platform สำหรับมือใหม่ Step by Step

@asset(deps=[raw_orders, raw_products])

# raw_orders and raw_products run in PARALLEL

# enriched_orders waits for both

from dataclasses import dataclass

@dataclass

class AssetPerformance:

asset: str

before_sec: float

แนะนำเพิ่มเติม — อีบุ๊กการลงทุน SiamCafeBook

after_sec: float

improvement: str

technique: str

assets = [

AssetPerformance("daily_sales", 300, 15, "20x faster", "Partitioning"),

AssetPerformance("user_features", 180, 45, "4x faster", "Caching"),

AssetPerformance("product_catalog", 120, 30, "4x faster", "IO Manager"),

AssetPerformance("enriched_orders", 240, 60, "4x faster", "Concurrency"),

AssetPerformance("ml_predictions", 600, 120, "5x faster", "Partition + Parallel"),

]

IO Manager และ Caching

=== IO Manager & Caching ===

Custom IO Manager — Parquet instead of CSV

from dagster import IOManager, io_manager

import pyarrow.parquet as pq

self.base_path = base_path

เนื้อหาเกี่ยวข้อง — อ่านต่อ: AWS EventBridge Certification Path

path = f"{self.base_path}/{context.asset_key.path[-1]}.parquet"

obj.to_parquet(path, compression='snappy')

context.log.info(f"Wrote {len(obj)} rows to {path}")

path = f"{self.base_path}/{context.asset_key.path[-1]}.parquet"

return pd.read_parquet(path)

@io_manager

return ParquetIOManager("/data/assets")

Freshness Policy — Skip if not stale

@asset(

freshness_policy=FreshnessPolicy(maximum_lag_minutes=60),

auto_materialize_policy=AutoMaterializePolicy.eager(),

)

# Only runs if data is stale (>60 min old)

...

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

Resource Pool — Limit DB connections

from dagster import resource

@resource(config_schema={"pool_size": int})

return create_engine(

DB_URL,

pool_size=context.resource_config["pool_size"],

max_overflow=5,

)

io_comparison = {

"CSV IO Manager": {"write_sec": 45, "read_sec": 30, "size_mb": 500},

"Parquet IO Manager": {"write_sec": 8, "read_sec": 3, "size_mb": 120},

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Kubernetes CRD Cloud Migration Strategy

"Delta Lake IO Manager": {"write_sec": 12, "read_sec": 5, "size_mb": 100},

"Snowflake IO Manager": {"write_sec": 20, "read_sec": 15, "size_mb": 0},

}

for name, perf in io_comparison.items():

Monitoring และ Profiling

=== Monitoring & Profiling ===

Dagster UI — Built-in Monitoring

dagster dev # Start UI at localhost:3000

• Asset Lineage Graph
• Run Timeline
• Asset Materialization History
• Partition Status Matrix

Custom Logging

@asset

import time

start = time.time()

context.log.info("Starting data load...")

df = load_data()

context.log.info(f"Loaded {len(df)} rows in {time.time()-start:.1f}s")

start2 = time.time()

result = transform(df)

context.log.info(f"Transformed in {time.time()-start2:.1f}s")

context.add_output_metadata({

"row_count": len(result),

"duration_sec": time.time() - start,

"memory_mb": result.memory_usage(deep=True).sum() / 1024**2,

})

return result

Prometheus Metrics

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Midjourney Prompt RBAC ABAC Policy —

from prometheus_client import Counter, Histogram

ASSET_DURATION = Histogram('dagster_asset_duration_seconds',

'Asset materialization duration',

['asset_name'])

ASSET_ROWS = Counter('dagster_asset_rows_total',

'Total rows processed', ['asset_name'])

pipeline_metrics = {

"Total Pipeline Duration": "12 min (was 45 min)",

"Longest Asset": "ml_predictions: 2 min",

"Parallelism": "4 assets concurrent",

"Partitions/Day": "365 (daily for 1 year)",

"Cache Hit Rate": "78% (skipped stale checks)",

"Memory Peak": "2.1 GB (was 8.5 GB)",

"Failed Runs (7d)": "2 (both auto-retried)",

}

for k, v in pipeline_metrics.items():

เคล็ดลับ

• Partition: แบ่ง Partition ตามวันหรือ Key ลด Data ที่ Process
• Parquet: ใช้ Parquet แทน CSV เร็วกว่า 5-10x เล็กกว่า 4x
• Freshness: ใช้ FreshnessPolicy ไม่ทำงานซ้ำถ้า Data ไม่เปลี่ยน
• Pool: จำกัด Connection Pool ไม่ให้ DB Overload
• Metadata: Log Duration และ Row Count ทุก Asset

การนำไปใช้งานจริงในองค์กร

สำหรับองค์กรขนาดกลางถึงใหญ่ แนะนำให้ใช้หลัก Three-Tier Architecture คือ Core Layer ที่เป็นแกนกลางของระบบ Distribution Layer ที่ทำหน้าที่กระจาย Traffic และ Access Layer ที่เชื่อมต่อกับผู้ใช้โดยตรง การแบ่ง Layer ชัดเจนช่วยให้การ Troubleshoot ง่ายขึ้นและสามารถ Scale ระบบได้ตามความต้องการ

เรื่อง Network Security ก็สำคัญไม่แพ้กัน ควรติดตั้ง Next-Generation Firewall ที่สามารถ Deep Packet Inspection ได้ ใช้ Network Segmentation แยก VLAN สำหรับแต่ละแผนก ติดตั้ง IDS/IPS เพื่อตรวจจับการโจมตี และทำ Regular Security Audit อย่างน้อยปีละ 2 ครั้ง

สรุป

Dagster Pipeline Performance Tuning Asset Partitioning Concurrency IO Manager Parquet Caching FreshnessPolicy Resource Pool Monitoring Profiling Metadata Python Data Orchestrator