Prefect 2026: Công Cụ Điều Phối Workflow Hiện Đại cho Pipeline Dữ Liệu & AI — Hướng Dẫn Tự Host

from prefect import flow, task

@task(retries=3, retry_delay_seconds=5)
def fetch_data(url: str) -> dict:
    """Fetch data from an API with automatic retry."""
    import requests
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

@flow(name="data-ingestion-pipeline")
def main_flow():
    """Main pipeline orchestrating multiple tasks."""
    raw_data = fetch_data("https://api.example.com/data")
    # ... more tasks

python -m venv prefect-env
source prefect-env/bin/activate  # Linux/Mac
# prefect-env\Scripts\activate  # Windows

# Cài đặt Prefect
pip install prefect>=3.3.0

# Xác minh cài đặt
prefect version
# Expected output: 3.3.0+

# Tùy chọn A: Khởi động nhanh với SQLite (một máy)
prefect server start

# Server khởi động tại http://localhost:4200
# Mở dashboard trong trình duyệt

# Tùy chọn B: Docker Compose với PostgreSQL
cat > docker-compose.yml << 'EOF'
services:
  prefect-server:
    image: prefecthq/prefect:3.3.0-python3.12
    ports:
      - "4200:4200"
    environment:
      - PREFECT_API_DATABASE_CONNECTION_URL=postgresql+asyncpg://prefect:prefect@postgres:5432/prefect
      - PREFECT_HOME=/home/prefect
    command: prefect server start --host 0.0.0.0
    depends_on:
      - postgres

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: prefect
      POSTGRES_PASSWORD: prefect
      POSTGRES_DB: prefect
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:
EOF

docker-compose up -d

# Chỉ Prefect CLI đến server của bạn
prefect config set PREFECT_API_URL=http://localhost:4200/api

# Xác minh kết nối
prefect version
# Should show: Server: http://localhost:4200/api

from prefect import flow, task
from prefect.tasks import task_input_hash
from prefect.artifacts import create_table_artifact
import requests
import pandas as pd
from datetime import timedelta

