Amazon SageMaker ML on Local Machine via VS Code

$ aws configure
AWS Access Key ID [****************BIGP]: 
AWS Secret Access Key [****************/5l8]: 
Default region name [us-east-1]: 
Default output format [json]: 

$ vi ~/.aws/config

[default]
region = ap-northeast-2
output = json

$ virtualenv venv
$ source venv/bin/activate
(venv) $ pip install -r requirements.txt

import boto3
import sagemaker
import pandas as pd
import sklearn

print(f"boto3 version: {boto3.__version__}")
print(f"sagemaker version: {sagemaker.__version__}")
print(f"pandas version: {pd.__version__}")
print(f"sklearn version: {sklearn.__version__}")

[결과]
boto3 version: 1.37.9
sagemaker version: 2.241.0
pandas version: 2.2.3
sklearn version: 1.6.1

# 시스템 관련 라이브러리
import os

# AWS 관련 라이브러리
import boto3
import sagemaker
from sagemaker.sklearn import SKLearn  # SageMaker에서 scikit-learn 모델 학습을 위한 클래스
from sagemaker.local import LocalSession  # 로컬 환경에서 SageMaker 실행을 위한 세션

# 데이터 처리 및 머신러닝 라이브러리
import pandas as pd
from sklearn.datasets import load_iris  # 붓꽃 데이터셋 로드
from sklearn.model_selection import train_test_split  # 학습/테스트 데이터 분리

# SageMaker 로컬 세션 생성
sagemaker_local_session = LocalSession()

# 로컬 코드 실행을 위한 설정
sagemaker_local_session.config = {'local': {'local_code': True}}

# AWS 계정 ID 가져오기
account = boto3.client('sts').get_caller_identity().get('Account')

# SageMaker 서비스 카탈로그 역할 (사용하지 않음)
# role = f"arn:aws:iam::{account}:role/service-role/AmazonSageMakerServiceCatalogProductsUseRole"
# ML/DL 실행을 위한 IAM 역할 설정
role = f'arn:aws:iam::{account}:role/awsMLDLRole'
print(f"role: {role}")

[결과]
role: arn:aws:iam::YOURID:role/service-role/AmazonSageMakerServiceCatalogProductsUseRole

# 현재 프로젝트 디렉토리 가져오기
home_dir = os.getcwd()

# 데이터 디렉토리가 없으면 생성
data_dir = os.path.join(home_dir, 'data')
os.makedirs(data_dir, exist_ok=True)

# 붓꽃 데이터셋 로드
iris = load_iris()
X = pd.DataFrame(iris.data, columns=iris.feature_names)
y = pd.DataFrame(iris.target, columns=['target'])
data = pd.concat([X, y], axis=1)

# 데이터 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 특성과 타겟 변수 결합하여 저장
train_data = pd.concat([X_train, y_train], axis=1)
test_data = pd.concat([X_test, y_test], axis=1)

# CSV 파일로 저장
train_path = os.path.join(data_dir, 'train.csv')
test_path = os.path.join(data_dir, 'test.csv')

train_data.to_csv(train_path, index=False)
test_data.to_csv(test_path, index=False)

print(f"학습 데이터 저장 위치: {train_path}")
print(f"테스트 데이터 저장 위치: {test_path}")

# S3 버킷 생성 및 데이터 업로드
bucket_name = f"sagemaker-local-training-{account}"  # 계정 ID를 포함하여 유니크한 버킷 이름 생성
region = boto3.session.Session().region_name
s3_client = boto3.client('s3')

# 버킷이 존재하지 않으면 생성
try:
    s3_client.head_bucket(Bucket=bucket_name)
    print(f"버킷 {bucket_name}이(가) 이미 존재합니다")
except:
    try:
        if region == 'us-east-1':
            s3_client.create_bucket(Bucket=bucket_name)
        else:
            s3_client.create_bucket(
                Bucket=bucket_name,
                CreateBucketConfiguration={'LocationConstraint': region}
            )
        print(f"새로운 버킷 생성: {bucket_name}")
    except Exception as e:
        print(f"버킷 생성 중 오류 발생: {str(e)}")

