AI 엔지니어를 위한 클라우드 컴퓨팅 가이드: AWS/GCP/Azure로 AI 서비스 구축

import sagemaker
from sagemaker.pytorch import PyTorch
from sagemaker import get_execution_role

role = get_execution_role()
sess = sagemaker.Session()

# SageMaker Training Job
estimator = PyTorch(
    entry_point='train.py',
    source_dir='./src',
    role=role,
    instance_type='ml.p4d.24xlarge',
    instance_count=4,
    framework_version='2.1.0',
    py_version='py310',
    hyperparameters={
        'epochs': 10,
        'batch-size': 32,
        'learning-rate': 0.001
    },
    distribution={
        'torch_distributed': {'enabled': True}
    }
)

estimator.fit({'train': 's3://bucket/train', 'val': 's3://bucket/val'})

from sagemaker.pytorch import PyTorchModel

model = PyTorchModel(
    model_data='s3://bucket/model.tar.gz',
    role=role,
    framework_version='2.1.0',
    py_version='py310',
    entry_point='inference.py'
)

predictor = model.deploy(
    initial_instance_count=2,
    instance_type='ml.g4dn.xlarge',
    endpoint_name='my-pytorch-endpoint'
)

# 예측 호출
result = predictor.predict({'inputs': 'Hello, cloud AI!'})

import boto3
import json

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

response = bedrock.invoke_model(
    modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
    body=json.dumps({
        'anthropic_version': 'bedrock-2023-05-31',
        'max_tokens': 1024,
        'messages': [{'role': 'user', 'content': 'AI 트렌드를 설명해줘'}]
    })
)
result = json.loads(response['body'].read())
print(result['content'][0]['text'])

import json
import boto3
import numpy as np

def lambda_handler(event, context):
    # 이벤트에서 입력 데이터 파싱
    body = json.loads(event['body'])
    input_data = body['input']

    # S3에서 모델 로드 (컨테이너 이미지에 번들로 포함 권장)
    prediction = run_inference(input_data)

    return {
        'statusCode': 200,
        'body': json.dumps({'prediction': prediction})
    }

from google.cloud import aiplatform

aiplatform.init(project='my-project', location='us-central1')

job = aiplatform.CustomTrainingJob(
    display_name='pytorch-training',
    script_path='train.py',
    container_uri='us-docker.pkg.dev/vertex-ai/training/pytorch-gpu.2-0:latest',
    requirements=['transformers', 'datasets']
)

model = job.run(
    dataset=None,
    machine_type='a2-highgpu-8g',
    accelerator_type='NVIDIA_TESLA_A100',
    accelerator_count=8,
    args=['--epochs=10', '--batch_size=32']
)

from google.cloud import aiplatform

# 모델 업로드
model = aiplatform.Model.upload(
    display_name='my-pytorch-model',
    artifact_uri='gs://bucket/model/',
    serving_container_image_uri='us-docker.pkg.dev/vertex-ai/prediction/pytorch-gpu.2-0:latest'
)

# 엔드포인트 생성 및 배포
endpoint = aiplatform.Endpoint.create(display_name='my-endpoint')
model.deploy(
    endpoint=endpoint,
    dedicated_resources_machine_type='n1-standard-4',
    dedicated_resources_accelerator_type='NVIDIA_TESLA_T4',
    dedicated_resources_accelerator_count=1,
    min_replica_count=1,
    max_replica_count=5
)

-- BigQuery ML로 분류 모델 학습
CREATE OR REPLACE MODEL `dataset.fraud_model`
OPTIONS(
    model_type = 'BOOSTED_TREE_CLASSIFIER',
    num_parallel_tree = 1,
    max_iterations = 50,
    input_label_cols = ['is_fraud']
) AS
SELECT * FROM `dataset.transactions_train`;

-- 모델 평가
SELECT *
FROM ML.EVALUATE(MODEL `dataset.fraud_model`,
  (SELECT * FROM `dataset.transactions_test`)
);

-- 예측 실행
SELECT *
FROM ML.PREDICT(MODEL `dataset.fraud_model`,
  (SELECT * FROM `dataset.new_transactions`)
);

