ZenML 2026: 20개 이상의 도구를 프로덕션 파이프라인으로 연결하는 MLOps 프레임워크 — 완전 설정 가이드

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

# ZenML 코어 설치
pip install zenml

# 설치 확인
zenml version
# Output: ZenML version 0.80.0

# ZenML 리포지토리 초기화 (.zen 디렉토리 생성)
zenml init

# 상태 확인
zenml status

# 로컬 아티팩트 저장소 등록
zenml artifact-store register local_store --flavor=local --path=./artifacts

# 로컬 오케스트레이터 등록
zenml orchestrator register local_orchestrator --flavor=local

# 두 구성 요소를 결합한 스택 생성
zenml stack register local_stack \
  -o local_orchestrator \
  -a local_store \
  --set

# 활성 스택 확인
zenml stack describe

from zenml import pipeline, step
import pandas as pd
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

@step
def load_data() -> pd.DataFrame:
    """Load the iris dataset."""
    iris = load_iris(as_frame=True)
    df = iris.frame
    return df

@step
def split_data(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
    """Split data into training and test sets."""
    X = df.drop("target", axis=1)
    y = df["target"]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    return X_train, X_test, y_train, y_test

@step
def train_model(X_train: pd.DataFrame, y_train: pd.Series) -> RandomForestClassifier:
    """Train a Random Forest classifier."""
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)
    return clf

@step
def evaluate_model(
    model: RandomForestClassifier,
    X_test: pd.DataFrame,
    y_test: pd.Series
) -> float:
    """Evaluate the trained model."""
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    print(f"Model accuracy: {accuracy:.4f}")
    return accuracy

@pipeline
def training_pipeline():
    """End-to-end ML training pipeline."""
    df = load_data()
    X_train, X_test, y_train, y_test = split_data(df)
    model = train_model(X_train, y_train)
    accuracy = evaluate_model(model, X_test, y_test)

if __name__ == "__main__":
    run = training_pipeline()
    print(f"Pipeline run completed: {run.name}")

python first_pipeline.py

# Airflow 통합 설치
pip install zenml[airflow]

# Airflow 오케스트레이터 등록
zenml orchestrator register airflow_orchestrator \
  --flavor=airflow \
  --local=True

# Airflow 스택으로 전환
zenml stack update local_stack -o airflow_orchestrator

# MLflow 통합 설치
pip install zenml[mlflow]

# MLflow UI 시작 (별도 터미널에서)
mlflow ui --port 5000

# MLflow 실험 트래커 등록
zenml experiment-tracker register mlflow_tracker \
  --flavor=mlflow \
  --tracking_uri=http://localhost:5000

# MLflow 모델 레지스트리 등록
zenml model-registry register mlflow_registry \
  --flavor=mlflow \
  --uri=http://localhost:5000

# 스택 업데이트
zenml stack update local_stack \
  -e mlflow_tracker \
  -r mlflow_registry

from zenml import pipeline, step
from zenml.client import Client
import mlflow
import mlflow.sklearn

@step(experiment_tracker="mlflow_tracker")
def train_model(X_train: pd.DataFrame, y_train: pd.Series) -> RandomForestClassifier:
    """Train with MLflow logging."""
    mlflow.autolog()  # Auto-log parameters, metrics, and model
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)
    
    # Log custom metrics
    mlflow.log_param("n_estimators", 100)
    mlflow.log_metric("train_samples", len(X_train))
    
    return clf

# Register the model after training
@step(model_registry="mlflow_registry")
def register_model(
    model: RandomForestClassifier,
    accuracy: float
) -> str:
    """Register model to MLflow model registry."""
    if accuracy > 0.90:
        model_version = mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            registered_model_name="iris-classifier"
        )
        print(f"Model registered: {model_version}")
        return "iris-classifier"
    return "below-threshold"

# S3 아티팩트 저장소 등록
zenml artifact-store register s3_store \
  --flavor=s3 \
  --path=s3://my-ml-bucket/zenml-artifacts \
  --aws_access_key_id=$AWS_ACCESS_KEY_ID \
  --aws_secret_access_key=$AWS_SECRET_ACCESS_KEY

# S3를 사용하도록 스택 업데이트
zenml stack update local_stack -a s3_store