# S3에 데이터 업로드
try:
    # 학습 데이터(train.csv) 업로드
    s3_train_path = 'data/train/train.csv'
    s3_client.upload_file(train_path, bucket_name, s3_train_path)
    print(f"학습 데이터를 s3://{bucket_name}/{s3_train_path}에 업로드했습니다")
    
    # 테스트 데이터(test.csv) 업로드
    s3_test_path = 'data/test/test.csv'
    s3_client.upload_file(test_path, bucket_name, s3_test_path)
    print(f"테스트 데이터를 s3://{bucket_name}/{s3_test_path}에 업로드했습니다")
    
    # SageMaker 학습을 위한 입력 경로 업데이트
    training_input_path = f"s3://{bucket_name}/{s3_train_path}"
    print(f"\n학습 입력 경로가 다음으로 업데이트되었습니다: {training_input_path}")
    
except Exception as e:
    print(f"S3에 파일 업로드 중 오류 발생: {str(e)}")

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
import joblib
import os

# 모델 아티팩트 디렉토리가 없으면 생성
model_artifacts_dir = 'model_artifacts'
if not os.path.exists(model_artifacts_dir):
    os.makedirs(model_artifacts_dir)

# 학습 데이터 로드
train_data = pd.read_csv(train_path, skiprows=1)

# 특성과 타겟 분리
X = train_data.iloc[:, :-1]  # 마지막 열을 제외한 모든 열
y = train_data.iloc[:, -1]   # 마지막 열이 타겟

X_cols = X.columns
# 모델 생성 및 학습
model = RandomForestClassifier(n_estimators=20, random_state=42)
model.fit(X, y)

# 모델 저장
model_path = os.path.join(model_artifacts_dir, 'sk_rf_model.joblib')
joblib.dump(model, model_path)
print(f"Model saved to {model_path}")

try:
    # 학습된 모델을 S3에 업로드
    s3_model_path = 'model_artifacts/sk_rf_model.joblib'
    s3_client.upload_file(model_path, bucket_name, s3_model_path)
    print(f"Uploaded model to s3://{bucket_name}/{s3_model_path}")
except Exception as e:
    print(f"Error uploading model to S3: {str(e)}")

[결과]
Uploaded model to s3://sagemaker-local-training-381282212299/model_artifacts/sk_rf_model.joblib

# 테스트 데이터 불러오기
test_path = 'data/test.csv'
test_data = pd.read_csv(test_path, skiprows=1)

# 테스트 데이터의 특성과 타겟 분리
X_test = test_data.iloc[:, :-1]  # 마지막 열을 제외한 모든 열
y_test = test_data.iloc[:, -1]   # 마지막 열이 타겟

# 학습된 모델 불러오기
model_path = 'model_artifacts/sk_rf_model.joblib'
loaded_model = joblib.load(model_path)

# 테스트 데이터의 컬럼명을 학습 데이터와 동일하게 설정
X_test.columns = X_cols # 학습 데이터의 컬럼명 사용

# 예측 수행
predictions = loaded_model.predict(X_test)

# 정확도 계산 및 출력
accuracy = (predictions == y_test).mean()
print(f"테스트 정확도: {accuracy:.4f}")

# 처음 몇 개의 예측값과 실제값 출력
print("\n처음 5개의 예측값과 실제값 비교:")
for pred, actual in zip(predictions[:5], y_test[:5]):
    print(f"예측값: {pred}, 실제값: {actual}")

# S3에서 테스트 데이터 가져오기
try:
    import tempfile
    
    # 버킷 이름을 사용하여 S3 경로 구성
    s3_test_path = f"s3://{bucket_name}/data/test/test.csv"
    
    # S3 데이터를 다운로드할 임시 파일 생성
    with tempfile.NamedTemporaryFile(delete=False, suffix='.csv') as tmp_file:
        s3_client.download_file(bucket_name, 'data/test/test.csv', tmp_file.name)
        
        # 다운로드한 CSV를 판다스 DataFrame으로 읽기
        test_data_s3 = pd.read_csv(tmp_file.name, skiprows=1)
        
    print(f"{s3_test_path}에서 테스트 데이터를 성공적으로 불러왔습니다")
    print(f"테스트 데이터 크기: {test_data_s3.shape}")
    
except Exception as e:
    print(f"S3에서 테스트 데이터를 불러오는 중 오류 발생: {str(e)}")

