lakeFS와 MLflow 함께 사용하기

MLflow는 머신러닝 라이프사이클을 관리하는 종합 도구로, ML 프로세스의 복잡성을 다루는 데 실무자와 팀을 돕는다. lakeFS는 MLflow의 MLflow Tracking 구성 요소와 매끄럽게 통합된다. lakeFS의 데이터 버전 관리와 MLflow의 입력 로깅을 조합하면 실험 재현성과 제로 카피 병렬 실험이 가능해져요.

출처: 문서

본문

MLflow는 머신러닝 라이프사이클을 관리하기 위해 설계된 종합 도구예요. ML 프로세스가 지닌 복잡성을 다루는 데 실무자와 팀을 도와주죠. 머신러닝 프로젝트의 전체 라이프사이클에 초점을 맞춰, 각 단계가 관리 가능하고 추적 가능하며 재현 가능하도록 보장해요.

MLflow는 여러 핵심 구성 요소로 이뤄져 있고, lakeFS는 MLflow Tracking 구성 요소와 매끄럽게 통합돼요. MLflow tracking은 입력과 출력을 모두 고려하는 실험 추적을 가능하게 해서, 실험 결과를 시각화하고 비교할 수 있게 해줘요.

MLflow와 lakeFS를 통합할 때의 이점

MLflow와 lakeFS를 통합하면 머신러닝 워크플로를 한층 끌어올리는 여러 장점이 있어요:

• 실험 재현성: MLflow의 입력 로깅 기능과 lakeFS의 데이터 버전 관리를 함께 활용하면, 각 실험 실행에서 사용한 정확한 데이터셋 버전을 추적할 수 있어요. 데이터셋이 진화하더라도 실험이 시간이 지나도 재현 가능하게 유지되죠.

• 제로 데이터 카피로 병렬 실험: lakeFS는 데이터를 복제하지 않고도 효율적으로 브랜치를 만들 수 있게 해줘요. 그래서 여러 실험을 병렬로 진행할 수 있고, 각 브랜치는 데이터셋 수정을 위한 격리된 환경을 제공해요. 한 브랜치의 변경은 다른 브랜치에 영향을 주지 않으니 팀원 간 안전한 협업이 촉진돼요. 실험이 끝나면 브랜치를 원본 데이터셋으로 매끄럽게 머지해 새로운 인사이트를 반영할 수 있어요.

lakeFS와 MLflow 함께 사용하는 방법

안전한 실험과 정확한 결과 재현을 위해 MLflow와 lakeFS의 결합된 능력을 활용하려면, 아래 워크플로를 참고하고 다음 섹션의 실용적인 예제를 살펴보세요.

권장 워크플로

• 실험마다 브랜치 만들기: 각 실험을 시작할 때 전용 lakeFS 브랜치를 만드세요. 이 접근법은 입력 데이터셋을 복제하지 않고도 안전하게 변경할 수 있게 해줘요. 나중에 이 브랜치에서 데이터를 읽어 MLflow 실험 실행에 사용하게 돼요.

• 실험 브랜치에서 데이터셋 읽기: 전용 브랜치에서 데이터를 직접 읽으며 실험을 진행하세요. 정확한 버전 추적을 위해 브랜치의 헤드 커밋에서 데이터셋을 읽는 걸 권장해요.

• lakeFS를 가리키는 MLflow Dataset 만들기: MLflow의 Dataset을 사용하되, 데이터셋 소스가 lakeFS를 가리키도록 하세요.

• 입력 로깅하기: MLflow의 log_input 함수로 lakeFS에 버전 관리되어 저장된 데이터셋을 로그로 남기세요.

• 데이터셋 변경 커밋하기: 머신러닝 개발은 본질적으로 반복적이에요. 입력 데이터셋을 변경하면, 의미 있는 커밋 메시지와 함께 lakeFS의 실험 브랜치에 커밋하세요. 실험 실행 중에는 브랜치 헤드 커밋에 해당하는 데이터셋 버전을 로드하고, 이 참조를 추적해 이후 결과 재현에 활용하세요.

• 실험 결과 머지하기: 실험을 마쳤다면, 선택한 실험 실행에 사용된 브랜치를 main 브랜치로 머지하세요.

실험마다 브랜치 Vs. 실험 실행마다 브랜치

실험 실행마다 lakeFS 브랜치를 만드는 것도 가능해요. lakeFS 브랜치는 만들기도 빠르고 비용도 거의 없으니까요. 하지만 실험마다 브랜치를 하나씩 만드는 게 보통 더 효율적이에요. 실험 브랜치의 헤드 커밋에서 바로 읽으면, 과도한 브랜치를 만들지 않고도 데이터셋 버전을 구분할 수 있어요. 이런 습관은 lakeFS 안에서 브랜치 위생을 지켜 줘요.