# Docker 컨테이너 레지스트리 등록
zenml container-registry register docker_registry \
  --flavor=default \
  --uri=myregistry.azurecr.io

# Docker를 사용하도록 스택 업데이트
zenml stack update local_stack -c docker_registry
zenml pipeline run first_pipeline.py --build-docker

pip install zenml[wandb]

zenml experiment-tracker register wandb_tracker \
  --flavor=wandb \
  --api_key=$WANDB_API_KEY \
  --project_name="zenml-mlops"

# stack.yaml — 코드로 전체 MLOps 스택 정의
stack_name: production_stack
components:
  orchestrator:
    flavor: kubernetes
    configuration:
      kubernetes_context: prod-cluster
      namespace: ml-pipelines
  artifact_store:
    flavor: s3
    configuration:
      path: s3://prod-ml-artifacts/zenml
      authentication_secret: aws-s3-secret
  container_registry:
    flavor: default
    configuration:
      uri: 123456789.dkr.ecr.us-east-1.amazonaws.com
  experiment_tracker:
    flavor: mlflow
    configuration:
      tracking_uri: http://mlflow.internal:5000
  model_registry:
    flavor: mlflow
    configuration:
      uri: http://mlflow.internal:5000
  step_operator:
    flavor: sagemaker
    configuration:
      role: arn:aws:iam::123456789:role/SageMakerRole
      instance_type: ml.p3.2xlarge

zenml stack register -f stack.yaml --set

from zenml.step_operators import BaseStepOperator

@step(step_operator="sagemaker_gpu")
def train_deep_learning_model(X_train: pd.DataFrame, y_train: pd.Series):
    """SageMaker를 통해 GPU에서 학습, 다른 단계는 로컬에서 실행."""
    import tensorflow as tf
    
    # 이 단계는 SageMaker에서 ml.p3.2xlarge에서 실행
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(3, activation='softmax')
    ])
    
    model.compile(optimizer='adam', loss='sparse_categorical_crossentropy')
    model.fit(X_train, y_train, epochs=50, batch_size=32)
    
    return model

from zenml.pipelines import Schedule

# 매일 UTC 3시에 파이프라인 실행
daily_schedule = Schedule(
    cron_expression="0 3 * * *",
    pipeline_name="training_pipeline",
    stack_name="production_stack"
)

zenml.pipeline_schedule register daily_schedule

@step(enable_cache=True)  # 기본 동작
def expensive_preprocessing(df: pd.DataFrame) -> pd.DataFrame:
    """입력 df 또는 이 함수가 변경될 때만 재실행."""
    # 30분 걸리는 무거운 변환
    return processed_df

# 필요시 강제 재실행
zenml pipeline run training_pipeline.py --no-cache

# 데이터베이스 자격 증명을 위한 시크릿 등록
zenml secrets-manager register aws_secrets \
  --flavor=aws \
  --region_name=us-east-1

zenml stack update local_stack -x aws_secrets

# 시크릿 생성
zenml secrets-manager secret register db_credentials \
  --schema=username_password \
  --username=ml_user \
  --password=$DB_PASSWORD

from zenml.client import Client

@step
def load_from_database() -> pd.DataFrame:
    """ZenML 시크릿 매니저의 자격 증명을 사용해 데이터 로드."""
    client = Client()
    credentials = client.get_secret("db_credentials")
    
    import psycopg2
    conn = psycopg2.connect(
        host="db.internal",
        user=credentials.username,
        password=credentials.password
    )
    df = pd.read_sql("SELECT * FROM training_data", conn)
    return df

# .github/workflows/ml-pipeline.yml
name: ML Pipeline CI
on:
  push:
    branches: [main]
  schedule:
    - cron: "0 2 * * *"

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup ZenML
        run: |
          pip install zenml[mlflow,aws]
          zenml connect --url $ZENML_SERVER_URL --api-key $ZENML_API_KEY
      
      - name: Run training pipeline
        run: |
          zenml pipeline run training_pipeline.py \
            --stack production_stack \
            --build-docker
      
      - name: Notify on failure
        if: failure()
        run: |
          curl -X POST $SLACK_WEBHOOK \
            -d '{"text":"Pipeline failed! Check ZenML dashboard."}'