고급 사용법

이 섹션에서는 Atio의 고급 기능들을 다룹니다.

스냅샷 기반 버전 관리

Atio는 데이터의 버전을 관리할 수 있는 스냅샷 시스템을 제공합니다.

기본 스냅샷 쓰기

import atio
import pandas as pd

# 데이터 생성
df = pd.DataFrame({
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"]
})

# 스냅샷으로 저장 (버전 관리)
atio.write_snapshot(df, "users_table", format="parquet")

# 새로운 데이터로 업데이트
df_new = pd.DataFrame({
    "id": [1, 2, 3, 4],
    "name": ["Alice", "Bob", "Charlie", "David"]
})

# append 모드로 스냅샷 추가
atio.write_snapshot(df_new, "users_table", mode="append", format="parquet")

스냅샷 읽기

# 최신 버전 읽기
latest_data = atio.read_table("users_table")

# 특정 버전 읽기
version_1_data = atio.read_table("users_table", version=1)

# Polars로 읽기
polars_data = atio.read_table("users_table", output_as="polars")

스냅샷 정리

from datetime import timedelta

# 7일 이상 된 스냅샷 삭제 (dry run)
atio.expire_snapshots("users_table", keep_for=timedelta(days=7), dry_run=True)

# 실제 삭제 실행
atio.expire_snapshots("users_table", keep_for=timedelta(days=7), dry_run=False)

데이터베이스 연동

Pandas와 Polars를 사용하여 데이터베이스에 안전하게 데이터를 저장할 수 있습니다.

Pandas SQL 연동

import atio
import pandas as pd
from sqlalchemy import create_engine

# 데이터베이스 연결
engine = create_engine('postgresql://user:password@localhost/dbname')

df = pd.DataFrame({
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"]
})

# SQL 데이터베이스에 저장
atio.write(df, format="sql", name="users", con=engine)

Polars 데이터베이스 연동

import atio
import polars as pl

df = pl.DataFrame({
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"]
})

# 데이터베이스에 저장
atio.write(df, format="database",
           table_name="users",
           connection_uri="postgresql://user:password@localhost/dbname")

성능 최적화

진행도 표시

대용량 파일 처리 시 진행 상황을 실시간으로 확인할 수 있습니다.

import atio
import pandas as pd
import numpy as np

# 대용량 데이터 생성
large_df = pd.DataFrame(np.random.randn(1000000, 10))

# 진행도 표시와 함께 저장
atio.write(large_df, "large_data.parquet",
           format="parquet",
           show_progress=True)

상세 로깅

성능 진단을 위한 상세한 로깅을 활성화할 수 있습니다.

# 상세한 성능 정보 출력
atio.write(df, "data.parquet", format="parquet", verbose=True)

이를 통해 다음과 같은 정보를 확인할 수 있습니다:

각 단계별 소요 시간
임시 파일 생성 및 교체 과정
백업 및 롤백 과정
성능 병목점 분석

에러 처리

Atio는 다양한 에러 상황에 대해 안전하게 처리합니다.

파일 시스템 에러

import atio
import pandas as pd

df = pd.DataFrame({"a": [1, 2, 3]})

try:
    # 권한이 없는 디렉토리에 저장 시도
    atio.write(df, "/root/data.parquet", format="parquet")
except PermissionError as e:
    print(f"권한 에러: {e}")
    # 원본 파일은 그대로 보존됨

포맷 에러

try:
    # 지원하지 않는 포맷 사용
    atio.write(df, "data.unknown", format="unknown")
except ValueError as e:
    print(f"지원하지 않는 포맷: {e}")

데이터베이스 에러

try:
    # 잘못된 데이터베이스 연결 정보
    atio.write(df, format="sql",
               name="users",
               con="invalid_connection")
except Exception as e:
    print(f"데이터베이스 에러: {e}")

플러그인 확장

Atio는 플러그인 아키텍처를 통해 새로운 형식을 쉽게 추가할 수 있습니다.

커스텀 형식 등록

from atio.plugins import register_writer
import pandas as pd

# 커스텀 형식 등록
def custom_writer(df, path, **kwargs):
    # 커스텀 저장 로직
    with open(path, 'w') as f:
        f.write("Custom format\n")
        f.write(df.to_string())

# Pandas DataFrame에 대한 커스텀 형식 등록
register_writer(pd.DataFrame, "custom", custom_writer)

# 사용
df = pd.DataFrame({"a": [1, 2, 3]})
atio.write(df, "data.custom", format="custom")

NumPy 배열 처리

NumPy 배열의 다양한 저장 방식을 지원합니다.

단일 배열 저장

import atio
import numpy as np

arr = np.array([[1, 2, 3], [4, 5, 6]])

# .npy 파일로 저장
atio.write(arr, "array.npy", format="npy")

# .csv 파일로 저장
atio.write(arr, "array.csv", format="csv")

여러 배열 저장

# 여러 배열을 딕셔너리로 저장
arrays = {
    "features": np.random.randn(1000, 10),
    "labels": np.random.randint(0, 2, 1000),
    "metadata": np.array([1, 2, 3, 4, 5])
}

# 압축된 .npz 파일로 저장
atio.write(arrays, "data.npz", format="npz_compressed")

실제 사용 사례

머신러닝 파이프라인

import atio
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier

# 데이터 로드 및 전처리
df = pd.read_csv("raw_data.csv")

# 전처리된 데이터를 안전하게 저장
atio.write(df, "processed_data.parquet", format="parquet")

# 학습/테스트 분할
X_train, X_test, y_train, y_test = train_test_split(
    df.drop('target', axis=1), df['target'], test_size=0.2
)

# 분할된 데이터를 스냅샷으로 저장
atio.write_snapshot(X_train, "training_data", format="parquet")
atio.write_snapshot(X_test, "test_data", format="parquet")

# 모델 학습
model = RandomForestClassifier()
model.fit(X_train, y_train)

# 예측 결과를 안전하게 저장
predictions = model.predict(X_test)
results_df = pd.DataFrame({
    'actual': y_test,
    'predicted': predictions
})

atio.write(results_df, "predictions.parquet", format="parquet")

ETL 파이프라인

import atio
import pandas as pd
from sqlalchemy import create_engine

# 1. 원본 데이터 로드
raw_data = pd.read_csv("source_data.csv")

# 2. 데이터 정제
cleaned_data = raw_data.dropna()
cleaned_data = cleaned_data[cleaned_data['value'] > 0]

# 3. 안전하게 정제된 데이터 저장
atio.write(cleaned_data, "cleaned_data.parquet", format="parquet")

# 4. 데이터베이스에 저장
engine = create_engine('postgresql://user:password@localhost/warehouse')
atio.write(cleaned_data, format="sql",
           name="processed_table",
           con=engine,
           if_exists='replace')

# 5. 스냅샷으로 버전 관리
atio.write_snapshot(cleaned_data, "daily_processed", format="parquet")