FinePDFs: 허깅페이스 혁신적 PDF 데이터셋 완전 가이드
⏱️ 예상 읽기 시간: 12분
서론
FinePDFs는 대규모 문서 처리 분야의 혁신적 발전을 나타내며, 총 67.9GB에 달하는 470만 개의 추출된 PDF 문서로 구성된 고품질 텍스트 데이터를 제공합니다. 이 포괄적인 튜토리얼은 이 혁신적인 데이터셋 작업의 모든 측면을 안내합니다.
FinePDFs의 특별한 점
FinePDFs는 가장 포괄적인 PDF 기반 데이터셋 중 하나로, 다음과 같은 특징을 보입니다:
- 대규모 스케일: 1,808개 언어에 걸친 470만 문서
- 고급 추출: 최첨단 docling 및 RolmOCR 기술 활용
- 도메인 풍부: 과학, 법률, 기술 콘텐츠의 광범위한 커버리지
- 오픈 라이선스: 상업적 및 연구 용도를 위한 ODC-By 라이선스
- 품질 중심: 진정한 콘텐츠 다양성 보존을 위한 최소 필터링
사전 요구사항 및 설정
환경 요구사항
# Python 3.8+ 권장
python --version
# 필수 패키지
pip install datasets transformers torch huggingface_hub
pip install pandas numpy matplotlib seaborn
pip install tqdm rich
하드웨어 권장사항
- RAM: 최소 16GB, 권장 32GB+
- 저장공간: 전체 데이터셋을 위한 100GB+ 여유 공간
- 네트워크: 초기 다운로드를 위한 안정적인 인터넷
- GPU: 선택사항이지만 처리를 위해 권장
FinePDFs 시작하기
기본 데이터셋 로딩
from datasets import load_dataset
import pandas as pd
from collections import Counter
import matplotlib.pyplot as plt
# 데이터셋 로드 (영어 서브셋부터 시작)
print("FinePDFs 데이터셋 로딩 중...")
dataset = load_dataset("HuggingFaceFW/finepdfs", "eng_Latn")
print(f"데이터셋 크기: {len(dataset['train'])} 문서")
print(f"사용 가능한 컬럼: {dataset['train'].column_names}")
# 처음 몇 개 샘플 검사
for i, sample in enumerate(dataset['train'].take(3)):
print(f"\n--- 샘플 {i+1} ---")
print(f"언어: {sample['language']}")
print(f"텍스트 길이: {len(sample['text'])} 문자")
print(f"텍스트 미리보기: {sample['text'][:200]}...")
사용 가능한 언어 탐색
# 언어 메타데이터 로드
all_configs = ["eng_Latn", "arb_Arab", "spa_Latn", "fra_Latn", "deu_Latn",
"ita_Latn", "por_Latn", "rus_Cyrl", "jpn_Jpan", "kor_Hang"]
language_stats = {}
for config in all_configs[:5]: # 처음 5개 언어 샘플
try:
ds = load_dataset("HuggingFaceFW/finepdfs", config, split="train")
language_stats[config] = len(ds)
print(f"{config}: {len(ds):,} 문서")
except Exception as e:
print(f"{config} 로드 실패: {e}")
# 언어 분포 시각화
plt.figure(figsize=(12, 6))
languages = list(language_stats.keys())
counts = list(language_stats.values())
plt.bar(languages, counts)
plt.title("언어별 문서 수")
plt.xlabel("언어 코드")
plt.ylabel("문서 수")
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig("finepdfs_language_distribution.png", dpi=300, bbox_inches='tight')
plt.show()
고급 데이터셋 작업
텍스트 품질 분석
import re
from textstat import flesch_reading_ease, flesch_kincaid_grade
def analyze_text_quality(sample_texts, sample_size=1000):
"""데이터셋 샘플의 텍스트 품질 메트릭 분석"""
metrics = {
'avg_length': [],
'word_count': [],
'sentence_count': [],
'reading_ease': [],
'grade_level': [],
'special_chars_ratio': []
}
for text in sample_texts[:sample_size]:
# 기본 메트릭
text_length = len(text)
word_count = len(text.split())
sentence_count = len(re.split(r'[.!?]+', text))
# 읽기 메트릭 (오류 처리)
try:
reading_ease = flesch_reading_ease(text)
grade_level = flesch_kincaid_grade(text)
except:
reading_ease = grade_level = 0
# 특수 문자 비율
special_chars = len(re.findall(r'[^\w\s]', text))
special_ratio = special_chars / max(text_length, 1)
# 메트릭 저장
metrics['avg_length'].append(text_length)
metrics['word_count'].append(word_count)
metrics['sentence_count'].append(sentence_count)
metrics['reading_ease'].append(reading_ease)
metrics['grade_level'].append(grade_level)
metrics['special_chars_ratio'].append(special_ratio)
return metrics
# 데이터셋 품질 분석
print("텍스트 품질 분석 중...")
sample_texts = [item['text'] for item in dataset['train'].take(1000)]
quality_metrics = analyze_text_quality(sample_texts)
# 결과 표시
for metric, values in quality_metrics.items():
avg_value = sum(values) / len(values)
print(f"{metric}: {avg_value:.2f}")
도메인별 필터링
def filter_by_domain(dataset, domain_keywords, min_matches=2):
"""도메인별 키워드로 데이터셋 필터링"""
def contains_domain_keywords(example):
text_lower = example['text'].lower()
matches = sum(1 for keyword in domain_keywords if keyword in text_lower)
return matches >= min_matches
filtered_dataset = dataset.filter(contains_domain_keywords)
return filtered_dataset
# 예시: 과학 콘텐츠 필터링
scientific_keywords = [
'연구', '조사', '분석', '실험', '가설',
'방법론', '결과', '결론', '동료 검토',
'저널', '출판', '인용', '초록'
]
print("과학 콘텐츠 필터링 중...")
scientific_subset = filter_by_domain(dataset['train'], scientific_keywords)
print(f"과학 문서: {len(scientific_subset)} / {len(dataset['train'])}")
# 예시: 법률 콘텐츠 필터링
legal_keywords = [
'법원', '판사', '법', '법적', '법령', '규정',
'계약', '협정', '원고', '피고',
'관할권', '소송', '변호사'
]
legal_subset = filter_by_domain(dataset['train'], legal_keywords)
print(f"법률 문서: {len(legal_subset)} / {len(dataset['train'])}")
실용적 응용 사례
1. 언어 모델 훈련
from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer
import torch
def prepare_training_data(dataset, tokenizer, max_length=512):
"""언어 모델 훈련을 위한 데이터셋 준비"""
def tokenize_function(examples):
# 텍스트 토큰화
tokenized = tokenizer(
examples['text'],
truncation=True,
padding=True,
max_length=max_length,
return_tensors="pt"
)
# Causal LM을 위한 라벨 설정
tokenized["labels"] = tokenized["input_ids"].clone()
return tokenized
tokenized_dataset = dataset.map(
tokenize_function,
batched=True,
remove_columns=dataset.column_names
)
return tokenized_dataset
# 토크나이저와 모델 초기화
model_name = "distilgpt2" # 데모용 경량 모델
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(model_name)
# 훈련 데이터 준비 (데모용 작은 서브셋 사용)
print("훈련 데이터 준비 중...")
train_subset = dataset['train'].select(range(1000)) # 작은 샘플
tokenized_data = prepare_training_data(train_subset, tokenizer)
# 훈련 구성
training_args = TrainingArguments(
output_dir="./finepdfs-model",
num_train_epochs=1,
per_device_train_batch_size=4,
gradient_accumulation_steps=2,
warmup_steps=100,
logging_steps=50,
save_steps=500,
evaluation_strategy="no",
save_total_limit=2,
load_best_model_at_end=False,
)
# 트레이너 초기화
trainer = Trainer(
model=model,
args=training_args,
train_dataset=tokenized_data,
tokenizer=tokenizer,
)
print("훈련 시작... (시간이 오래 걸릴 수 있습니다)")
# trainer.train() # 실제 훈련하려면 주석 해제
print("훈련 구성 준비 완료!")
2. 문서 유사도 검색
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
def create_document_embeddings(texts, model_name="all-MiniLM-L6-v2"):
"""문서 유사도 검색을 위한 임베딩 생성"""
model = SentenceTransformer(model_name)
# 메모리 관리를 위해 배치 처리
batch_size = 32
embeddings = []
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
batch_embeddings = model.encode(batch_texts, show_progress_bar=True)
embeddings.extend(batch_embeddings)
return np.array(embeddings), model
def find_similar_documents(query, embeddings, texts, model, top_k=5):
"""쿼리와 가장 유사한 문서 찾기"""
# 쿼리 인코딩
query_embedding = model.encode([query])
# 유사도 계산
similarities = cosine_similarity(query_embedding, embeddings)[0]
# 상위 k개 인덱스 가져오기
top_indices = np.argsort(similarities)[-top_k:][::-1]
results = []
for idx in top_indices:
results.append({
'index': idx,
'similarity': similarities[idx],
'text': texts[idx][:200] + "..."
})
return results
# 사용 예시
print("문서 임베딩 생성 중...")
sample_texts = [item['text'] for item in dataset['train'].take(100)]
embeddings, model = create_document_embeddings(sample_texts)
# 유사한 문서 검색
query = "머신러닝 알고리즘과 인공지능"
similar_docs = find_similar_documents(query, embeddings, sample_texts, model)
print(f"\n'{query}'에 대한 유사 문서:")
for i, doc in enumerate(similar_docs):
print(f"\n{i+1}. 유사도: {doc['similarity']:.3f}")
print(f"텍스트: {doc['text']}")
3. 다국어 분석
def analyze_multilingual_content(language_configs, sample_size=500):
"""여러 언어에 걸친 콘텐츠 분석"""
results = {}
for lang_config in language_configs:
try:
print(f"{lang_config} 로딩 중...")
ds = load_dataset("HuggingFaceFW/finepdfs", lang_config, split="train")
# 문서 샘플링
sample = ds.take(min(sample_size, len(ds)))
texts = [item['text'] for item in sample]
# 통계 계산
avg_length = sum(len(text) for text in texts) / len(texts)
avg_words = sum(len(text.split()) for text in texts) / len(texts)
results[lang_config] = {
'total_docs': len(ds),
'avg_length': avg_length,
'avg_words': avg_words,
'sample_text': texts[0][:100] + "..." if texts else ""
}
except Exception as e:
print(f"{lang_config} 로딩 오류: {e}")
results[lang_config] = {'error': str(e)}
return results
# 여러 언어 분석
multilingual_configs = ["eng_Latn", "spa_Latn", "fra_Latn", "deu_Latn"]
multilingual_analysis = analyze_multilingual_content(multilingual_configs)
# 결과 표시
for lang, stats in multilingual_analysis.items():
print(f"\n--- {lang} ---")
if 'error' in stats:
print(f"오류: {stats['error']}")
else:
print(f"총 문서: {stats['total_docs']:,}")
print(f"평균 길이: {stats['avg_length']:.0f} 문자")
print(f"평균 단어: {stats['avg_words']:.0f}")
print(f"샘플: {stats['sample_text']}")
성능 최적화
메모리 관리
import gc
from datasets import Dataset
def process_large_dataset_in_chunks(dataset, chunk_size=1000, process_func=None):
"""메모리 효율적 청크 단위 대용량 데이터셋 처리"""
total_size = len(dataset)
results = []
for i in range(0, total_size, chunk_size):
print(f"청크 {i//chunk_size + 1}/{(total_size-1)//chunk_size + 1} 처리 중")
# 청크 가져오기
chunk = dataset.select(range(i, min(i + chunk_size, total_size)))
# 청크 처리
if process_func:
chunk_result = process_func(chunk)
results.append(chunk_result)
# 가비지 컬렉션 강제 실행
gc.collect()
return results
def chunk_text_analysis(chunk):
"""청크용 예시 처리 함수"""
texts = [item['text'] for item in chunk]
return {
'chunk_size': len(texts),
'avg_length': sum(len(text) for text in texts) / len(texts),
'total_chars': sum(len(text) for text in texts)
}
# 예시: 메모리 효율적 청크 단위 데이터셋 처리
print("메모리 효율적 청크로 데이터셋 처리 중...")
chunk_results = process_large_dataset_in_chunks(
dataset['train'].select(range(5000)), # 데모용 서브셋 사용
chunk_size=1000,
process_func=chunk_text_analysis
)
for i, result in enumerate(chunk_results):
print(f"청크 {i+1}: {result}")
대용량 데이터셋 스트리밍
from datasets import load_dataset
def stream_process_dataset(dataset_name, config, process_func, max_items=None):
"""메모리에 로드하지 않고 스트림 방식으로 데이터셋 처리"""
# 스트리밍 데이터셋으로 로드
dataset = load_dataset(dataset_name, config, streaming=True, split="train")
processed_count = 0
results = []
for item in dataset:
if max_items and processed_count >= max_items:
break
result = process_func(item)
results.append(result)
processed_count += 1
if processed_count % 100 == 0:
print(f"{processed_count}개 항목 처리됨...")
return results
def extract_keywords(item):
"""예시 처리 함수"""
text = item['text'].lower()
# 간단한 키워드 추출 (더 정교한 방법으로 대체 가능)
keywords = []
for word in ['ai', '머신러닝', '신경망', '알고리즘']:
if word in text:
keywords.append(word)
return {'keywords': keywords, 'text_length': len(text)}
# 예시: 스트림 처리
print("스트림 방식으로 데이터셋 처리 중...")
stream_results = stream_process_dataset(
"HuggingFaceFW/finepdfs",
"eng_Latn",
extract_keywords,
max_items=1000
)
print(f"스트리밍으로 {len(stream_results)}개 항목 처리됨")
고급 사용 사례
1. 도메인 특화 말뭉치 구축
def build_domain_corpus(dataset, domain_rules, output_path=None):
"""도메인 규칙을 기반으로 전문 말뭉치 구축"""
def evaluate_domain_match(text, rules):
score = 0
text_lower = text.lower()
# 필수 키워드 (최소 하나는 있어야 함)
required = rules.get('required', [])
if required and any(keyword in text_lower for keyword in required):
score += 2
elif required: # 필수 키워드가 있지만 없음
return 0
# 보너스 키워드
bonus = rules.get('bonus', [])
for keyword in bonus:
if keyword in text_lower:
score += 1
# 페널티 키워드
penalty = rules.get('penalty', [])
for keyword in penalty:
if keyword in text_lower:
score -= 2
return max(0, score)
# 도메인 규칙 정의
scientific_rules = {
'required': ['연구', '조사', '분석', '실험'],
'bonus': ['동료 검토', '방법론', '가설', '통계적'],
'penalty': ['광고', '홍보', '지금 구매']
}
# 문서 점수 매기기 및 필터링
scored_docs = []
for item in dataset:
score = evaluate_domain_match(item['text'], scientific_rules)
if score >= 3: # 최소 점수 임계값
scored_docs.append({
'text': item['text'],
'language': item.get('language', 'unknown'),
'domain_score': score
})
print(f"도메인 말뭉치 구축 완료: {len(scored_docs)}개 문서")
if output_path:
# 말뭉치 저장
import json
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(scored_docs, f, ensure_ascii=False, indent=2)
print(f"말뭉치가 {output_path}에 저장됨")
return scored_docs
# 과학 말뭉치 구축
scientific_corpus = build_domain_corpus(
dataset['train'].take(2000), # 데모용 서브셋 사용
{}, # 규칙은 함수 내에서 정의됨
"scientific_corpus.json"
)
2. 품질 평가 파이프라인
def assess_document_quality(text):
"""문서의 포괄적 품질 평가"""
# 기본 메트릭
char_count = len(text)
word_count = len(text.split())
sentence_count = len(re.split(r'[.!?]+', text))
# 품질 지표
quality_score = 0
issues = []
# 길이 검사
if word_count < 50:
issues.append("너무 짧음")
quality_score -= 2
elif word_count > 100:
quality_score += 1
# 일관성 검사
avg_word_length = sum(len(word) for word in text.split()) / max(word_count, 1)
if avg_word_length > 6:
issues.append("지나치게 복잡한 어휘")
quality_score -= 1
# 콘텐츠 다양성
unique_words = len(set(text.lower().split()))
diversity_ratio = unique_words / max(word_count, 1)
if diversity_ratio > 0.7:
quality_score += 2
elif diversity_ratio < 0.3:
issues.append("낮은 어휘 다양성")
quality_score -= 1
# 특수 문자 노이즈
special_char_ratio = len(re.findall(r'[^\w\s]', text)) / max(char_count, 1)
if special_char_ratio > 0.3:
issues.append("높은 특수 문자 노이즈")
quality_score -= 2
return {
'quality_score': quality_score,
'issues': issues,
'metrics': {
'char_count': char_count,
'word_count': word_count,
'sentence_count': sentence_count,
'diversity_ratio': diversity_ratio,
'special_char_ratio': special_char_ratio
}
}
# 데이터셋 샘플에서 품질 평가
print("문서 품질 평가 중...")
quality_results = []
for item in dataset['train'].take(500):
quality = assess_document_quality(item['text'])
quality_results.append(quality)
# 품질 분포 분석
quality_scores = [result['quality_score'] for result in quality_results]
avg_quality = sum(quality_scores) / len(quality_scores)
high_quality_count = sum(1 for score in quality_scores if score >= 2)
print(f"평균 품질 점수: {avg_quality:.2f}")
print(f"고품질 문서: {high_quality_count}/{len(quality_results)}")
print(f"품질 분포: {Counter(quality_scores)}")
테스트 및 검증
포괄적 테스트 스크립트
#!/usr/bin/env python3
"""
FinePDFs 데이터셋 테스트 스크립트
데이터셋 로딩, 처리, 분석 기능 검증
"""
import sys
import time
import traceback
from datasets import load_dataset
import pandas as pd
import numpy as np
def test_basic_loading():
"""기본 데이터셋 로딩 기능 테스트"""
print("기본 데이터셋 로딩 테스트 중...")
try:
dataset = load_dataset("HuggingFaceFW/finepdfs", "eng_Latn", split="train")
print(f"✓ {len(dataset)}개 문서로 데이터셋 로딩 성공")
# 데이터 접근 테스트
first_item = dataset[0]
required_fields = ['text', 'language']
for field in required_fields:
if field not in first_item:
raise ValueError(f"필수 필드 누락: {field}")
print("✓ 데이터셋 구조 검증 통과")
return True
except Exception as e:
print(f"✗ 기본 로딩 테스트 실패: {e}")
return False
def test_processing_functions():
"""데이터 처리 및 분석 함수 테스트"""
print("\n처리 함수 테스트 중...")
try:
# 작은 샘플 로드
dataset = load_dataset("HuggingFaceFW/finepdfs", "eng_Latn", split="train")
sample = dataset.take(10)
# 텍스트 분석 테스트
texts = [item['text'] for item in sample]
avg_length = sum(len(text) for text in texts) / len(texts)
if avg_length == 0:
raise ValueError("유효하지 않은 텍스트 분석 결과")
print(f"✓ 텍스트 분석 성공 (평균 길이: {avg_length:.0f})")
# 필터링 테스트
filtered = [text for text in texts if len(text) > 100]
print(f"✓ 필터링 성공 ({len(filtered)}/{len(texts)} 문서)")
return True
except Exception as e:
print(f"✗ 처리 함수 테스트 실패: {e}")
return False
def test_memory_efficiency():
"""메모리 효율성 테스트"""
print("\n메모리 효율성 테스트 중...")
try:
# 스트림 처리 테스트
dataset = load_dataset("HuggingFaceFW/finepdfs", "eng_Latn",
streaming=True, split="train")
count = 0
for item in dataset:
count += 1
if count >= 5: # 처음 5개 항목 테스트
break
if count != 5:
raise ValueError("스트림 처리 실패")
print("✓ 스트림 처리 성공")
return True
except Exception as e:
print(f"✗ 메모리 효율성 테스트 실패: {e}")
return False
def run_all_tests():
"""포괄적 테스트 스위트 실행"""
print("=" * 50)
print("FinePDFs 데이터셋 테스트 스위트")
print("=" * 50)
tests = [
test_basic_loading,
test_processing_functions,
test_memory_efficiency
]
results = []
start_time = time.time()
for test in tests:
try:
result = test()
results.append(result)
except Exception as e:
print(f"✗ 테스트 {test.__name__} 크래시: {e}")
traceback.print_exc()
results.append(False)
# 요약
print("\n" + "=" * 50)
print("테스트 요약")
print("=" * 50)
passed = sum(results)
total = len(results)
print(f"통과한 테스트: {passed}/{total}")
print(f"성공률: {passed/total*100:.1f}%")
print(f"총 시간: {time.time() - start_time:.1f}초")
if passed == total:
print("🎉 모든 테스트 통과!")
return True
else:
print("⚠️ 일부 테스트 실패. 위 출력을 확인하세요.")
return False
if __name__ == "__main__":
success = run_all_tests()
sys.exit(0 if success else 1)
베스트 프랙티스 및 팁
1. 효율적 데이터 로딩
# 베스트 프랙티스: 적절한 서브셋 사용
def load_dataset_efficiently(language_code, max_size=None):
"""최적 설정으로 데이터셋 로드"""
try:
if max_size and max_size < 10000:
# 작은 데이터셋: 완전 로드
dataset = load_dataset("HuggingFaceFW/finepdfs", language_code,
split=f"train[:{max_size}]")
else:
# 큰 데이터셋: 스트리밍 사용
dataset = load_dataset("HuggingFaceFW/finepdfs", language_code,
streaming=True, split="train")
return dataset
except Exception as e:
print(f"{language_code} 로드 실패: {e}")
return None
# 사용 예시
efficient_dataset = load_dataset_efficiently("eng_Latn", max_size=5000)
2. 오류 처리 및 강건성
def robust_text_processing(text, max_retries=3):
"""오류 처리 및 재시도 기능이 있는 텍스트 처리"""
for attempt in range(max_retries):
try:
# 입력 검증
if not isinstance(text, str):
raise TypeError("입력은 문자열이어야 합니다")
if len(text) == 0:
return {"status": "empty", "result": None}
# 텍스트 처리
processed = {
"length": len(text),
"word_count": len(text.split()),
"status": "success"
}
return processed
except Exception as e:
print(f"시도 {attempt + 1} 실패: {e}")
if attempt == max_retries - 1:
return {"status": "failed", "error": str(e)}
time.sleep(0.1) # 재시도 전 짧은 지연
# 오류 처리를 포함한 사용 예시
def process_dataset_safely(dataset, max_items=1000):
"""포괄적 오류 처리로 데이터셋 처리"""
successful = 0
failed = 0
results = []
for i, item in enumerate(dataset.take(max_items)):
try:
result = robust_text_processing(item['text'])
if result['status'] == 'success':
successful += 1
results.append(result)
else:
failed += 1
except Exception as e:
print(f"항목 {i} 처리 실패: {e}")
failed += 1
print(f"처리 완료: {successful}개 성공, {failed}개 실패")
return results
결론
FinePDFs는 문서 기반 머신러닝의 패러다임 전환을 나타내며, 고품질의 다양한 텍스트 콘텐츠에 대한 전례 없는 접근성을 제공합니다. 이 튜토리얼에서 다음 내용을 다뤘습니다:
핵심 요점
- 데이터셋 숙련: FinePDFs의 구조와 기능 이해
- 효율적 처리: 대규모 작업을 위한 메모리 인식 기법
- 실용적 응용: 훈련부터 분석까지의 실제 사용 사례
- 품질 평가: 콘텐츠 평가 및 필터링 방법
- 다국어 지원: 데이터셋의 언어적 다양성 활용
다음 단계
- 실험: 다양한 언어 구성과 도메인 필터를 시도해보세요
- 확장: 인프라가 허용하는 범위에서 더 큰 서브셋에 기법을 적용하세요
- 통합: FinePDFs를 기존 ML 파이프라인에 통합하세요
- 기여: 개선사항과 발견사항을 커뮤니티와 공유하세요
참고 자료
FinePDFs와 함께 문서 기반 AI의 미래가 여기에 있습니다. 오늘부터 구축을 시작하세요! 🚀
질문이나 피드백이 있으신가요? Hugging Face 커뮤니티 포럼에서 토론에 참여하거나 데이터셋의 지속적인 개발에 기여해보세요.