@task(retries=3, retry_delay_seconds=[10, 30, 60], cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def extract_api_data(endpoint: str, api_key: str) -> list[dict]:
    """Extract data from REST API with retry and caching."""
    headers = {"Authorization": f"Bearer {api_key}"}
    response = requests.get(endpoint, headers=headers, timeout=30)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {len(data)} records from {endpoint}")
    return data

@task(retries=2)
def transform_validate(raw_data: list[dict]) -> pd.DataFrame:
    """Transform and validate raw API data."""
    df = pd.DataFrame(raw_data)
    
    # Data quality checks
    assert not df.empty, "Dataset is empty"
    assert df["id"].is_unique, "Duplicate IDs found"
    
    # Type conversions
    df["created_at"] = pd.to_datetime(df["created_at"])
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    
    # Remove invalid rows
    df = df.dropna(subset=["amount"])
    
    print(f"Validated {len(df)} records after cleaning")
    return df

@task
def load_to_database(df: pd.DataFrame, table_name: str) -> int:
    """Load cleaned data to PostgreSQL."""
    from sqlalchemy import create_engine
    
    engine = create_engine("postgresql://user:pass@localhost:5432/analytics")
    rows_inserted = df.to_sql(table_name, engine, if_exists="append", index=False)
    
    print(f"Loaded {rows_inserted} rows into {table_name}")
    return rows_inserted

@task
def generate_summary_report(df: pd.DataFrame) -> None:
    """Create a summary artifact visible in the dashboard."""
    summary = df.groupby("category").agg({
        "amount": ["sum", "mean", "count"]
    }).round(2).to_dict()
    
    create_table_artifact(
        key="processing-summary",
        table=[
            {"category": cat, "total": stats[("amount", "sum")], "avg": stats[("amount", "mean")], "count": stats[("amount", "count")]}
            for cat, stats in summary.items()
        ],
        description="Data processing summary by category"
    )

@flow(name="daily-etl-pipeline", log_prints=True)
def etl_pipeline(endpoint: str = "https://api.example.com/transactions", api_key: str = "demo-key"):
    """End-to-end ETL pipeline with full observability."""
    # Extract
    raw_data = extract_api_data(endpoint, api_key)
    
    # Transform
    cleaned_data = transform_validate(raw_data)
    
    # Load
    rows_loaded = load_to_database(cleaned_data, "transactions")
    
    # Report
    generate_summary_report(cleaned_data)
    
    return {"rows_processed": rows_loaded, "categories": cleaned_data["category"].nunique()}

if __name__ == "__main__":
    result = etl_pipeline()
    print(f"Pipeline completed: {result}")

python etl_pipeline.py

from prefect import flow
from prefect.schedules import IntervalSchedule
from datetime import timedelta

# Deploy với lịch trình lặp lại
etl_pipeline.serve(
    name="daily-etl",
    schedule=IntervalSchedule(interval=timedelta(hours=24)),
    tags=["production", "etl"]
)

# Deploy với lịch trình cron
prefect deployment build etl_pipeline.py:etl_pipeline \
  --name "daily-etl-cron" \
  --cron "0 6 * * *" \
  --apply

from prefect.docker import DockerImage

@flow
def containerized_flow():
    """Run tasks inside Docker containers."""
    pass

# Deploy với Docker
containerized_flow.deploy(
    name="docker-etl",
    work_pool_name="docker-pool",
    image=DockerImage(name="my-etl", tag="1.0")
)

# Tạo Docker work pool
prefect work-pool create docker-pool --type docker

# Khởi động worker
prefect worker start --pool docker-pool

# Tạo Kubernetes work pool
prefect work-pool create k8s-pool --type kubernetes

# Deploy với Kubernetes manifest
prefect deployment build etl_pipeline.py:etl_pipeline \
  --name "k8s-etl" \
  --pool k8s-pool \
  --infra kubernetes-job \
  --apply

from prefect import flow
from prefect_dbt.cli.commands import trigger_dbt_cli_command
from prefect_dbt.cli.configs import TargetConfigs

@flow(name="dbt-transform-pipeline")
def run_dbt_models():
    """Run dbt models with Prefect orchestration."""
    # Run dbt deps
    trigger_dbt_cli_command("dbt deps")
    
    # Run models
    trigger_dbt_cli_command("dbt run")
    
    # Run tests
    trigger_dbt_cli_command("dbt test")
    
    # Generate docs
    trigger_dbt_cli_command("dbt docs generate")

# Deploy
run_dbt_models.serve(name="dbt-daily")

pip install prefect-dbt[cli]

from prefect import flow, task
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

@task
def download_from_s3(bucket: str, key: str) -> str:
    """Download file from S3."""
    s3 = S3Bucket.load("my-s3-block")
    return s3.read_path(f"{bucket}/{key}")

@task
def upload_to_s3(local_path: str, bucket: str, key: str) -> None:
    """Upload file to S3."""
    s3 = S3Bucket.load("my-s3-block")
    s3.upload_from_path(local_path, f"{bucket}/{key}")

@flow(name="s3-data-pipeline")
def s3_pipeline():
    """Pipeline moving data through S3."""
    data = download_from_s3("raw-data", "input.csv")
    # ... process ...
    upload_to_s3("processed.csv", "processed-data", "output.csv")

pip install prefect-aws

# Đăng ký AWS credentials block
prefect block register --module prefect_aws.credentials

from prefect import flow
from prefect.blocks.notifications import SlackWebhook

@flow(on_failure=[send_slack_alert], on_crashed=[send_slack_alert])
def monitored_flow():
    """Flow with automatic Slack alerting on failure."""
    pass

def send_slack_alert(flow, flow_run, state):
    """Send alert to Slack when flow fails."""
    slack = SlackWebhook.load("alerts-webhook")
    slack.notify(
        body=f"Flow {flow.name} failed with state {state.name}. "
             f"Check: http://localhost:4200/flow-runs/{flow_run.id}"
    )

from prefect.events import emit_event
from prefect import flow

@flow
def on_file_uploaded(file_path: str):
    """Process file when S3 upload event fires."""
    result = process_file(file_path)
    
    # Phát event tùy chỉnh cho downstream flows
    emit_event(
        event="file.processed",
        resource={"prefect.resource.id": file_path},
        payload={"rows": len(result)}
    )

# Định nghĩa automation kích hoạt trên event này
# Cấu hình trong Prefect dashboard hoặc qua API

import asyncio
from prefect import flow, task

@task
async def fetch_async(url: str) -> dict:
    """Async HTTP fetch."""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

@flow
async def concurrent_fetch_flow(urls: list[str]):
    """Fetch all URLs concurrently."""
    tasks = [fetch_async.submit(url) for url in urls]
    results = [t.result() for t in tasks]
    return results

# Chạy 100 lệnh gọi API đồng thờ
urls = [f"https://api.example.com/item/{i}" for i in range(100)]
results = asyncio.run(concurrent_fetch_flow(urls))

from prefect import task
from datetime import timedelta

@task(
    retries=5,
    retry_delay_seconds=[1, 2, 4, 8, 16],  # Exponential backoff
    retry_jitter=True  # Thêm tính ngẫu nhiên để ngăn thundering herd
)
def call_external_api(endpoint: str) -> dict:
    """Call external API with smart retry logic."""
    import requests
    response = requests.get(endpoint, timeout=10)
    response.raise_for_status()
    return response.json()

from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def process_with_resource_limit(item_id: str) -> dict:
    """Process item with controlled concurrency."""
    with concurrency("database-slots", occupy=1):
        # Chỉ N task có thể thực thi block này đồng thờ
        return query_database(item_id)

@flow
def limited_processing_flow(item_ids: list[str]):
    """Process items with max 10 concurrent database queries."""
    from prefect.tasks import map
    results = map(process_with_resource_limit, item_ids)
    return results

# Tạo concurrency limit qua CLI
prefect concurrency-limit create database-slots 10

from prefect import flow, task
from pydantic import BaseModel, Field
from typing import List

class Transaction(BaseModel):
    """Validated transaction model."""
    id: str
    amount: float = Field(gt=0, description="Must be positive")
    currency: str = Field(pattern="^(USD|EUR|GBP)$")
    created_at: str

class PipelineOutput(BaseModel):
    """Validated pipeline output."""
    total_amount: float
    transaction_count: int
    currency: str

@task
def validate_transactions(raw_data: List[dict]) -> List[Transaction]:
    """Validate and parse raw transaction data."""
    return [Transaction(**item) for item in raw_data]

@flow
def validated_pipeline(raw_data: List[dict]) -> PipelineOutput:
    """Pipeline with full input/output validation."""
    transactions = validate_transactions(raw_data)
    
    return PipelineOutput(
        total_amount=sum(t.amount for t in transactions),
        transaction_count=len(transactions),
        currency=transactions[0].currency if transactions else "USD"
    )

# .github/workflows/prefect-deploy.yml
name: Deploy Prefect Flows
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Install dependencies
        run: |
          pip install prefect>=3.3.0
          pip install -r requirements.txt
      
      - name: Authenticate with Prefect Cloud
        run: |
          prefect config set PREFECT_API_URL=${{ secrets.PREFECT_API_URL }}
          prefect config set PREFECT_API_KEY=${{ secrets.PREFECT_API_KEY }}
      
      - name: Deploy flows
        run: |
          prefect deploy --all --prefect-file prefect.yaml
      
      - name: Run health check
        run: |
          prefect flow-run list --limit 5

# prefect.yaml — Định nghĩa deployments dưới dạng code
name: production-pipelines
prefect-version: 3.3.0

build:
  - prefect_docker.deployments.steps.build_docker_image:
      requires: prefect-docker
      image_name: my-pipeline
      tag: "{{ sha }}"
      dockerfile: Dockerfile

push:
  - prefect_docker.deployments.steps.push_docker_image:
      requires: prefect-docker
      image_name: my-pipeline
      tag: "{{ sha }}"
      credentials: "{{ prefect.blocks.docker-registry-credentials.prod-registry }}"

pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /opt/prefect

deployments:
  - name: daily-etl
    entrypoint: etl_pipeline.py:etl_pipeline
    work_pool:
      name: docker-pool
    schedule:
      cron: "0 6 * * *"
    parameters:
      endpoint: "https://api.production.example.com/v1/data"
    tags: ["production", "etl", "daily"]
    
  - name: hourly-analytics
    entrypoint: analytics_pipeline.py:hourly_flow
    work_pool:
      name: k8s-pool
    schedule:
      interval: 3600
    tags: ["production", "analytics"]

from prefect import flow
from prefect.blocks.webhook import Webhook
from datetime import timedelta

@flow(
    timeout_seconds=3600,
    on_failure=[notify_team],
    on_crashed=[notify_team, escalate_to_pagerduty]
)
def critical_revenue_pipeline():
    """Revenue pipeline with full monitoring."""
    # Pipeline logic here
    pass

def notify_team(flow, flow_run, state):
    """Send notification on failure."""
    webhook = Webhook.load("slack-alerts")
    webhook.notify(
        body=f"CRITICAL: {flow.name} failed after {flow_run.total_run_time}s"
    )

def escalate_to_pagerduty(flow, flow_run, state):
    """Escalate to PagerDuty for crashed flows."""
    webhook = Webhook.load("pagerduty-integration")
    webhook.notify(
        body=json.dumps({
            "routing_key": "YOUR_ROUTING_KEY",
            "event_action": "trigger",
            "payload": {
                "summary": f"Prefect flow {flow.name} crashed",
                "severity": "critical"
            }
        })
    )

from prefect import flow, task
from prefect.tasks import map

@task
def process_file(filename: str) -> dict:
    """Process a single file."""
    # ... processing logic ...
    return {"file": filename, "rows": 1000}

@flow
def dynamic_processing_flow(directory: str):
    """Dynamically process all files in a directory."""
    import os
    files = [f for f in os.listdir(directory) if f.endswith(".csv")]
    results = map(process_file, files)
    return results