[HuggingFace] 허깅페이스 데이터 다운로드하기

benchmark dataset을 다운받는 경우 예전에는 다른 외부 사이트나 구글드라이브에 올려진 데이터를 다운받는 경우가 많았던 것 같은데 요새는 hugging face에 많이 업로드하는 것 같다. 로그인 없이 직접 다운로드 받을수도 있지만, access를 요청해서 승인을 받아야 다운로드를 받을 수 있는 경우도 존재한다. 그래서, python에서 직접 hugging face에 업로드 된 benchmark datatset을 다운로드 받는 방법을 정리해보았다. 

 

 

1. hugging face 사이트에서 로그인 후 dataset에 대한 access 요청

 

https://huggingface.co/

Hugging Face – The AI community building the future.

 

위 링크에 들어가 hugging face에 로그인을 해준 뒤 원하는 데이터를 고르고 dataset에 대한 access 요청을 해야하는 받을 수 있는 경우 요청한다. 예시로 PathCap이란 데이터셋을 다운로드 하고자 하는 경우 아래와 사진과 같이 된다.

 

로그인과 요청 전 화면

 

로그인과 요청 후 화면

 

 

요청을 위해서 복잡한 사항을 요구하진 않고 간단한 소속이나 이용목적 등 정도를 물어봤던 것 같다. 상업적으로 만들어서 팔 것이 아니라면 승인을 대부분 해준다.

 

 

2.  hugging face 계정의 token 생성

 

본인의 계정을 등록해야 승인을 받았다는 것을 검증할 수 있으므로 파이썬 코드에서 사용할 hugging face 계정의 token을 받아준다.

 

hugging face token 관리 화면

 

 

이때, read 나 write로 생성해야 파이썬 코드에서 접근이 가능하다.

 

 

3. 개별 파일 혹은 레파지토리 자체 다운로드는 아래의 코드로 다운로드 한다.

 

개별 파일 다운로드 시 아래의 코드를 이용한다.

import requests
import os

# Hugging Face Access Token 설정
HF_TOKEN = ""  # 실제 토큰으로 교체하세요.

# 다운로드할 파일의 경로와 이름
repo_id = "jamessyx/PathCap"
file_path = "data.json"
repo_type = "dataset"

# 출력 디렉토리 설정
output_dir = "/data/Public/PathCap/"

# 출력 디렉토리가 없으면 생성
os.makedirs(output_dir, exist_ok=True)

# 파일 다운로드 URL 생성
# Hugging Face Hub의 파일 다운로드 URL 형식은 다음과 같습니다:
# https://huggingface.co/datasets/{repo_id}/resolve/main/{file_path}
# 또는 특정 브랜치를 사용하는 경우:
# https://huggingface.co/datasets/{repo_id}/resolve/{branch}/{file_path}

# 여기서는 기본 브랜치(main)를 사용한다고 가정합니다.
download_url = f"https://huggingface.co/datasets/{repo_id}/resolve/main/{file_path}"

# HTTP GET 요청 헤더에 인증 토큰 포함
headers = {
    "Authorization": f"Bearer {HF_TOKEN}"
}

# 파일 다운로드
print("데이터 다운로드 중...")
response = requests.get(download_url, headers=headers, stream=True)

if response.status_code == 200:
    destination_path = os.path.join(output_dir, os.path.basename(file_path))
    with open(destination_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)
    print(f"파일이 {destination_path}에 다운로드되었습니다.")
else:
    print(f"파일 다운로드 실패: HTTP {response.status_code}")
    print("응답 내용:", response.text)

 

 

레파지토리 자체를 다운로드 하고 싶을 경우 아래의 코드를 이용한다.

import os
import requests
from huggingface_hub import HfApi
from tqdm import tqdm
import concurrent.futures

# Hugging Face Access Token 설정
HF_TOKEN = ""  # 실제 토큰으로 교체하세요.

# 다운로드할 리포지토리 정보
repo_id = "jamessyx/PathCap"
repo_type = "dataset"  # 또는 "model" 등 적절한 타입으로 설정

# 출력 디렉토리 설정
output_dir = "PathCap/"

# 출력 디렉토리가 없으면 생성
os.makedirs(output_dir, exist_ok=True)

# HfApi 인스턴스 생성
api = HfApi()

# 리포지토리의 파일 목록 가져오기
print("리포지토리의 파일 목록을 가져오는 중...")
try:
    files = api.list_repo_files(repo_id=repo_id, repo_type=repo_type, use_auth_token=HF_TOKEN)
    print(f"총 {len(files)}개의 파일을 발견했습니다.")
except Exception as e:
    print(f"파일 목록을 가져오는 중 오류 발생: {e}")
    files = []

# 리포지토리의 기본 브랜치 정보 가져오기
print("리포지토리의 기본 브랜치를 가져오는 중...")
try:
    repo_info = api.repo_info(repo_id=repo_id, repo_type=repo_type, use_auth_token=HF_TOKEN)
    branch = repo_info.default_branch
    print(f"기본 브랜치: {branch}")
except Exception as e:
    print(f"기본 브랜치를 가져오는 중 오류 발생: {e}")
    branch = "main"  # 기본값

# 파일 다운로드 함수 정의
def download_file(file_path, base_url, headers, output_directory):
    download_url = f"{base_url}/{file_path}"
    try:
        response = requests.get(download_url, headers=headers, stream=True, timeout=30)
        if response.status_code == 200:
            destination_path = os.path.join(output_directory, file_path)
            destination_dir = os.path.dirname(destination_path)
            os.makedirs(destination_dir, exist_ok=True)
            with open(destination_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
            return f"성공: {destination_path}"
        else:
            return f"실패: {file_path} (HTTP {response.status_code})"
    except requests.exceptions.RequestException as e:
        return f"실패: {file_path} (오류: {e})"

# 파일 다운로드를 위한 기본 URL 설정
base_url = f"https://huggingface.co/datasets/{repo_id}/resolve/{branch}"

# HTTP 요청 헤더에 인증 토큰 포함
headers = {
    "Authorization": f"Bearer {HF_TOKEN}"
}

# 병렬 다운로드를 위한 설정
max_workers = 5  # 동시에 실행할 스레드 수

print("파일 다운로드를 시작합니다...")

# 병렬 다운로드 수행 및 진행 상황 표시
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    # tqdm을 사용하여 진행 상황 표시
    futures = {executor.submit(download_file, file, base_url, headers, output_dir): file for file in files}
    for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures), desc="다운로드 진행 중"):
        result = future.result()
        print(result)

print("모든 파일의 다운로드가 완료되었습니다.")

 

 

자주 쓰는 건 아니다 보니 가끔씩 다운로드 할 때마다 확인하고 사용하려고 아카이빙 해둔다. 혹시나 필요하신 분들에게 도움이 되길!