2.1 데이터 수집이란?
데이터 수집(Data Ingestion)은 다양한 소스에서 데이터를 추출하여 파이프라인으로 가져오는 첫 번째 단계입니다. 이 단계의 품질이 전체 파이프라인의 품질을 결정합니다.
Garbage In, Garbage Out
잘못된 데이터를 수집하면 아무리 좋은 변환을 해도 결과는 쓸모없습니다. 수집 단계에서 데이터 품질을 검증하는 것이 핵심입니다.
2.2 수집 패턴
Full Load (전체 적재)
개념: 매번 전체 데이터를 다시 가져옴
Day 1: 1만 건 로드
Day 2: 1.2만 건 로드 (전체)
Day 3: 1.5만 건 로드 (전체)
장점:
- 구현 간단
- 데이터 일관성 보장
- 삭제된 레코드도 반영
단점:
- 비효율적 (같은 데이터 반복 전송)
- 소스 시스템에 부하
- 처리 시간 증가
적합한 경우:
- 데이터가 작을 때 (수천 건)
- 삭제 추적이 중요할 때
- 단순함이 우선일 때
Incremental Load (증분 적재)
개념: 변경된 데이터만 가져옴
Day 1: 1만 건 로드
Day 2: 2천 건 로드 (신규/변경분만)
Day 3: 3천 건 로드 (신규/변경분만)
장점:
- 효율적 (변경분만 전송)
- 빠른 처리
- 소스 시스템 부하 감소
단점:
- 구현 복잡
- 변경 감지 메커니즘 필요
- 삭제 추적 어려움
변경 감지 방법:
- Timestamp: updated_at 컬럼 사용
- Version: version 번호 증가
- CDC: 데이터베이스 로그 읽기
Change Data Capture (CDC)
데이터베이스의 변경 로그를 읽어서 실시간으로 변경사항 추적:
MySQL Binlog:
INSERT INTO users VALUES (123, 'Alice')
UPDATE users SET email='alice@new.com' WHERE id=123
DELETE FROM users WHERE id=456
장점:
- 실시간 변경 감지
- Insert/Update/Delete 모두 추적
- 소스 시스템 영향 최소
도구:
- Debezium (오픈소스)
- Maxwell
- Airbyte
- Fivetran
2.3 데이터 소스별 수집 방법
관계형 데이터베이스
-- PostgreSQL에서 증분 로드
SELECT * FROM orders
WHERE updated_at > '2025-12-25 00:00:00'
ORDER BY updated_at
LIMIT 10000;
주의사항:
- 인덱스 확인 (updated_at에 인덱스 필요)
- 타임존 일관성
- 페이지네이션 (메모리 부족 방지)
- 커넥션 풀 관리
REST API
import requests
import time
def fetch_api_data(endpoint, since_timestamp):
page = 1
all_data = []
while True:
response = requests.get(
f"{endpoint}?since={since_timestamp}&page={page}",
headers={"Authorization": f"Bearer {token}"}
)
if response.status_code == 429: # Rate limit
time.sleep(60)
continue
data = response.json()
if not data:
break
all_data.extend(data)
page += 1
time.sleep(0.1) # 예의있게
return all_data
주의사항:
- Rate limiting (속도 제한)
- 재시도 로직 (네트워크 오류)
- 페이지네이션
- 인증 토큰 갱신
Kafka 스트림
from kafka import KafkaConsumer
consumer = KafkaConsumer(
'user-events',
bootstrap_servers=['localhost:9092'],
group_id='pipeline-consumer',
auto_offset_reset='earliest',
enable_auto_commit=False
)
for message in consumer:
process_event(message.value)
consumer.commit() # 처리 성공 후 커밋
주의사항:
- Consumer Group 관리
- 오프셋 관리 (정확히 한 번 처리)
- Backpressure 처리
- 파티션 리밸런싱
CSV/JSON 파일
import pandas as pd
from pathlib import Path
def process_files(directory):
for file_path in Path(directory).glob('*.csv'):
# 청크 단위로 읽기 (메모리 효율)
for chunk in pd.read_csv(file_path, chunksize=10000):
validate_chunk(chunk)
transform_chunk(chunk)
load_chunk(chunk)
# 처리 완료 후 아카이브
archive_file(file_path)
2.4 데이터 검증
수집 즉시 검증하여 문제를 조기에 발견:
스키마 검증
from pydantic import BaseModel, validator
class UserEvent(BaseModel):
user_id: int
event_type: str
timestamp: datetime
amount: float
@validator('amount')
def amount_must_be_positive(cls, v):
if v < 0:
raise ValueError('amount must be positive')
return v
비즈니스 룰 검증
def validate_order(order):
assert order['total'] > 0, "Total must be positive"
assert order['items'], "Order must have items"
assert order['user_id'] is not None, "User ID required"
assert order['created_at'] <= datetime.now(), "Future orders not allowed"
2.5 에러 핸들링
재시도 전략
import time
from functools import wraps
def retry_with_backoff(max_retries=3, backoff_factor=2):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
retries += 1
if retries >= max_retries:
raise
wait_time = backoff_factor ** retries
print(f"Retry {retries} after {wait_time}s: {e}")
time.sleep(wait_time)
return wrapper
return decorator
@retry_with_backoff(max_retries=5)
def fetch_data_from_api(url):
response = requests.get(url)
response.raise_for_status()
return response.json()
Dead Letter Queue
처리 실패한 데이터를 별도 저장:
def process_message(message):
try:
validate(message)
transform(message)
load(message)
except ValidationError as e:
# 검증 실패 → DLQ
send_to_dlq(message, error=str(e))
except Exception as e:
# 기타 오류 → 재시도 큐
send_to_retry_queue(message)
2.6 성능 최적화
병렬 처리
from concurrent.futures import ThreadPoolExecutor
def fetch_all_tables(tables):
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {
executor.submit(fetch_table, table): table
for table in tables
}
for future in futures:
table = futures[future]
try:
data = future.result()
print(f"Fetched {len(data)} rows from {table}")
except Exception as e:
print(f"Failed to fetch {table}: {e}")
배치 처리
def fetch_in_batches(query, batch_size=10000):
offset = 0
while True:
batch = execute_query(
query + f" LIMIT {batch_size} OFFSET {offset}"
)
if not batch:
break
yield batch
offset += batch_size
# 사용
for batch in fetch_in_batches("SELECT * FROM large_table"):
process_batch(batch)
2.7 실전 예제: API to S3 Pipeline
import boto3
import requests
from datetime import datetime, timedelta
class APItoS3Pipeline:
def __init__(self, api_url, s3_bucket):
self.api_url = api_url
self.s3_bucket = s3_bucket
self.s3_client = boto3.client('s3')
def extract(self, since_date):
"""API에서 데이터 추출"""
response = requests.get(
self.api_url,
params={'since': since_date.isoformat()}
)
return response.json()
def load(self, data, key):
"""S3에 적재"""
self.s3_client.put_object(
Bucket=self.s3_bucket,
Key=key,
Body=json.dumps(data)
)
def run(self):
"""파이프라인 실행"""
# 어제 데이터 가져오기
since = datetime.now() - timedelta(days=1)
# Extract
print(f"Extracting data since {since}")
data = self.extract(since)
print(f"Extracted {len(data)} records")
# Load
key = f"raw/date={since.date()}/data.json"
print(f"Loading to s3://{self.s3_bucket}/{key}")
self.load(data, key)
print("Complete!")
# 실행
pipeline = APItoS3Pipeline(
api_url="https://api.example.com/events",
s3_bucket="my-data-lake"
)
pipeline.run()
2.8 모니터링 메트릭
수집 단계에서 추적해야 할 주요 메트릭:
- Records extracted: 추출된 레코드 수
- Extraction duration: 추출 소요 시간
- Error rate: 실패율
- Data freshness: 데이터 신선도 (얼마나 최신인가)
- Source lag: 소스와의 지연 시간
- API rate limit remaining: API 할당량 잔여
핵심 원칙
- 빠른 실패: 문제를 조기에 발견
- 멱등성: 재실행해도 안전
- 관찰성: 모든 것을 로깅
- 복원력: 실패해도 복구
2.9 다음 장 미리보기
다음 장에서는 수집한 데이터를 변환하는 방법을 배웁니다:
- 데이터 정제 (Cleaning)
- 데이터 정규화 (Normalization)
- 데이터 보강 (Enrichment)
- 데이터 집계 (Aggregation)
- SQL vs 코드 기반 변환