# 모델 가져오기
try:
    # S3에서 모델 다운로드
    local_model_path = os.path.join('model_artifacts', 'sk_rf_model.joblib')
    s3_model_path = 'model_artifacts/sk_rf_model.joblib'
    s3_client.download_file(bucket_name, s3_model_path, local_model_path)
    print(f"S3에서 모델을 성공적으로 다운로드했습니다")
    
    # 모델 불러오기
    model = joblib.load(local_model_path)
    
    # 테스트 데이터 준비 (타겟 컬럼 제외)
    X_test_s3 = test_data_s3.iloc[:, :-1]
    y_test_s3 = test_data_s3.iloc[:, -1]
    
    # 테스트 데이터의 컬럼명을 학습 데이터와 동일하게 설정
    X_test_s3.columns = X_cols
  
    # 예측 수행
    predictions = model.predict(X_test_s3)
    
    # 정확도 계산 및 출력
    accuracy = (predictions == y_test_s3).mean()
    print(f"\n테스트 정확도: {accuracy:.4f}")
    
    # 처음 몇 개의 예측값과 실제값 출력
    print("\n처음 5개의 예측값과 실제값 비교:")
    for pred, actual in zip(predictions[:5], y_test_s3[:5]):
        print(f"예측값: {pred}, 실제값: {actual}")
        
except Exception as e:
    print(f"S3 모델 테스트 중 오류 발생: {str(e)}")

import argparse  # 명령줄 인자(argument) 처리를 위한 라이브러리
import pandas as pd  # 데이터 처리를 위한 라이브러리
from sklearn.ensemble import RandomForestClassifier  # 랜덤 포레스트 분류기
import joblib  # 모델 저장 및 불러오기를 위한 라이브러리
import os  # 파일 경로 및 환경 변수 처리를 위한 라이브러리

if __name__ == "__main__":
    # 명령줄 인자 파서 설정
    parser = argparse.ArgumentParser()

    # 하이퍼파라미터 (트리 개수) 인자를 받음, 기본값 -1 (예외 처리용)
    parser.add_argument('--n_estimators', type=int, default=-1)

    # 학습 데이터 경로 (SageMaker 환경변수 SM_CHANNEL_TRAIN을 통해 자동 지정됨)
    parser.add_argument("--train", type=str, default=os.environ.get("SM_CHANNEL_TRAIN"))

    # 모델이 저장될 경로 (SageMaker 환경변수 SM_MODEL_DIR을 통해 자동 지정됨)
    parser.add_argument("--model_dir", type=str, default=os.environ.get("SM_MODEL_DIR"))

    # 인자 파싱
    args = parser.parse_args()

    # 학습 데이터 경로 및 설정된 하이퍼파라미터, 모델 저장 경로를 출력
    print(f"Training data directory: {args.train}")
    print(f"Number of estimators: {args.n_estimators}")
    print(f"Model save directory: {args.model_dir}")

    # 학습 데이터가 포함된 디렉토리 내 모든 파일 경로를 리스트로 가져오기
    input_files = [os.path.join(args.train, file) for file in os.listdir(args.train)]

    # 학습 데이터가 없을 경우 예외 발생 (학습 경로 지정 오류 또는 권한 문제 가능)
    if len(input_files) == 0:
        raise ValueError(
            ('There are no files in {}.\n'
             'This usually indicates that the channel ({}) was incorrectly specified,\n'
             'the data specification in S3 was incorrectly specified or the role specified\n'
             'does not have permission to access the data.').format(args.train, "train")
        )

    # 모든 CSV 파일을 하나의 Pandas 데이터프레임으로 병합 (첫 번째 행을 스킵하여 헤더 제거)
    raw_data = [pd.read_csv(file, header=None, engine="python", skiprows=1) for file in input_files]
    train_data = pd.concat(raw_data)

    # 데이터셋 분할: 레이블 (y)은 4번째 열, 특성 (X)은 마지막 열을 제외한 모든 열
    train_y = train_data.iloc[:, 4]  # 타겟 변수
    train_x = train_data.iloc[:, :-1]  # 피처 데이터

    # 학습 데이터 출력 (디버깅용)
    print(train_x)
    print(train_y)

    # 하이퍼파라미터 설정 (명령줄에서 제공된 값 사용)
    n_estimators = args.n_estimators if args.n_estimators > 0 else 20  # 기본값 20

    # 랜덤 포레스트 모델 초기화 및 학습 수행
    model = RandomForestClassifier(n_estimators=n_estimators)
    model.fit(train_x, train_y)

    # 학습된 모델을 지정된 디렉토리에 저장
    joblib.dump(model, os.path.join(args.model_dir, "model.joblib"))

