파이프라인 생성 및 배포¶

코드 준비¶

파이프라인을 운영하기 전에 프로덕션용 코드를 준비합니다. 노트북으로 시작했다면 먼저 코드를 각 주요 단계(데이터 준비, 기능 엔지니어링, 모델 학습, 평가)가 명확한 입력과 출력을 가진 별도의 함수가 되는 재사용 가능한 모듈식 함수로 재구성하세요. 이미 모듈식 스크립트가 있는 경우 각 함수에 잘 정의된 인터페이스와 책임이 지정되어 있는지 확인하세요. 테이블 이름 및 하이퍼 매개 변수와 같은 모든 구성 값을 매개 변수화하여 환경 간에 배포할 수 있도록 합니다. 또한 디버깅 및 향후 개발을 위해 엔드투엔드 파이프라인을 로컬에서 실행하는 진입점 스크립트를 작성하는 것이 좋습니다.

디렉터리 구조의 예:

ml_pipeline_project/
├── README.md
├── requirements.txt
├── config/
├── src/ml_pipeline/
│   ├── utils/                     # Common utilities
│   ├── data/                      # Data preparation
│   ├── features/                  # Feature engineering
│   ├── models/                    # Model training
│   └── inference/                 # Model inference
├── scripts/
│   ├── run_pipeline.py            # Main entry point
│   └── dag.py
├── tests/
└── notebooks/

run_pipeline.py 스크립트의 예:

import argparse
from ml_pipeline.utils.config_loader import load_config
from ml_pipeline.data.ingestion import load_raw_data
from ml_pipeline.data.validation import validate_data_quality
from ml_pipeline.features.transformers import create_features
from ml_pipeline.models.training import train_model
from ml_pipeline.models.evaluation import evaluate_model
from ml_pipeline.models.registry import register_model

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True, help="Config file path")
    parser.add_argument("--env", default="dev", help="Environment (dev/prod)")
    args = parser.parse_args()

    # Load configuration
    config = load_config(args.config, args.env)

    # Execute pipeline stages
    raw_data = load_raw_data(config.data.source_table)
    validate_data_quality(raw_data, config.data.quality_checks)
    features = create_features(raw_data, config.features.transformations)
    model = train_model(features, config.model.hyperparameters)
    metrics = evaluate_model(model, features, config.model.eval_metrics)
    register_model(model, metrics, config.model.registry_name)

if __name__ == "__main__":
    main()

Notebooks에서 ML Job으로 마이그레이션¶

Snowflake Notebooks로 작성된 대부분의 코드는 코드 변경 없이도 ML Job에서 잘 작동합니다. 알아야 할 몇 가지 측면은 다음과 같습니다.

런타임 APIs

특정 분산 ML APIs는 Container Runtime 내에서만 사용할 수 있으며, Container Runtime 환경 외부로 가져오려고 하면 실패합니다. 이러한 APIs는 ML Job 내부에서 사용할 수 있지만 ML Job 페이로드 내부로 가져와야 합니다.

# Attempting to import distributed runtime APIs in local/external
# environments will fail!
from snowflake.ml.modeling.distributors.xgboost import XGBEstimator

from snowflake.ml.jobs import remote

@remote(...)
def my_remote_function(...):
  # Move imports *inside* your ML Job payloads
  from snowflake.ml.modeling.distributors.xgboost import XGBEstimator  # This works!
  ...

job = my_remote_function()  # Start ML Job
job.wait()  # Wait for job to complete

클러스터 확장

scale_cluster() API는 Notebook 내에서만 작동하며 ML Job 내에서는 작동하지 않습니다. 대신, 작업 제출 시 원하는 클러스터 크기를 지정하세요. 자세한 내용은 Snowflake Multi-Node ML Jobs 섹션을 참조하십시오.

from snowflake.ml.jobs import remote

@remote(..., target_instances=4)
def my_remote_function(...):
  # 4-node cluster will be provisioned for distributed processing
  # inside this job. The cluster will be automatically cleaned up on
  # job termination.

파이프라인 오케스트레이션¶

엔드투엔드 파이프라인을 준비한 후 Snowflake 작업 그래프, 예약된 Notebooks와 같은 오케스트레이터 또는 Airflow와 같은 외부 오케스트레이터를 사용하여 파이프라인을 운영하세요. 오케스트레이션 프레임워크를 사용하면 다음과 같은 몇 가지 주요 이점이 있습니다.

