본문으로 건너뛰기

데이터 입수 실무 고려사항

실무에서 데이터를 입수할 때는 기술적인 측면뿐만 아니라 법적, 윤리적, 비즈니스적 측면도 종합적으로 고려해야 한다. 이번 섹션에서는 실제 프로젝트에서 마주치게 되는 다양한 고려사항들을 살펴보자.

데이터 품질 관리​

수집 데이터 검증 체크리스트​

def comprehensive_data_quality_check(df, source_info):
    """포괄적인 데이터 품질 검사"""
    
    checklist = {
        'basic_info': {
            'source': source_info.get('name', 'Unknown'),
            'collection_date': pd.Timestamp.now(),
            'rows': len(df),
            'columns': len(df.columns),
            'size_mb': df.memory_usage(deep=True).sum() / 1024 / 1024
        },
        'completeness': {},
        'accuracy': {},
        'consistency': {},
        'timeliness': {},
        'validity': {}
    }
    
    # 완성도 검사
    for col in df.columns:
        missing_count = df[col].isnull().sum()
        missing_rate = missing_count / len(df) * 100
        checklist['completeness'][col] = {
            'missing_count': missing_count,
            'missing_rate': round(missing_rate, 2),
            'status': 'GOOD' if missing_rate < 5 else 'WARNING' if missing_rate < 20 else 'CRITICAL'
        }
    
    # 유효성 검사
    for col in df.columns:
        if df[col].dtype == 'object':
            # 문자열 길이 검사
            max_length = df[col].str.len().max()
            avg_length = df[col].str.len().mean()
            checklist['validity'][col] = {
                'max_length': max_length,
                'avg_length': round(avg_length, 2) if pd.notna(avg_length) else 0,
                'unique_values': df[col].nunique()
            }
        elif pd.api.types.is_numeric_dtype(df[col]):
            # 숫자 범위 검사
            checklist['validity'][col] = {
                'min_value': df[col].min(),
                'max_value': df[col].max(),
                'mean_value': df[col].mean(),
                'std_value': df[col].std()
            }
    
    return checklist

# 사용 예시
# quality_report = comprehensive_data_quality_check(df, {'name': 'Sales Data'})

결측치와 이상치 초기 탐지​

def detect_data_anomalies(df):
    """데이터 이상 현상 탐지"""
    
    anomalies = {
        'missing_patterns': {},
        'outliers': {},
        'duplicates': {},
        'format_issues': {}
    }
    
    # 결측치 패턴 분석
    missing_matrix = df.isnull()
    
    # 모든 컬럼이 결측인 행
    all_missing_rows = missing_matrix.all(axis=1).sum()
    anomalies['missing_patterns']['all_missing_rows'] = all_missing_rows
    
    # 특정 컬럼 조합의 결측 패턴
    for col in df.columns:
        if df[col].isnull().sum() > 0:
            pattern = missing_matrix.groupby(col).size()
            anomalies['missing_patterns'][f'{col}_pattern'] = pattern.to_dict()
    
    # 이상치 탐지 (수치형 컬럼)
    for col in df.select_dtypes(include=['number']).columns:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        outliers = df[(df[col] < lower_bound) | (df[col] > upper_bound)]
        anomalies['outliers'][col] = {
            'count': len(outliers),
            'percentage': len(outliers) / len(df) * 100,
            'lower_bound': lower_bound,
            'upper_bound': upper_bound
        }
    
    # 중복 데이터 탐지
    duplicate_rows = df.duplicated().sum()
    anomalies['duplicates']['total_duplicates'] = duplicate_rows
    
    # 각 컬럼별 중복값
    for col in df.columns:
        duplicate_values = df[col].duplicated().sum()
        anomalies['duplicates'][f'{col}_duplicates'] = duplicate_values
    
    return anomalies

법적/윤리적 고려사항​

개인정보보호법과 GDPR 준수​