from azure.ai.ml import MLClient
from azure.ai.ml.entities import AmlCompute, Command
from azure.identity import DefaultAzureCredential

ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="YOUR_SUBSCRIPTION",
    resource_group_name="rg-ai",
    workspace_name="ai-workspace"
)

# GPU 클러스터 생성
compute_config = AmlCompute(
    name="gpu-cluster",
    type="amlcompute",
    size="Standard_ND96asr_v4",
    min_instances=0,
    max_instances=4,
    idle_time_before_scale_down=120
)
ml_client.compute.begin_create_or_update(compute_config).result()

from azure.ai.ml.entities import Command
from azure.ai.ml import Input

job = Command(
    code="./src",
    command="python train.py --epochs 10 --learning_rate 0.001",
    environment="AzureML-pytorch-2.0-ubuntu20.04-py38-cuda11-gpu@latest",
    compute="gpu-cluster",
    inputs={
        "train_data": Input(type="uri_folder", path="azureml://datastores/mydata/paths/train/")
    },
    display_name="pytorch-training-job"
)

returned_job = ml_client.jobs.create_or_update(job)
print(f"Job URL: {returned_job.studio_url}")

from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_key="YOUR_API_KEY",
    api_version="2024-02-01"
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are an AI assistant."},
        {"role": "user", "content": "클라우드 AI의 장점을 설명해줘"}
    ]
)
print(response.choices[0].message.content)

# GKE GPU 노드 풀 (Terraform)
resource "google_container_node_pool" "gpu_pool" {
name       = "gpu-pool"
cluster    = google_container_cluster.primary.name
node_count = 2

node_config {
machine_type = "a2-highgpu-1g"
guest_accelerator {
type  = "nvidia-tesla-a100"
count = 1
}
oauth_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
}
}

kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.14.1/nvidia-device-plugin.yml

import kfp
from kfp import dsl

@dsl.component(
    base_image='python:3.10',
    packages_to_install=['scikit-learn', 'pandas']
)
def train_model(
    data_path: str,
    model_path: kfp.dsl.OutputPath(str)
):
    import pickle
    from sklearn.ensemble import RandomForestClassifier
    import pandas as pd

    df = pd.read_csv(data_path)
    X = df.drop('label', axis=1)
    y = df['label']

    model = RandomForestClassifier(n_estimators=100)
    model.fit(X, y)

    with open(model_path, 'wb') as f:
        pickle.dump(model, f)

@dsl.pipeline(name='ml-pipeline')
def ml_pipeline(data_path: str = 'gs://bucket/data.csv'):
    train_task = train_model(data_path=data_path)

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: inference-scaler
spec:
  scaleTargetRef:
    name: inference-deployment
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: aws-sqs-queue
      metadata:
        queueURL: https://sqs.us-east-1.amazonaws.com/123456789/inference-queue
        queueLength: '5'

# Lambda 콜드 스타트 최적화 패턴
import boto3
import json
from transformers import pipeline

# 전역 변수로 모델 로드 (핸들러 외부)
# 컨테이너가 재사용되면 이 코드는 다시 실행되지 않음
classifier = pipeline('sentiment-analysis', model='distilbert-base-uncased-finetuned-sst-2-english')

def lambda_handler(event, context):
    body = json.loads(event['body'])
    text = body['text']

    result = classifier(text)

    return {
        'statusCode': 200,
        'body': json.dumps(result)
    }

# ECS Task Definition (Fargate + GPU)
{
  'family': 'ai-inference-task',
  'networkMode': 'awsvpc',
  'requiresCompatibilities': ['FARGATE'],
  'cpu': '4096',
  'memory': '16384',
  'containerDefinitions':
    [
      {
        'name': 'inference-container',
        'image': '123456789.dkr.ecr.us-east-1.amazonaws.com/my-model:latest',
        'portMappings': [{ 'containerPort': 8080 }],
        'environment': [{ 'name': 'MODEL_PATH', 'value': '/opt/ml/model' }],
      },
    ],
}

