브랜치 & 머지 다루기

브랜치는 lakeFS 버전 컨트롤의 기반이에요. 병렬 개발, 안전한 실험, 협업 데이터 관리를 가능하게 해주죠. 이 가이드는 Python SDK를 사용한 모든 브랜치 연산을 다룹니다.

출처: 브랜치 & 머지 다루기

본문

저장소와 브랜치 이해하기

저장소(Repository)

저장소는 모든 데이터와 버전 히스토리를 담는 버전 관리 스토리지 네임스페이스예요. 스토리지 위치를 지정해서 저장소를 만들어요:

import lakefs

repo = lakefs.repository("my-repo").create(
    storage_namespace="s3://my-bucket/data"
)

브랜치(Branch)

브랜치는 병렬 개발과 실험을 가능하게 해줘요. 각 브랜치는 데이터의 자체 버전을 유지합니다:

branch = lakefs.repository("my-repo").branch("main")
experiment_branch = lakefs.repository("my-repo").branch("experiment1").create(
    source_reference="main"
)

브랜치 생성 및 목록 조회

브랜치 생성

임의의 레퍼런스(브랜치, 태그, 커밋)에서 새 브랜치를 만들어요:

import lakefs

# Initialize repository
repo = lakefs.repository("my-data-repo")

# Create a branch from main
experiment_branch = repo.branch("experiment-1").create(source_reference="main")

print(f"Created branch: {experiment_branch.id}")

# Check what commit this branch points to
commit = experiment_branch.get_commit()
print(f"Branch head: {commit.id}")

명시적 클라이언트 사용:

import lakefs
from lakefs.client import Client

# Create client
client = Client(
    host="http://localhost:8000",
    username="your-access-key",
    password="your-secret-key"
)

# Create branch with explicit client
branch = lakefs.repository("my-repo", client=client).branch("dev").create(
    source_reference="main"
)

브랜치 목록 조회

저장소의 모든 브랜치를 나열해요:

import lakefs

repo = lakefs.repository("my-data-repo")

print("All branches:")
for branch in repo.branches():
    commit = branch.get_commit()
    print(f"  {branch.id:20} -> {commit.id[:8]}... ({commit.message})")

브랜치 헤드 확인

브랜치가 현재 가리키는 커밋을 가져와요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("main")

# Get the head commit
head_commit = branch.get_commit()
print(f"Current head: {head_commit.id}")
print(f"Committed by: {head_commit.committer}")
print(f"Message: {head_commit.message}")

# Or use the head property
head_ref = branch.head
print(f"Head reference ID: {head_ref.id}")

브랜치 콘텐츠 다루기

커밋되지 않은 변경 사항 보기

마지막 커밋 이후 브랜치에서 무엇이 바뀌었는지 확인해요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("feature-branch")

# List uncommitted changes
print("Uncommitted changes:")
for change in branch.uncommitted():
    print(f"  {change.type:10} {change.path} ({change.size_bytes} bytes)")

# Count changes
change_count = len(list(branch.uncommitted()))
print(f"Total changes: {change_count}")

# Filter changes by prefix
data_changes = [c for c in branch.uncommitted() if c.path.startswith("data/")]
print(f"Changes in data/ folder: {len(data_changes)}")

변경 사항 커밋

변경 사항으로 커밋을 만들어요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("feature-branch")

# Upload some data first
branch.object("data/dataset.csv").upload(data=b"id,value\n1,100\n2,200")

# Commit with message and metadata
ref = branch.commit(
    message="Add customer dataset",
    metadata={
        "author": "data-team",
        "version": "1.0",
        "dataset-type": "raw"
    }
)

print(f"Committed: {ref.id}")
print(f"Message: {ref.get_commit().message}")
print(f"Metadata: {ref.get_commit().metadata}")

브랜치 비교하기

레퍼런스 간 Diff

두 브랜치 또는 커밋 사이에 무엇이 바뀌었는지 확인해요:

