대용량 처리와 AI/ML 활용

# 대용량 파일 업로드: 청크 단위 업로드
import os

def upload_large_file(local_path, volume_path, chunk_size=50*1024*1024):
    """50MB 청크 단위로 대용량 파일 업로드"""
    file_size = os.path.getsize(local_path)

    if file_size < chunk_size:
        # 소규모 파일: 직접 복사
        dbutils.fs.cp(f"file:{local_path}", volume_path)
    else:
        # 대용량 파일: dbutils.fs.cp 사용 (내부적으로 멀티파트 업로드)
        dbutils.fs.cp(f"file:{local_path}", volume_path)
        print(f"Uploaded {file_size:,} bytes to {volume_path}")

# 디렉토리 전체 업로드 (병렬)
dbutils.fs.cp("file:/tmp/large_dataset/",
              "/Volumes/production/ml/training_data/",
              recurse=True)

from datetime import datetime, timedelta

def cleanup_old_files(volume_path, days_to_keep=90):
    """지정 기간보다 오래된 파일을 자동 삭제"""
    cutoff = datetime.now() - timedelta(days=days_to_keep)

    files = dbutils.fs.ls(volume_path)
    deleted_count = 0

    for f in files:
        # 파일 수정 시간 확인
        mod_time = datetime.fromtimestamp(f.modificationTime / 1000)
        if mod_time < cutoff:
            dbutils.fs.rm(f.path)
            deleted_count += 1

    print(f"Deleted {deleted_count} files older than {days_to_keep} days")

# 매일 실행 (Jobs에서 스케줄)
cleanup_old_files("/Volumes/production/ecommerce/temp_files/", days_to_keep=30)

# Volume을 소스로 하는 Auto Loader 스트리밍
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", "/Volumes/production/ecommerce/_schemas/orders/")
    .option("header", "true")
    .load("/Volumes/production/ecommerce/raw_files/orders/")
)

# Delta 테이블로 적재
(df.writeStream
    .option("checkpointLocation", "/Volumes/production/ecommerce/_checkpoints/orders/")
    .trigger(availableNow=True)
    .toTable("production.ecommerce.bronze_orders")
)

# 처리 완료 파일을 아카이브 Volume으로 이동
import os
from datetime import datetime

landing_path = "/Volumes/production/etl/landing/orders/"
archive_path = "/Volumes/production/etl/archive/orders/"

for f in dbutils.fs.ls(landing_path):
    if f.name.endswith(".csv"):
        # 처리 완료 확인 후 이동
        archive_dest = f"{archive_path}{datetime.now().strftime('%Y/%m/')}{f.name}"
        dbutils.fs.mv(f.path, archive_dest)

# 이미지 분류 모델용 학습 데이터 구조
# /Volumes/production/ml/training_images/
#   ├── cats/
#   │    ├── cat_001.jpg
#   │    └── cat_002.jpg
#   └── dogs/
#        ├── dog_001.jpg
#        └── dog_002.jpg

# PyTorch DataLoader에서 Volume 경로 직접 사용
from torchvision import datasets, transforms

train_dataset = datasets.ImageFolder(
    root="/Volumes/production/ml/training_images/",
    transform=transforms.Compose([
        transforms.Resize((224, 224)),
        transforms.ToTensor(),
    ])
)

# 대형 모델 파일을 Volume에 저장
model_path = "/Volumes/production/ml/model_artifacts/"

# Hugging Face 모델 다운로드 → Volume에 저장
from huggingface_hub import snapshot_download
snapshot_download(
    repo_id="meta-llama/Llama-3-8B",
    local_dir=f"{model_path}llama-3-8b/"
)

# MLflow에서 Volume의 모델 참조
import mlflow
mlflow.pyfunc.log_model(
    artifact_path="model",
    python_model=my_model,
    artifacts={"model_weights": f"{model_path}llama-3-8b/"}
)

# RAG 파이프라인의 문서 소스로 Volume 활용
docs_volume = "/Volumes/production/rag/source_documents/"

# 문서 업로드
dbutils.fs.cp("file:/tmp/company_manual.pdf", f"{docs_volume}manuals/company_manual.pdf")

# 문서 파싱 → 청킹 → Vector Search 인덱스
parsed_docs = spark.sql(f"""
    SELECT
        path,
        ai_parse_document(path, 'markdown') AS content
    FROM read_files('{docs_volume}manuals/', format => 'binaryFile')
""")