def check_privacy_compliance(df, column_mapping):
    """개인정보 보호 규정 준수 확인"""
    
    compliance_report = {
        'personal_data_detected': [],
        'sensitive_data_detected': [],
        'recommendations': [],
        'risk_level': 'LOW'
    }
    
    # 개인정보 식별 패턴
    personal_patterns = {
        'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        'phone': r'\b\d{2,3}-\d{3,4}-\d{4}\b',
        'ssn': r'\b\d{6}-\d{7}\b',  # 주민등록번호 패턴
        'credit_card': r'\b\d{4}-\d{4}-\d{4}-\d{4}\b'
    }
    
    for col in df.columns:
        if df[col].dtype == 'object':
            col_sample = df[col].dropna().astype(str).head(100)
            
            for pattern_name, pattern in personal_patterns.items():
                if col_sample.str.contains(pattern, regex=True).any():
                    compliance_report['personal_data_detected'].append({
                        'column': col,
                        'type': pattern_name,
                        'sample_count': col_sample.str.contains(pattern, regex=True).sum()
                    })
    
    # 민감 정보 컬럼 확인
    sensitive_keywords = ['age', 'salary', 'income', 'health', 'medical', 'religion', 'political']
    
    for col in df.columns:
        col_lower = col.lower()
        for keyword in sensitive_keywords:
            if keyword in col_lower:
                compliance_report['sensitive_data_detected'].append({
                    'column': col,
                    'type': keyword
                })
    
    # 위험도 평가
    personal_count = len(compliance_report['personal_data_detected'])
    sensitive_count = len(compliance_report['sensitive_data_detected'])
    
    if personal_count > 0 or sensitive_count > 2:
        compliance_report['risk_level'] = 'HIGH'
        compliance_report['recommendations'].append('개인정보 비식별화 처리 필요')
        compliance_report['recommendations'].append('데이터 접근 권한 제한 필요')
    elif sensitive_count > 0:
        compliance_report['risk_level'] = 'MEDIUM'
        compliance_report['recommendations'].append('민감 정보 처리 방침 수립 필요')
    
    return compliance_report

웹사이트 이용약관과 저작권​

import requests
from urllib.robotparser import RobotFileParser

def check_scraping_permission(url):
    """웹 스크래핑 허용 여부 확인"""
    
    permission_check = {
        'url': url,
        'robots_txt_check': False,
        'robots_txt_content': '',
        'scraping_allowed': False,
        'restrictions': [],
        'recommendations': []
    }
    
    try:
        # robots.txt 확인
        from urllib.parse import urljoin, urlparse
        
        parsed_url = urlparse(url)
        robots_url = f"{parsed_url.scheme}://{parsed_url.netloc}/robots.txt"
        
        rp = RobotFileParser()
        rp.set_url(robots_url)
        rp.read()
        
        permission_check['robots_txt_check'] = True
        
        # User-agent * 에 대한 허용 여부 확인
        can_fetch = rp.can_fetch('*', url)
        permission_check['scraping_allowed'] = can_fetch
        
        if not can_fetch:
            permission_check['restrictions'].append('robots.txt에서 접근 금지')
            permission_check['recommendations'].append('사이트 관리자에게 허가 요청')
        
        # Crawl-delay 확인
        crawl_delay = rp.crawl_delay('*')
        if crawl_delay:
            permission_check['restrictions'].append(f'크롤링 지연 시간: {crawl_delay}초')
            permission_check['recommendations'].append(f'요청 간격을 {crawl_delay}초 이상으로 설정')
        
    except Exception as e:
        permission_check['robots_txt_check'] = False
        permission_check['error'] = str(e)
        permission_check['recommendations'].append('robots.txt 확인 실패 - 수동으로 확인 필요')
    
    return permission_check

# 사용 예시
# permission = check_scraping_permission('https://example.com/data')
# print(permission)

기술적 최적화​

메모리 효율적인 데이터 로딩​

def memory_efficient_loading(file_path, target_columns=None, sample_rate=1.0):
    """메모리 효율적인 데이터 로딩"""
    
    # 파일 크기 확인
    file_size = os.path.getsize(file_path) / (1024 * 1024)  # MB
    print(f"파일 크기: {file_size:.2f} MB")
    
    # 큰 파일의 경우 청크 단위로 처리
    if file_size > 100:  # 100MB 이상
        return load_large_file_efficiently(file_path, target_columns, sample_rate)
    else:
        return load_small_file(file_path, target_columns, sample_rate)