import lakefs

repo = lakefs.repository("my-data-repo")
main_branch = repo.branch("main")
feature_branch = repo.branch("feature-add-models")

# Compare branches
print("Changes in feature-add-models vs main:")
for change in main_branch.diff(other_ref=feature_branch):
    print(f"  {change.type:10} {change.path}")

# Count different types of changes
changes = list(main_branch.diff(other_ref=feature_branch))
added = len([c for c in changes if c.type == "added"])
removed = len([c for c in changes if c.type == "removed"])
modified = len([c for c in changes if c.type == "modified"])

print(f"Added: {added}, Removed: {removed}, Modified: {modified}")

브랜치 머지

다른 브랜치로 머지

한 브랜치의 변경 사항을 다른 브랜치로 머지해요:

import lakefs

repo = lakefs.repository("my-data-repo")
feature_branch = repo.branch("feature-branch")
main_branch = repo.branch("main")

try:
    # Merge feature branch into main
    merge_result = feature_branch.merge_into(main_branch)
    print(f"Merge successful: {merge_result}")

    # Verify merge by checking that differences are gone
    remaining_diffs = list(main_branch.diff(other_ref=feature_branch))
    print(f"Remaining differences: {len(remaining_diffs)}")

except Exception as e:
    print(f"Merge failed: {e}")
    # Resolve conflicts or adjust data and try again

충돌 감지와 함께 머지

import lakefs

repo = lakefs.repository("my-data-repo")

try:
    branch1 = repo.branch("feature-1")
    branch2 = repo.branch("feature-2")
    main = repo.branch("main")

    # Check for conflicts before merging
    conflicts = list(main.diff(other_ref=branch1))

    if conflicts:
        print(f"Potential conflicts: {len(conflicts)} changes")
        for change in conflicts[:5]:  # Show first 5
            print(f"  {change.type}: {change.path}")

    # Proceed with merge if acceptable
    merge_commit = branch1.merge_into(main)
    print(f"Merged into main: {merge_commit}")

except Exception as e:
    print(f"Merge error: {e}")

커밋 체리픽(Cherry-Picking)

한 브랜치의 특정 커밋을 다른 브랜치에 적용해요:

import lakefs

repo = lakefs.repository("my-data-repo")
main_branch = repo.branch("main")
release_branch = repo.branch("release-v1.0")

try:
    # Cherry-pick a specific commit onto release branch
    # First, get a commit ID from main
    main_commits = list(main_branch.log(max_amount=5))
    if main_commits:
        commit_to_cherry_pick = main_commits[0]

        # Cherry-pick it onto release branch
        new_commit = release_branch.cherry_pick(commit_to_cherry_pick.id)
        print(f"Cherry-picked commit: {new_commit.id}")
        print(f"Message: {new_commit.message}")

except Exception as e:
    print(f"Cherry-pick failed: {e}")

변경 사항 되돌리기

커밋 되돌리기(Revert)

특정 커밋의 변경 사항을 취소해요:

import lakefs

repo = lakefs.repository("my-data-repo")
branch = repo.branch("develop")

try:
    # Get recent commits
    recent_commits = list(branch.log(max_amount=10))

    if len(recent_commits) > 1:
        # Revert the most recent commit
        commit_to_revert = recent_commits[0]

        revert_result = branch.revert(reference=commit_to_revert.id)
        print(f"Reverted commit: {commit_to_revert.id}")
        print(f"New commit created: {revert_result.id}")

except Exception as e:
    print(f"Revert failed: {e}")

머지 커밋 되돌리기

머지 커밋을 되돌릴 때는 어떤 부모를 기준으로 되돌릴지 지정할 수 있어요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("main")

try:
    # Get commit history
    commits = list(branch.log(max_amount=5))

    for commit in commits:
        # If this is a merge commit (has multiple parents)
        if len(commit.parents) > 1:
            # Revert against parent 1 (the original main)
            result = branch.revert(reference=commit.id, parent_number=1)
            print(f"Reverted merge commit: {commit.id}")
            break