Raw Layer (Bronze)
    └── 원본 데이터 그대로 보존
    └── 파티션: year/month/day/

Processed Layer (Silver)
    └── 정제, 중복 제거, 스키마 강제
    └── Parquet/Delta Lake 포맷

Feature Layer (Gold)
    └── ML 피처 엔지니어링 완료
    └── Feature Store에 등록

import boto3
from pathlib import Path

def save_checkpoint_to_s3(model, optimizer, epoch, loss, bucket, prefix):
    """모델 체크포인트를 S3에 저장"""
    import torch
    checkpoint = {
        'epoch': epoch,
        'model_state_dict': model.state_dict(),
        'optimizer_state_dict': optimizer.state_dict(),
        'loss': loss,
    }
    local_path = f'/tmp/checkpoint_epoch_{epoch}.pt'
    torch.save(checkpoint, local_path)

    s3 = boto3.client('s3')
    s3_key = f'{prefix}/checkpoint_epoch_{epoch}.pt'
    s3.upload_file(local_path, bucket, s3_key)
    print(f'Checkpoint saved to s3://{bucket}/{s3_key}')

from sagemaker.model_monitor import DataCaptureConfig, DefaultModelMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat

# 데이터 캡처 설정
data_capture_config = DataCaptureConfig(
    enable_capture=True,
    sampling_percentage=20,
    destination_s3_uri='s3://bucket/capture'
)

# 모델 모니터 생성
monitor = DefaultModelMonitor(
    role=role,
    instance_count=1,
    instance_type='ml.m5.xlarge',
    volume_size_in_gb=20,
    max_runtime_in_seconds=3600
)

# 베이스라인 생성
monitor.suggest_baseline(
    baseline_dataset='s3://bucket/train_data.csv',
    dataset_format=DatasetFormat.csv(header=True)
)

import boto3

cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')

# 커스텀 메트릭 전송
cloudwatch.put_metric_data(
    Namespace='MLOps/ModelPerformance',
    MetricData=[
        {
            'MetricName': 'PredictionAccuracy',
            'Value': 0.94,
            'Unit': 'None',
            'Dimensions': [
                {'Name': 'ModelName', 'Value': 'fraud-detector-v2'},
                {'Name': 'Environment', 'Value': 'production'}
            ]
        }
    ]
)

name: ML Pipeline CI/CD

on:
  push:
    branches: [main]

jobs:
  train-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1

      - name: Run SageMaker training
        run: |
          python scripts/run_training.py \
            --instance-type ml.p3.2xlarge \
            --output-path s3://bucket/models/

      - name: Deploy to staging
        run: |
          python scripts/deploy_model.py \
            --endpoint-name staging-endpoint \
            --instance-type ml.g4dn.xlarge

import mlflow
import mlflow.pytorch

mlflow.set_tracking_uri('s3://bucket/mlflow')
mlflow.set_experiment('fraud-detection')

with mlflow.start_run():
    # 파라미터 기록
    mlflow.log_params({
        'learning_rate': 0.001,
        'batch_size': 32,
        'epochs': 10
    })

    # 학습 루프
    for epoch in range(10):
        train_loss = train_one_epoch(model, train_loader, optimizer)
        val_accuracy = evaluate(model, val_loader)

        mlflow.log_metrics({
            'train_loss': train_loss,
            'val_accuracy': val_accuracy
        }, step=epoch)

    # 모델 저장
    mlflow.pytorch.log_model(model, 'model')
    mlflow.register_model(
        model_uri=f'runs:/{mlflow.active_run().info.run_id}/model',
        name='fraud-detector'
    )

# SageMaker 카나리 배포
import boto3

sm = boto3.client('sagemaker')

# 트래픽의 10%를 새 모델로 전환
sm.update_endpoint_weights_and_capacities(
    EndpointName='production-endpoint',
    DesiredWeightsAndCapacities=[
        {
            'VariantName': 'current-model',
            'DesiredWeight': 90,
            'DesiredInstanceCount': 4
        },
        {
            'VariantName': 'new-model',
            'DesiredWeight': 10,
            'DesiredInstanceCount': 1
        }
    ]
)