# SageMaker가 모델을 로드할 때 사용하는 함수
def model_fn(model_dir):
    """SageMaker가 학습된 모델을 로드하는 함수

    SageMaker가 모델을 배포할 때 `model_fn`을 호출하여 모델을 불러온다.
    """
    # 저장된 모델 파일을 불러와 역직렬화 (Joblib을 이용하여 복원)
    model = joblib.load(os.path.join(model_dir, "model.joblib"))
    return model

# 모델 아티팩트(학습된 모델 파일) 저장 디렉토리 및 SageMaker 세션 생성
sagemaker_session = sagemaker.Session()

# 학습 데이터가 저장된 S3 경로 (train.csv 파일 위치)
training_input_path = f"s3://{bucket_name}/data/train/train.csv"

# 학습된 모델 아티팩트가 저장될 S3 경로
output_path = f"s3://{bucket_name}/model_artifacts"

# SageMaker Scikit-Learn Estimator 객체 생성
sk_estimator = SKLearn(
    entry_point="train.py",  # 학습 코드가 포함된 Python 파일 (SageMaker가 실행)
    role=role,  # SageMaker 실행 권한을 가진 IAM Role
    instance_count=1,  # EC2 인스턴스 개수 (단일 인스턴스에서 학습)
    instance_type="ml.c5.xlarge",  # 학습을 실행할 AWS EC2 인스턴스 유형 (CPU 기반)
    py_version="py3",  # Python 3 버전 사용
    framework_version="1.2-1",  # Scikit-Learn 프레임워크 버전 (로컬 scikit-learn 버전과 맞춰야 함)
    sagemaker_session=sagemaker_session,  # SageMaker 세션 연결
    hyperparameters={"n_estimators": 4},  # 학습 하이퍼파라미터 (랜덤 포레스트 트리 개수)
    output_path=output_path  # 학습된 모델이 저장될 S3 경로 지정
)

# 학습 시작 (S3에 저장된 학습 데이터를 사용하여 `train.py` 실행)
sk_estimator.fit({"train": training_input_path})

# SageMaker SKLearn 모델 클래스 임포트
from sagemaker.sklearn.model import SKLearnModel

# S3에 저장된 학습된 모델 아티팩트 경로 지정
model_data = f"s3://{bucket_name}/model_artifacts/sagemaker-scikit-learn-2025-03-10-00-35-44-515/output/model.tar.gz"

# SageMaker 세션 생성
sagemaker_session = sagemaker.Session()

# SKLearn 모델 객체 생성
sk_model = SKLearnModel(
    model_data=model_data,  # 모델 데이터 경로
    role=role,  # IAM 역할
    framework_version='1.2-1',  # SKLearn 프레임워크 버전
    entry_point='train.py',  # 모델 추론 코드가 있는 스크립트
    sagemaker_session=sagemaker_session  # SageMaker 세션
)

# 모델을 엔드포인트로 배포 (실시간 추론용)
predictor = sk_model.deploy(initial_instance_count=1, instance_type="ml.m5.xlarge")  # 인스턴스 1개로 ml.m5.xlarge 타입 사용

# 배포된 엔드포인트 테스트
from sklearn.metrics import accuracy_score

# 테스트 데이터 로드
test_data = pd.read_csv(os.path.join(data_dir, 'test.csv'))
X_test = test_data.drop('target', axis=1)

# 배포된 엔드포인트를 사용하여 예측 수행
predictions = predictor.predict(X_test.values)
print(predictions)

# 정확도 계산 및 출력
test_accuracy = accuracy_score(test_data['target'], predictions)
print(f"배포된 모델의 테스트 정확도: {test_accuracy:.4f}")

# 지속적인 비용 발생을 방지하기 위해 엔드포인트 삭제
predictor.delete_endpoint()

.
├── data
├── local_training.ipynb
├── model_artifacts
│   └── sk_rf_model.joblib
├── requirements.txt
└── train.py