except Exception as e:
    print(f"Error: {e}")

브랜치 리셋

커밋되지 않은 변경 사항 리셋

브랜치에서 커밋되지 않은 변경 사항을 버려요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("feature-branch")

# Show uncommitted changes before reset
changes_before = list(branch.uncommitted())
print(f"Uncommitted changes before reset: {len(changes_before)}")

try:
    # Reset all changes
    branch.reset_changes(path_type="reset")

    # Verify changes are gone
    changes_after = list(branch.uncommitted())
    print(f"Uncommitted changes after reset: {len(changes_after)}")

except Exception as e:
    print(f"Reset failed: {e}")

특정 경로의 변경 사항 리셋

특정 경로나 프리픽스에 대해서만 변경 사항을 리셋해요:

import lakefs

branch = lakefs.repository("my-data-repo").branch("develop")

try:
    # Reset changes for a specific object
    branch.reset_changes(path_type="object", path="data/temp.csv")
    print("Reset: data/temp.csv")

    # Reset all changes in a folder (common prefix)
    branch.reset_changes(path_type="common_prefix", path="logs/")
    print("Reset: logs/*")

    # Verify remaining changes
    remaining = list(branch.uncommitted())
    print(f"Remaining changes: {len(remaining)}")

except Exception as e:
    print(f"Reset error: {e}")

브랜치 삭제

브랜치 삭제

더 이상 필요 없는 브랜치를 제거해요:

import lakefs

repo = lakefs.repository("my-data-repo")

try:
    # Delete a branch
    branch = repo.branch("old-experiment")
    branch.delete()
    print("Branch deleted: old-experiment")

except Exception as e:
    print(f"Delete failed: {e}")

# Verify it's gone
remaining_branches = [b.id for b in repo.branches()]
print(f"Remaining branches: {remaining_branches}")

실전 워크플로

격리된 개발/테스트 환경

피처 브랜치 워크플로

전형적인 피처 개발 워크플로를 구현해요:

import lakefs

def feature_workflow(repo_name, feature_name, data_updates):
    """
    Complete feature branch workflow:
    1. Create feature branch
    2. Make changes
    3. Test/commit
    4. Merge to main
    """
    repo = lakefs.repository(repo_name)
    main = repo.branch("main")
    feature = repo.branch(feature_name)

    try:
        # 1. Create feature branch from main
        feature.create(source_reference="main")
        print(f"Created feature branch: {feature_name}")

        # 2. Make changes
        for path, data in data_updates.items():
            feature.object(path).upload(data=data)
        print(f"Uploaded {len(data_updates)} objects")

        # 3. Review changes
        changes = list(feature.uncommitted())
        print(f"Changes to commit: {len(changes)}")

        # 4. Commit
        commit_ref = feature.commit(
            message=f"Implement {feature_name}",
            metadata={"feature": feature_name}
        )
        print(f"Committed: {commit_ref.id}")

        # 5. Verify diff before merge
        diff = list(main.diff(other_ref=feature))
        print(f"Ready to merge {len(diff)} changes")

        # 6. Merge to main
        merge_result = feature.merge_into(main)
        print(f"Merged to main: {merge_result}")

        # 7. Cleanup
        feature.delete()
        print(f"Deleted feature branch: {feature_name}")

        return True

    except Exception as e:
        print(f"Feature workflow failed: {e}")
        # Cleanup on error
        try:
            feature.delete()
        except:
            pass
        return False

# Usage:
feature_workflow(
    repo_name="analytics-repo",
    feature_name="add-customer-metrics",
    data_updates={
        "data/metrics/customer_v2.csv": b"id,value\n1,100\n2,200",
        "data/metrics/metadata.json": b'{"version": "2", "date": "2024-01-15"}'
    }
)

실험 브랜치 워크플로

실험을 위한 격리된 브랜치를 만들어요:

import lakefs
import time

def create_experiment(repo_name, experiment_name, experiment_logic):
    """Run an experiment in isolation, keep results if successful"""
    repo = lakefs.repository(repo_name)
    exp_branch = repo.branch(experiment_name)

    try:
        # Create isolated experiment branch
        exp_branch.create(source_reference="main")
        print(f"Started experiment: {experiment_name}")

        # Run experiment logic
        experiment_logic(exp_branch)

        # Commit results
        changes = list(exp_branch.uncommitted())
        if changes:
            exp_branch.commit(
                message=f"Results from {experiment_name}",
                metadata={"experiment": experiment_name, "timestamp": str(time.time())}
            )
            print(f"Experiment successful, committed {len(changes)} changes")

        return exp_branch.get_commit().id

    except Exception as e:
        print(f"Experiment failed: {e}")
        # Clean up failed experiment
        try:
            exp_branch.delete()
        except:
            pass
        return None

# Example experiment
def model_training_experiment(branch):
    # Simulate model training
    branch.object("models/trained_v2.pkl").upload(data=b"model_binary_data")
    branch.object("logs/training.log").upload(data=b"Training complete")

# Usage:
result_commit = create_experiment(
    repo_name="ml-repo",
    experiment_name="neural-net-v3",
    experiment_logic=model_training_experiment
)

if result_commit:
    print(f"Keep results from commit: {result_commit}")

릴리스 브랜치 관리

브랜치와 태그로 릴리스 버전을 관리해요:

import lakefs

def create_release(repo_name, version_tag):
    """Create a release: branch from main, prepare, merge, tag"""
    repo = lakefs.repository(repo_name)
    main = repo.branch("main")
    release_branch = repo.branch(f"release-{version_tag}")

    try:
        # 1. Create release branch
        release_branch.create(source_reference="main")
        print(f"Created release branch: release-{version_tag}")

        # 2. Any release-specific changes (version bumps, etc)
        release_branch.object("VERSION").upload(data=version_tag.encode())
        release_branch.object("RELEASE_NOTES.md").upload(
            data=f"# Release {version_tag}\nDate: 2024-01-15".encode()
        )

        # 3. Commit release changes
        release_ref = release_branch.commit(
            message=f"Release {version_tag}",
            metadata={"version": version_tag, "type": "release"}
        )

        # 4. Merge back to main
        release_branch.merge_into(main)
        print(f"Merged release to main")

        # 5. Create tag for this release
        tag = repo.tag(f"v{version_tag}").create(source_ref=release_ref)
        print(f"Tagged release: v{version_tag}")

        # 6. Clean up release branch
        release_branch.delete()
        print(f"Cleaned up release branch")

        return f"v{version_tag}"

    except Exception as e:
        print(f"Release creation failed: {e}")
        return None

# Usage:
release_tag = create_release("prod-repo", "1.5.0")
if release_tag:
    print(f"Release ready: {release_tag}")

에러 처리

자주 발생하는 브랜치 에러 다루기

import lakefs
from lakefs.exceptions import NotFoundException, ConflictException, ForbiddenException

repo = lakefs.repository("my-data-repo")

# Branch already exists
try:
    branch = repo.branch("main").create(source_reference="main", exist_ok=False)
except ConflictException:
    print("Branch already exists")
    branch = repo.branch("main")

# Branch doesn't exist
try:
    branch = repo.branch("non-existent")
    commit = branch.get_commit()
except NotFoundException:
    print("Branch not found")

# Protected branch (cannot delete)
try:
    repo.branch("main").delete()
except ForbiddenException:
    print("Cannot delete protected branch")

# Source reference not found
try:
    branch = repo.branch("new-branch").create(
        source_reference="non-existent-ref"
    )
except NotFoundException:
    print("Source reference does not exist")

더 알아보기 (Learn more)

공식 문서의 원문은 https://docs.lakefs.io/reference/python/branches/ 에서 확인할 수 있어요.