• 자동 재시도 및 장애 격리를 통한 내결함성 및 안정성

• 실행 기록, 실시간 상태, 경고가 포함된 가시성

• 복잡한 종속성 그래프 및 다양한 트리거에 대한 예약 및 조정

• 버전 제어 통합 및 구성 관리를 통한 운영 위생

Snowflake ML은 Airflow, Dagster, Prefect를 포함한 대부분의 오케스트레이션 프레임워크와 호환됩니다. 기존 워크플로 및 DAG 설정이 이미 있는 경우 기존 워크플로를 Snowflake ML 기능와 통합하고 컴퓨팅 또는 데이터 집약적인 단계를 ML Job 또는 UDFs로 오프로딩하는 것이 좋습니다. 기존 DAG 설정이 없는 경우 Snowflake 네이티브 솔루션을 위해 Snowflake 작업 그래프를 사용할 수 있습니다.

Snowflake에서 DAG로 오케스트레이션을 설정하려면 다음과 같은 개략적인 단계를 따르세요.

1. :ref:`label-create-pipelines-prepare-code`에 따라 로컬 파이프라인 코드 준비

2. DAG 정의를 포함할 새 dag.py 파일(또는 다른 이름) 만들기

3. 이 가이드에 따라 DAG 형식의 파이프라인 구현

4. dag.py 스크립트를 실행하여 작업 그래프를 Snowflake 계정에 배포

개발 및 프로덕션 분리¶

DAG 스크립트를 매개 변수화하여 개발(DEV) 및 프로덕션(PROD) 환경 격리를 지원하는 것이 좋습니다. Snowflake 연결 관리, 애플리케이션별 구성 또는 이 두 가지의 조합을 사용하여 이러한 지원을 달성할 수 있습니다. 필요한 격리 수준은 거버넌스 요구 사항에 따라 다르지만, 일반적으로 DEV 및 PROD에는 별도의 데이터베이스를 사용하는 것이 좋습니다. 여기서 PROD 데이터베이스는 관리자 및 특수 서비스 계정으로 액세스를 제한하는 RBAC 정책에 의해 보호됩니다.

CI 및 CD¶

Azure Pipelines 및 GitHub Actions와 같은 CI 및 CD 파이프라인을 사용하여 파이프라인의 유효성 검사와 배포를 자동화할 수 있습니다. 일반적으로 PROD에 배포하기 전에 DEV 또는 STAGING 환경에서 테스트하는 것이 좋습니다. 모범 사례는 프로덕션 분기에 병합하기 전에 DEV에서 코드 변경 사항의 유효성을 검사하는 병합 게이트로 소스 제어 리포지토리를 구성하는 것입니다. 프로덕션 분기에 대한 변경 사항은 지속적으로(즉, 모든 변경 사항에 대해) 또는 정기적으로(매일 또는 매주) PROD에 배포할 수 있습니다. 모범 사례는 PROD에 변경 사항을 배포하기 전에 DEV 또는 STAGING 환경에서 프로덕션 분기 상태에 대한 최종 유효성 검사를 실행하는 것입니다. GitHub Actions Deployments 및 Environments와 같은 플랫폼 기능을 사용하여 각 배포 환경에 대한 연결을 정의하고 구성하세요. 변경 사항을 배포 환경으로 푸시하도록 다음과 같이 CI 및 CD 파이프라인을 구성합니다.

• (선택 사항) 라이브러리 및 모듈을 Python 패키지로 빌드하고 독점 패키지 피드로 푸시

• (선택 사항) Snowflake 스테이지에 파일 업로드

  파이프라인에서 :code:`snowflake.ml.jobs.submit_from_stage()`를 사용할 때 가장 일반적으로 필요합니다.

  또는 Snowflake의 GitHub 통합을 사용하여 GitHub 리포지토리를 Snowflake 스테이지로 직접 추적할 수 있습니다.

• :file:`dag.py`를 실행하여 구성된 환경에 작업 그래프 배포

• (선택 사항) 새로 배포된 작업 그래프의 실행을 트리거하고 모니터링하여 유효성 확인