def load_large_file_efficiently(file_path, target_columns, sample_rate):
    """대용량 파일 효율적 로딩"""
    
    chunk_size = 10000
    chunks = []
    
    # 첫 번째 청크로 컬럼 정보 확인
    first_chunk = pd.read_csv(file_path, nrows=chunk_size)
    
    if target_columns:
        available_columns = [col for col in target_columns if col in first_chunk.columns]
    else:
        available_columns = first_chunk.columns.tolist()
    
    # 샘플링과 함께 청크 단위 로딩
    for chunk in pd.read_csv(file_path, chunksize=chunk_size, usecols=available_columns):
        if sample_rate < 1.0:
            chunk = chunk.sample(frac=sample_rate)
        
        # 메모리 사용량 최적화
        chunk = optimize_memory_usage(chunk)
        chunks.append(chunk)
        
        # 메모리 사용량 모니터링
        if len(chunks) % 10 == 0:
            print(f"처리된 청크: {len(chunks)}")
    
    return pd.concat(chunks, ignore_index=True)

def optimize_memory_usage(df):
    """메모리 사용량 최적화"""
    
    for col in df.columns:
        col_type = df[col].dtype
        
        if col_type != 'object':
            c_min = df[col].min()
            c_max = df[col].max()
            
            if str(col_type)[:3] == 'int':
                if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:
                    df[col] = df[col].astype(np.int8)
                elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:
                    df[col] = df[col].astype(np.int16)
                elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:
                    df[col] = df[col].astype(np.int32)
            
            elif str(col_type)[:5] == 'float':
                if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max:
                    df[col] = df[col].astype(np.float32)
        
        else:
            # 문자열 컬럼의 경우 카테고리로 변환 (반복되는 값이 많을 때)
            if df[col].nunique() / len(df) < 0.5:  # 유니크 비율이 50% 미만
                df[col] = df[col].astype('category')
    
    return df

네트워크 오류와 재시도 로직​

import time
import random
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1, max_delay=60, backoff_factor=2):
    """지수 백오프를 사용한 재시도 데코레이터"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                
                except (requests.exceptions.ConnectionError, 
                       requests.exceptions.Timeout,
                       requests.exceptions.HTTPError) as e:
                    
                    last_exception = e
                    
                    if attempt == max_retries - 1:
                        break
                    
                    # 지수 백오프 계산
                    delay = min(base_delay * (backoff_factor ** attempt), max_delay)
                    # 지터 추가 (동시 요청 시 충돌 방지)
                    jitter = random.uniform(0, delay * 0.1)
                    total_delay = delay + jitter
                    
                    print(f"시도 {attempt + 1} 실패: {e}")
                    print(f"{total_delay:.2f}초 후 재시도...")
                    time.sleep(total_delay)
                
                except Exception as e:
                    # 재시도하지 않을 오류
                    print(f"재시도하지 않을 오류: {e}")
                    raise e
            
            # 모든 재시도 실패
            raise last_exception
        
        return wrapper
    return decorator

@retry_with_backoff(max_retries=5, base_delay=2)
def robust_api_request(url, params=None, headers=None):
    """강건한 API 요청"""
    
    response = requests.get(
        url, 
        params=params, 
        headers=headers, 
        timeout=30
    )
    response.raise_for_status()
    
    return response.json()

정리​

데이터 입수는 단순한 기술적 작업이 아니라 법적, 윤리적, 비즈니스적 고려사항이 복합적으로 얽힌 복잡한 과정이다. 실무에서는 데이터의 품질과 신뢰성을 확보하면서도 관련 규정을 준수하고, 기술적 최적화를 통해 효율성을 높이는 것이 중요하다.

특히 개인정보 보호와 저작권 문제는 법적 리스크와 직결되므로 반드시 사전에 검토해야 하며, 데이터 품질 관리는 후속 분석의 신뢰성을 좌우하는 핵심 요소이므로 체계적으로 접근해야 한다.

이번 장에서 다룬 다양한 데이터 입수 방법과 고려사항들을 종합적으로 활용하면, 실무에서 마주치는 대부분의 데이터 입수 상황에 효과적으로 대응할 수 있을 것이다.