예제: Pandas 사용하기

import lakefs
import mlflow
import pandas as pd

repo = lakefs.Repository("my-repo")
repo_id = repo.id

exp_branch = repo.branch("experiment-1").create(source_reference="main", exist_ok=True)
branch_id = exp_branch.id
head_commit_id = exp_branch.head.id

table_path = "famous_people.csv"

dataset_source_url = f"s3://{repo_id}/{head_commit_id}/{table_path}"

# Use Pandas to read from lakeFS, at its most updated version to which the head commit id is pointing
raw_data = pd.read_csv(dataset_source_url, delimiter=";", storage_options={
        "key": "«redacted:AKIA…»",
        "secret": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
        "client_kwargs": {"endpoint_url": "http://localhost:8000"}
    })

# Create an instance of a PandasDataset
dataset = mlflow.data.from_pandas(
    raw_data, source=dataset_source_url, name="famous_people"
)

# View some of the recorded Dataset information
print(f"Dataset name: {dataset.name}")
print(f"Dataset source URI: {dataset.source.uri}")

# Use mlflow input logging to track the dataset versioned by lakeFS
with mlflow.start_run() as run:
    mlflow.log_input(dataset, context="training")
    mlflow.set_tag("lakefs_repo", repo_id)
    mlflow.set_tag("lakefs_branch", branch_id)
    mlflow.set_tag("lakefs_commit", head_commit_id)

# Inspect run's dataset
logged_run = mlflow.get_run(run.info.run_id) #

# Retrieve the Dataset object
logged_dataset = logged_run.inputs.dataset_inputs[0].dataset

# View some of the recorded Dataset information
print(f"Logged dataset name: {logged_dataset.name}")
print(f"Logged dataset source URI: {logged_dataset.source}")

Output

Dataset name: famous_people
Dataset source URI: s3://my-repo/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/fp.csv
Logged dataset name: famous_people
Logged dataset source URI: {"uri": "s3://my-repo/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/fp.csv"}

예제: Spark 사용하기

아래 예제는 lakeFS의 S3 호환 API에 접근하도록 Spark를 설정하고, 실험에 Delta Lake 테이블을 로드해요.

import lakefs
import mlflow
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("lakeFS / Mlflow") \
    .config("spark.hadoop.fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.endpoint", 'http://localhost:8000') \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .config("spark.hadoop.fs.s3a.access.key", 'AKIAlakefs12345EXAMPLE') \
    .config("spark.hadoop.fs.s3a.secret.key", 'abc/lakefs/1234567bPxRfiCYEXAMPLEKEY') \
    .config("spark.jars.packages", "io.delta:delta-core_2.12:2.3.0") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

repo = lakefs.Repository("my-repo")
repo_id = repo.id

exp_branch = repo.branch("experiment-1").create(source_reference="main", exist_ok=True)
branch_id = exp_branch.id
head_commit_id = exp_branch.head.id

table_path = "gold/train_v2/"

dataset_source_url = f"s3://{repo_id}/{head_commit_id}/{table_path}"

# Load delta lake table from lakeFS, at its most updated version to which the head commit id is pointing
dataset = mlflow.data.load_delta(path=dataset_source_url, name="boat-images")

# View some of the recorded Dataset information
print(f"Dataset name: {dataset.name}")
print(f"Dataset source URI: {dataset.source.path}")

# Use mlflow input logging to track the dataset versioned by lakeFS
with mlflow.start_run() as run:
    mlflow.log_input(dataset, context="training")
    mlflow.set_tag("lakefs_repo", repo_id)
    mlflow.set_tag("lakefs_branch", branch_id) # Log the branch id, to have a friendly lakeFS reference to search the input dataset in
    mlflow.set_tag("lakefs_commit", head_commit_id)

# Inspect run's dataset
logged_run = mlflow.get_run(run.info.run_id) #

# Retrieve the Dataset object
logged_dataset = logged_run.inputs.dataset_inputs[0].dataset

# View some of the recorded Dataset information
print(f"Logged dataset name: {logged_dataset.name}")
print(f"Logged dataset source URI: {logged_dataset.source}")

Output:

Dataset name: boat-images
Dataset source URI: s3://my-repo/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/gold/train_v2/
Logged dataset name: boat-images
Logged dataset source URI: {"path": "s3://my-repo/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/gold/train_v2/"}

실험 결과 재현하기

MLflow에서 특정 실험 실행의 결과를 재현하려면, 그 실행에서 사용된 정확한 데이터셋과 관련 메타데이터를 꺼내는 게 필수적이에요. MLflow Tracking UI는 전반적인 개요를 제공하지만, 상세한 데이터셋 정보와 소스는 프로그래밍 방식으로 접근하는 게 가장 좋아요.

• Run ID 얻기: MLflow UI로 이동해 관심 있는 실험의 Run ID를 복사하세요.

• MLflow의 Python SDK로 데이터셋 정보 추출하기:

import mlflow

# Inspect run's dataset and tags
run_id = "c0f8fbb1b63748abaa0a6479115e272c"
run = mlflow.get_run(run_id)

# Retrieve the Dataset object
logged_dataset = run.inputs.dataset_inputs[0].dataset

# View some of the recorded Dataset information
print(f"Run ID: {run_id} Dataset name: {logged_dataset.name}")
print(f"Run ID: {run_id} Dataset source URI: {logged_dataset.source}")

# Retrieve run's tags
logged_tags = run.data.tags
print(f"Run ID: {run_id} tags: {logged_tags}")

Output

Run ID: c0f8fbb1b63748abaa0a6479115e272c Dataset name: boat-images
Run ID: c0f8fbb1b63748abaa0a6479115e272c Dataset source URI: {"path": "s3://my-repo/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/gold/train_v2/"}
Run ID: c0f8fbb1b63748abaa0a6479115e272c tags: {'lakefs_branch': 'experiment-1', 'lakefs_repo': 'my-repo', 'lakefs_commit': '3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3'}

Notes

• Dataset Source URI는 실행에서 사용된 정확한 버전의 데이터셋 위치를 제공해요.

• 'lakefs_branch'와 'lakefs_repo' 같은 Run 태그는 lakeFS 안에서 데이터셋의 출처에 대한 추가 맥락을 제공해요.

실행 간 입력 비교하기

서로 다른 두 MLflow 실행이 같은 입력 데이터셋을 사용했는지 판단하려면, 로그된 Dataset 객체의 특정 속성을 비교하면 돼요. 버전 관리된 데이터셋의 URI를 담고 있는 source 속성이 이 비교에 흔히 쓰여요. 예제예요:

import mlflow

first_run_id = "4c0464d665944dc5bb90587d455948b8"
first_run = mlflow.get_run(first_run_id)

# Retrieve the Dataset object
first_dataset = first_run.inputs.dataset_inputs[0].dataset
first_dataset_src = first_dataset.source

sec_run_id = "12b91e073a8b40df97ea8d570534de31"
sec_run = mlflow.get_run(sec_run_id)

# Retrieve the Dataset object
sec_dataset = sec_run.inputs.dataset_inputs[0].dataset
sec_dataset_src = sec_dataset.source

assert first_dataset_src == sec_dataset_src, "Dataset sources are not equal."

print(f"First dataset src: {first_dataset_src}")
print(f"Second dataset src: {sec_dataset_src}")

Output

First dataset src: {"uri": "s3://mlflow-tracking/f16682c0186a81adf72edaad23c3f59ed2ff3afddad4fef987b4919f5e82003a/gold/train_v2/"}
Second dataset src: {"uri": "s3://mlflow-tracking/3afddad4fef987b4919f5e82f16682c018f59ed2ff003a6a81adf72edaad23c3/gold/train_v2/"}

이 예제에서는 각 Dataset 객체의 source 속성을 비교해 입력 데이터셋이 동일한지 판단해요. 서로 다르다면 더 깊이 들여다볼 수 있어요. 데이터셋 소스 URI를 손에 넣었다면, lakeFS로 데이터셋에 가해진 변경에 대한 더 많은 인사이트를 얻을 수 있어요:

• lakeFS 커밋 ID 살펴보기: URI 안의 커밋 ID를 확인하면, 변경의 작성자와 목적을 포함한 커밋의 상세 정보를 얻을 수 있어요.

• lakeFS Diff 사용하기: lakeFS는 데이터의 서로 다른 버전을 비교할 수 있는 diff 기능을 제공해요.

이 도구들을 활용하면 여러 MLflow 실행에 걸쳐 데이터셋이 어떻게 진화했는지 효과적으로 추적하고 이해할 수 있어요.

더 알아보기 (Learn more)

공식 문서의 자세한 내용은 https://docs.lakefs.io/integrations/mlflow/에서 확인하실 수 있어요.