[Python] CSV 파일 다루기: 데이터 불러오기와 저장

안녕하세요.데이터 처리하면서 CSV 파일 다루는 일이 생각보다 많은데, 파이썬으로 이걸 효율적으로 처리하는 방법이 있습니다. 오늘은 실수 없이 CSV 파일을 다루는 방법을 정리해봤습니다.

목차

1. CSV 파일 기본 이해하기 2. 파이썬 기본 csv 모듈 사용법 3. pandas로 CSV 다루기 4. CSV 처리 시 흔한 오류와 해결법 5. 대용량 CSV 처리 성능 최적화 6. 실전 CSV 처리 예제

1. CSV 파일 기본 이해하기

CSV(Comma-Separated Values)는 단순히 쉼표로 구분된 텍스트 파일입니다. 단순해 보이지만 이 단순함이 오히려 혼란을 가져옵니다. 파일 확장자만 보고 모든 CSV 파일이 동일하다고 착각하는 개발자가 많습니다. 실제로는 구분자, 인코딩, 이스케이프 문자 처리 방식이 모두 다릅니다.

CSV 파일은 표준이라는 게 모호합니다. RFC 4180이라는 명세가 있긴 하지만, 실제로 이걸 엄격히 따르는 경우는 드뭅니다. 엑셀에서 내보낸 CSV와 다른 시스템에서 내보낸 CSV는 다를 가능성이 높고, 맥과 윈도우 간에도 차이가 있습니다.

기본적인 CSV 파일 구조는 이렇습니다:

이름,나이,이메일
홍길동,30,hong@example.com
김철수,25,kim@example.com
이영희,28,lee@example.com

단순해 보이지만 데이터에 쉼표가 포함되어 있거나 줄바꿈이 있다면? 이미 골치 아파집니다. 대부분의 CSV 파일은 이런 특수 케이스를 따옴표로 처리합니다.

이름,설명,이메일
홍길동,"프로그래머, 데이터 분석가",hong@example.com
김철수,"개발 경력 2년,
웹 전문",kim@example.com

2. 파이썬 기본 csv 모듈 사용법

파이썬 표준 라이브러리에는 csv 모듈이 포함되어 있어서 별도 설치 없이 바로 사용할 수 있습니다. 기본적인 기능만 제공하지만 대부분의 간단한 CSV 처리 작업에는 충분합니다.

다음은 파이썬 기본 csv 모듈로 할 수 있는 주요 작업들입니다:

작업	코드 예시	주의사항
CSV 파일 읽기	csv.reader()	기본 구분자는 쉼표
CSV 파일 쓰기	csv.writer()	줄바꿈 문자 처리 주의
딕셔너리 형태로 읽기	csv.DictReader()	첫 행을 헤더로 사용
딕셔너리 형태로 쓰기	csv.DictWriter()	필드명 지정 필요
사용자 정의 구분자	delimiter 매개변수	탭('\t'), 파이프('|') 등

기본 csv 모듈 사용 예제입니다.

import csv

# CSV 파일 읽기
with open('data.csv', 'r', encoding='utf-8') as file:
    csv_reader = csv.reader(file)
    header = next(csv_reader)  # 헤더 행 처리
    for row in csv_reader:
        print(row)  # 각 행은 리스트 형태

# CSV 파일 쓰기
with open('output.csv', 'w', newline='', encoding='utf-8') as file:
    csv_writer = csv.writer(file)
    csv_writer.writerow(['이름', '나이', '이메일'])  # 헤더 쓰기
    csv_writer.writerow(['홍길동', 30, 'hong@example.com'])  # 데이터 행 쓰기

일반적인 실수 중 하나는 newline='' 매개변수를 생략하는 것입니다. 이걸 빼면 윈도우에서 빈 줄이 추가되는 문제가 발생합니다. 또한 인코딩을 명시하지 않으면 시스템 기본값을 사용하므로 시스템 간 이식성이 떨어집니다.

3. pandas로 CSV 다루기

파이썬에서 데이터 분석을 한다면 pandas를 쓰는 게 효율적입니다. CSV 파일 처리도 훨씬 간편하게 할 수 있습니다. 물론 메모리 사용량이 많아지는 단점이 있지만, 데이터 분석이나 조작이 필요하다면 기본 csv 모듈보다 훨씬 강력합니다.

pandas로 CSV 파일을 다루는 기본 방법은 다음과 같습니다:

1. read_csv() 함수로 CSV 파일을 DataFrame으로 읽어들입니다.
2. 필요한 데이터 처리 작업을 DataFrame 메서드로 수행합니다.
3. 처리된 데이터를 to_csv() 메서드로 다시 CSV 파일로 저장합니다.

pandas를 사용한 CSV 파일 처리 예제입니다:

import pandas as pd

# CSV 파일 읽기 (기본적인 방법)
df = pd.read_csv('data.csv')

# 간단한 데이터 탐색
print(df.head())        # 처음 5행 보기
print(df.describe())    # 통계 요약 정보
print(df.info())        # 데이터 타입 및 누락 값 정보

# 데이터 처리 예시
df['나이'] = df['나이'].fillna(df['나이'].mean())  # 누락된 나이 값을 평균으로 채우기
filtered_df = df[df['나이'] > 25]  # 25세 이상인 행만 필터링

# CSV 파일로 저장
filtered_df.to_csv('filtered_data.csv', index=False, encoding='utf-8')

pandas는 옵션이 매우 많습니다. 데이터를 읽을 때 타입을 자동으로 추론하지만, 항상 정확한 것은 아닙니다. 특히 날짜 형식이나 숫자로 표현된 카테고리 데이터는 명시적으로 타입을 지정해주는 것이 좋습니다.

📝 인코딩 문제 해결하기

CSV 파일 처리할 때 가장 흔한 문제는 인코딩 오류입니다. 한글이 들어간 CSV 파일을 읽을 때 UnicodeDecodeError가 발생한다면:

• 먼저 encoding='utf-8'을 시도해보세요.
• 여전히 문제가 있다면 encoding='cp949'(윈도우 한글), encoding='euc-kr'(구 한글 인코딩) 등을 시도해보세요.
• 정확한 인코딩을 모른다면 chardet 라이브러리로 인코딩을 추측할 수 있습니다.

4. CSV 처리 시 흔한 오류와 해결법

CSV 파일을 다루다 보면 예상치 못한 문제들이 발생합니다. 고객이 엑셀로 만든 CSV 파일은 특히 문제가 많습니다. 이상하게도 같은 파일인데 시스템마다 다르게 보이기도 하죠. 다음은 CSV 파일 처리 중 흔히 발생하는 문제들입니다.

먼저 인코딩 문제입니다. 이건 정말 고전적인 문제인데 2025년인 지금도 여전히 개발자들을 괴롭히고 있습니다. 특히 한글이 포함된 CSV 파일은 더 까다롭습니다.

# 인코딩 문제 해결을 위한 접근법
import chardet

# 파일의 인코딩 추측하기
with open('unknown_encoding.csv', 'rb') as file:
    result = chardet.detect(file.read())
    
print(f"감지된 인코딩: {result['encoding']}, 신뢰도: {result['confidence']}")

# 감지된 인코딩으로 파일 열기
detected_encoding = result['encoding']
df = pd.read_csv('unknown_encoding.csv', encoding=detected_encoding)

두 번째 문제는 구분자 문제입니다. 쉼표 대신 탭이나 세미콜론 등 다른 구분자를 사용하는 경우도 많습니다. 특히 유럽에서는 세미콜론(;)을 구분자로 사용하는 경우가 많습니다.

# 다양한 구분자 처리하기
# 탭으로 구분된 파일
df_tab = pd.read_csv('data.tsv', sep='\t', encoding='utf-8')

# 세미콜론으로 구분된 파일
df_semicolon = pd.read_csv('data.csv', sep=';', encoding='utf-8')

# 파이프로 구분된 파일
df_pipe = pd.read_csv('data.txt', sep='|', encoding='utf-8')

4. CSV 처리 시 흔한 오류와 해결법

세 번째 문제는 데이터 타입 처리입니다. CSV는 기본적으로 텍스트 파일이라 숫자, 날짜 등의 데이터 타입을 제대로 인식하지 못합니다. 특히 날짜 형식은 국가나 지역마다 다를 수 있어 더 복잡합니다.

# 데이터 타입 명시적 지정
df = pd.read_csv('data.csv', dtype={
    '나이': 'int', 
    '이름': 'string',
    '금액': 'float'
})

# 날짜 형식 처리
df = pd.read_csv('data.csv', parse_dates=['생일', '가입일'])

# 복잡한 날짜 형식 지정
df = pd.read_csv('data.csv', 
                 parse_dates=['이상한_날짜_형식'],
                 date_parser=lambda x: pd.to_datetime(x, format='%d/%m/%Y'))

네 번째 문제는 따옴표와 이스케이프 문자 처리입니다. 데이터 자체에 쉼표나 따옴표가 포함된 경우 처리가 까다롭습니다.

# 따옴표 처리 옵션 지정
import csv

with open('quotes_data.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file, quotechar='"', quoting=csv.QUOTE_MINIMAL)
    for row in reader:
        print(row)
        
# pandas에서의 처리
df = pd.read_csv('quotes_data.csv', 
                 quotechar='"', 
                 escapechar='\\',
                 doublequote=True)

마지막으로, 대용량 CSV 파일을 처리할 때 메모리 부족 문제가 발생할 수 있습니다. 특히 pandas는 전체 데이터를 메모리에 올리므로 기가바이트 단위의 파일을 처리할 때는 문제가 됩니다.

# 청크 단위로 대용량 파일 처리
chunks = pd.read_csv('huge_file.csv', chunksize=10000)

result = pd.DataFrame()
for chunk in chunks:
    # 각 청크 처리
    processed = chunk[chunk['값'] > 0]
    result = pd.concat([result, processed])
    
# 또는 기본 csv 모듈 사용
with open('huge_file.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    header = next(reader)
    for row in reader:
        # 한 행씩 처리
        process_row(row)

5. 대용량 CSV 처리 성능 최적화

기가바이트 단위의 CSV 파일을 처리해보신 적 있나요? 보통 사람들은 "pandas로 읽으면 되지"라고 생각하지만, 실제로 해보면 메모리 오류가 발생하거나 처리 시간이 너무 오래 걸리는 문제가 발생합니다. 다음은 대용량 CSV 파일 처리를 위한 성능 최적화 방법들입니다.

최적화 방법	장점	단점	사용 사례
필요한 컬럼만 선택	메모리 사용량 감소	모든 데이터에 접근 불가	특정 컬럼만 필요한 경우
청킹(Chunking) 기법	메모리 효율적 사용	전체 데이터 조회 어려움	행 단위 처리가 가능한 경우
데이터 타입 최적화	메모리 사용량 대폭 감소	타입 지정 작업 필요	정형화된 데이터셋
Dask 활용	분산 처리로 대용량 처리	설정 복잡, 학습 곡선	테라바이트 단위 데이터
다중 프로세스 활용	CPU 코어 모두 활용	코드 복잡도 증가	CPU 집약적 처리

필요한 컬럼만 선택적으로 로드하는 방법은 간단하면서도 효과적입니다:

# 필요한 컬럼만 로드
df = pd.read_csv('large_file.csv', usecols=['이름', '금액', '날짜'])

# 데이터 타입 최적화
dtypes = {
    '이름': 'category',  # 반복되는 값이 많은 문자열은 category로
    'ID': 'int32',       # 64비트 대신 32비트 사용
    '금액': 'float32'     # 64비트 대신 32비트 사용
}
df = pd.read_csv('large_file.csv', dtype=dtypes)

# 메모리 사용량 확인
print(f"메모리 사용량: {df.memory_usage().sum() / 1024**2:.2f} MB")

Dask는 pandas와 유사한 API를 제공하면서 대용량 데이터를 처리할 수 있는 라이브러리입니다:

import dask.dataframe as dd

# 대용량 CSV 파일을 Dask DataFrame으로 읽기
ddf = dd.read_csv('very_large_file.csv')

# 간단한 계산 수행 (실제 계산은 compute() 호출 시 실행)
result = ddf[ddf['금액'] > 10000].groupby('카테고리').금액.mean().compute()

6. 실전 CSV 처리 예제

실제 업무에서 자주 마주치는 CSV 처리 시나리오와 그 해결책을 알아보겠습니다. 이론적인 이야기보다 실제 코드를 보는 게 도움이 될 겁니다.

여러 CSV 파일 병합하기

여러 CSV 파일을 하나로 합쳐야 하는 경우가 많습니다. 예를 들어 매일 생성되는 로그 파일을 한 번에 분석하거나, 여러 부서에서 받은 데이터를 통합할 때 유용합니다.

import pandas as pd
import glob
import os

# 특정 패턴의 모든 CSV 파일 찾기
csv_files = glob.glob('data_*.csv')

# 빈 DataFrame 생성
combined_df = pd.DataFrame()

# 각 파일을 읽어서 하나로 합치기
for file in csv_files:
    # 파일명에서 날짜 추출 (예: data_2025-03-15.csv)
    date_str = os.path.basename(file).split('_')[1].split('.')[0]
    
    # 현재 파일 읽기
    df = pd.read_csv(file)
    
    # 날짜 정보 추가
    df['날짜'] = date_str
    
    # 기존 데이터에 추가
    combined_df = pd.concat([combined_df, df], ignore_index=True)

# 결과 저장
combined_df.to_csv('combined_data.csv', index=False)

CSV 데이터 정제 및 변환

실무에서는 원본 데이터가 깨끗하지 않은 경우가 대부분입니다. 누락된 값, 잘못된 형식, 중복된 항목 등을 처리해야 합니다.

1. 누락된 값 처리: 평균, 중앙값, 최빈값 등으로 대체하거나 행/열 삭제
2. 이상치(Outlier) 처리: 통계적 방법으로 이상치 식별 및 처리
3. 데이터 형식 통일: 날짜, 숫자 등의 형식 통일
4. 중복 데이터 제거: 완전 중복 또는 특정 컬럼 기준 중복 제거
5. 텍스트 데이터 정제: 공백 제거, 대소문자 통일, 특수문자 처리 등

import pandas as pd
import numpy as np

# 더러운 데이터 읽기
df = pd.read_csv('dirty_data.csv')

# 1. 누락된 값 처리
# 숫자 컬럼은 중앙값으로 채우기
numeric_cols = df.select_dtypes(include=['number']).columns
for col in numeric_cols:
    median_value = df[col].median()
    df[col] = df[col].fillna(median_value)

# 범주형 데이터는 최빈값으로 채우기
categorical_cols = df.select_dtypes(include=['object', 'category']).columns
for col in categorical_cols:
    mode_value = df[col].mode()[0]
    df[col] = df[col].fillna(mode_value)

# 2. 이상치 처리 (IQR 방법)
for col in numeric_cols:
    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
    
    # 이상치를 경계값으로 대체
    df[col] = np.where(df[col] < lower_bound, lower_bound, df[col])
    df[col] = np.where(df[col] > upper_bound, upper_bound, df[col])

# 3. 데이터 형식 통일
# 예: '나이' 컬럼의 문자열 값을 숫자로 변환
df['나이'] = pd.to_numeric(df['나이'], errors='coerce')

# 날짜 형식 통일
df['날짜'] = pd.to_datetime(df['날짜'], errors='coerce')

# 4. 중복 제거
df = df.drop_duplicates()

# 5. 텍스트 데이터 정제
# 예: '이름' 컬럼의 앞뒤 공백 제거 및 대소문자 통일
df['이름'] = df['이름'].str.strip().str.title()

# 정제된 데이터 저장
df.to_csv('clean_data.csv', index=False)

📝 판다스 속도 최적화

pandas 연산이 너무 느리다면:

• 가능하면 apply() 대신 벡터화 연산 사용하기 (df['A'] + df['B'] 같은 형태)
• apply() 사용할 때는 axis=1보다 axis=0이 빠름
• 문자열 처리는 pandas의 str 액세서보다 정규식이 빠른 경우가 많음
• iterrows()보다 itertuples()가 훨씬 빠름
• 대규모 데이터 시각화 시 sample()로 표본추출 후 시각화하기

📝 CSV vs Excel: 구축 환경에 따른 선택 가이드

CSV와 Excel 중 어떤 형식을 선택해야 할까요?

• CSV 선택할 때: 용량이 중요하거나, 여러 시스템 간 호환성이 필요하거나, 버전 관리 시스템(Git 등)으로 추적할 때
• Excel 선택할 때: 복잡한 서식이 필요하거나, 여러 시트/워크북이 필요하거나, 최종 사용자가 직접 데이터를 편집해야 할 때
• 서버 환경에서는: Excel 파일 처리를 위해 추가 라이브러리(openpyxl, xlrd 등)가 필요하므로 의존성이 늘어납니다. 단순 데이터 저장/교환이면 CSV를 선택하세요.
• 자동화 파이프라인에서는: CSV가 처리 속도가 빠르고 메모리 효율적입니다.

실용적인 CSV 처리 종합 예제

다음은 실제 업무에서 유용하게 쓸 수 있는 종합적인 CSV 처리 스크립트입니다. 이 스크립트는 다양한 기능을 제공합니다:

1. 여러 CSV 파일을 하나로 병합
2. 데이터 정제 및 전처리
3. 기본적인 데이터 분석
4. 결과를 다양한 형식으로 저장

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CSV 파일 처리를 위한 종합적인 유틸리티
사용법: python csv_processor.py input_dir output_dir

주요 기능:
1. 여러 CSV 파일 병합
2. 데이터 정제 및 전처리
3. 기본 통계 분석
4. 결과 저장 (CSV, Excel, JSON)
"""

import pandas as pd
import numpy as np
import os
import glob
import sys
import json
import logging
import argparse
from datetime import datetime
from typing import Dict, List, Union, Any, Tuple, Optional

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("csv_processor.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class CSVProcessor:
    """CSV 파일 처리를 위한 종합 클래스"""
    
    def __init__(self, input_dir: str, output_dir: str, 
                 encoding: str = 'utf-8', 
                 date_cols: Optional[List[str]] = None,
                 numeric_cols: Optional[List[str]] = None) -> None:
        """
        초기화
        
        Args:
            input_dir: 입력 CSV 파일이 있는 디렉토리
            output_dir: 출력 파일을 저장할 디렉토리
            encoding: CSV 파일 인코딩 (기본: utf-8)
            date_cols: 날짜로 변환할 컬럼 목록
            numeric_cols: 숫자로 변환할 컬럼 목록
        """
        self.input_dir = input_dir
        self.output_dir = output_dir
        self.encoding = encoding
        self.date_cols = date_cols or []
        self.numeric_cols = numeric_cols or []
        self.combined_df = None
        self.stats = {}
        
        # 출력 디렉토리가 없으면 생성
        os.makedirs(output_dir, exist_ok=True)
        
        logger.info(f"CSVProcessor 초기화 완료: {input_dir} -> {output_dir}")
    
    def find_csv_files(self) -> List[str]:
        """입력 디렉토리에서 모든 CSV 파일 찾기"""
        pattern = os.path.join(self.input_dir, '*.csv')
        files = glob.glob(pattern)
        logger.info(f"{len(files)}개의 CSV 파일을 찾았습니다.")
        return files
    
    def detect_encoding(self, file_path: str) -> str:
        """
        파일의 인코딩 감지 시도
        실패 시 기본 인코딩 반환
        """
        try:
            import chardet
            with open(file_path, 'rb') as f:
                result = chardet.detect(f.read())
            detected = result['encoding']
            confidence = result['confidence']
            
            if confidence > 0.7:
                logger.info(f"감지된 인코딩: {detected} (신뢰도: {confidence:.2f})")
                return detected
            else:
                logger.warning(f"인코딩 감지 신뢰도가 낮습니다: {confidence:.2f}. 기본값 사용: {self.encoding}")
                return self.encoding
        except ImportError:
            logger.warning("chardet 라이브러리가 설치되지 않았습니다. 기본 인코딩을 사용합니다.")
            return self.encoding
        except Exception as e:
            logger.error(f"인코딩 감지 중 오류 발생: {str(e)}")
            return self.encoding
    
    def combine_csv_files(self) -> None:
        """모든 CSV 파일을 하나의 DataFrame으로 병합"""
        files = self.find_csv_files()
        if not files:
            raise ValueError("처리할 CSV 파일이 없습니다.")
        
        all_dfs = []
        for file in files:
            try:
                # 파일 인코딩 감지 (선택 사항)
                encoding = self.detect_encoding(file)
                
                # 파일 읽기 시도
                df = pd.read_csv(file, encoding=encoding, low_memory=False)
                
                # 파일 정보 추가
                file_name = os.path.basename(file)
                df['source_file'] = file_name
                df['processed_date'] = datetime.now().strftime('%Y-%m-%d')
                
                all_dfs.append(df)
                logger.info(f"파일 로드 성공: {file_name}, 행 수: {len(df)}")
            except Exception as e:
                logger.error(f"파일 로드 실패: {file}, 오류: {str(e)}")
        
        if not all_dfs:
            raise ValueError("모든 파일 로드에 실패했습니다.")
        
        # 모든 DataFrame 병합
        self.combined_df = pd.concat(all_dfs, ignore_index=True)
        logger.info(f"모든 CSV 파일 병합 완료. 총 행 수: {len(self.combined_df)}")
    
    def clean_data(self) -> None:
        """데이터 정제 및 전처리"""
        if self.combined_df is None:
            raise ValueError("병합된 데이터가 없습니다. combine_csv_files()를 먼저 호출하세요.")
        
        # 중복 행 제거
        before_len = len(self.combined_df)
        self.combined_df = self.combined_df.drop_duplicates()
        after_len = len(self.combined_df)
        logger.info(f"중복 제거: {before_len - after_len}개 행 제거됨")
        
        # 날짜 컬럼 변환
        for col in self.date_cols:
            if col in self.combined_df.columns:
                try:
                    self.combined_df[col] = pd.to_datetime(self.combined_df[col], errors='coerce')
                    logger.info(f"컬럼 변환: {col} -> datetime")
                except Exception as e:
                    logger.error(f"날짜 변환 실패: {col}, 오류: {str(e)}")
        
        # 숫자 컬럼 변환
        for col in self.numeric_cols:
            if col in self.combined_df.columns:
                try:
                    self.combined_df[col] = pd.to_numeric(self.combined_df[col], errors='coerce')
                    logger.info(f"컬럼 변환: {col} -> numeric")
                except Exception as e:
                    logger.error(f"숫자 변환 실패: {col}, 오류: {str(e)}")
        
        # 문자열 컬럼 공백 제거
        str_cols = self.combined_df.select_dtypes(include=['object']).columns
        for col in str_cols:
            self.combined_df[col] = self.combined_df[col].str.strip()
        
        # NA 값 처리
        na_counts = self.combined_df.isna().sum()
        logger.info("NA 값 개수:\n" + str(na_counts[na_counts > 0]))
        
        logger.info("데이터 정제 완료")
    
    def analyze_data(self) -> Dict[str, Any]:
        """기본적인 데이터 분석 수행"""
        if self.combined_df is None:
            raise ValueError("병합된 데이터가 없습니다.")
        
        stats = {}
        
        # 기본 통계
        stats['row_count'] = len(self.combined_df)
        stats['column_count'] = len(self.combined_df.columns)
        stats['columns'] = list(self.combined_df.columns)
        
        # 수치형 컬럼 통계
        numeric_stats = {}
        numeric_cols = self.combined_df.select_dtypes(include=['number']).columns
        for col in numeric_cols:
            numeric_stats[col] = {
                'min': float(self.combined_df[col].min()),
                'max': float(self.combined_df[col].max()),
                'mean': float(self.combined_df[col].mean()),
                'median': float(self.combined_df[col].median()),
                'std': float(self.combined_df[col].std()),
                'null_count': int(self.combined_df[col].isna().sum())
            }
        stats['numeric_stats'] = numeric_stats
        
        # 범주형 컬럼 통계
        categorical_stats = {}
        cat_cols = self.combined_df.select_dtypes(include=['object', 'category']).columns
        for col in cat_cols[:5]:  # 처음 5개 컬럼만 처리
            value_counts = self.combined_df[col].value_counts().head(10).to_dict()
            categorical_stats[col] = {
                'unique_count': self.combined_df[col].nunique(),
                'top_values': value_counts,
                'null_count': int(self.combined_df[col].isna().sum())
            }
        stats['categorical_stats'] = categorical_stats
        
        self.stats = stats
        logger.info("데이터 분석 완료")
        return stats
    
    def save_results(self) -> Dict[str, str]:
        """결과를 다양한 형식으로 저장"""
        if self.combined_df is None:
            raise ValueError("병합된 데이터가 없습니다.")
        
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        output_files = {}
        
        # CSV로 저장
        csv_path = os.path.join(self.output_dir, f"combined_data_{timestamp}.csv")
        self.combined_df.to_csv(csv_path, index=False, encoding=self.encoding)
        output_files['csv'] = csv_path
        logger.info(f"CSV 파일 저장 완료: {csv_path}")
        
        # Excel로 저장
        try:
            excel_path = os.path.join(self.output_dir, f"combined_data_{timestamp}.xlsx")
            with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
                self.combined_df.to_excel(writer, sheet_name='Data', index=False)
                
                # 통계 정보 별도 시트에 저장
                if self.stats:
                    # 수치형 통계
                    if 'numeric_stats' in self.stats:
                        numeric_stats_df = pd.DataFrame()
                        for col, stats in self.stats['numeric_stats'].items():
                            temp_df = pd.DataFrame(stats, index=[col]).T
                            numeric_stats_df = pd.concat([numeric_stats_df, temp_df], axis=1)
                        numeric_stats_df.to_excel(writer, sheet_name='Numeric_Stats')
                    
                    # 요약 정보
                    summary_data = {
                        '항목': ['총 행 수', '총 컬럼 수', '생성 시각'],
                        '값': [
                            self.stats['row_count'],
                            self.stats['column_count'],
                            datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        ]
                    }
                    pd.DataFrame(summary_data).to_excel(writer, sheet_name='Summary', index=False)
            
            output_files['excel'] = excel_path
            logger.info(f"Excel 파일 저장 완료: {excel_path}")
        except Exception as e:
            logger.error(f"Excel 파일 저장 실패: {str(e)}")
        
        # 통계 정보를 JSON으로 저장
        if self.stats:
            json_path = os.path.join(self.output_dir, f"stats_{timestamp}.json")
            with open(json_path, 'w', encoding='utf-8') as f:
                json.dump(self.stats, f, ensure_ascii=False, indent=2)
            output_files['json'] = json_path
            logger.info(f"JSON 통계 파일 저장 완료: {json_path}")
        
        return output_files
    
    def process_all(self) -> Dict[str, str]:
        """모든 처리 단계를 순차적으로 실행"""
        logger.info("=== CSV 처리 시작 ===")
        self.combine_csv_files()
        self.clean_data()
        self.analyze_data()
        output_files = self.save_results()
        logger.info("=== CSV 처리 완료 ===")
        return output_files


def main():
    """메인 함수"""
    parser = argparse.ArgumentParser(description='CSV 파일 처리 유틸리티')
    parser.add_argument('input_dir', help='입력 CSV 파일이 있는 디렉토리')
    parser.add_argument('output_dir', help='출력 파일을 저장할 디렉토리')
    parser.add_argument('--encoding', default='utf-8', help='CSV 파일 인코딩 (기본: utf-8)')
    parser.add_argument('--date-cols', nargs='+', help='날짜로 변환할 컬럼 목록')
    parser.add_argument('--numeric-cols', nargs='+', help='숫자로 변환할 컬럼 목록')
    
    args = parser.parse_args()
    
    try:
        processor = CSVProcessor(
            input_dir=args.input_dir,
            output_dir=args.output_dir,
            encoding=args.encoding,
            date_cols=args.date_cols,
            numeric_cols=args.numeric_cols
        )
        
        output_files = processor.process_all()
        
        print("\n=== 처리 완료 ===")
        print(f"입력 디렉토리: {args.input_dir}")
        print(f"출력 디렉토리: {args.output_dir}")
        print("\n생성된 파일:")
        for file_type, file_path in output_files.items():
            print(f"- {file_type.upper()}: {file_path}")
        
        return 0
    except Exception as e:
        logger.error(f"처리 중 오류 발생: {str(e)}", exc_info=True)
        return 1


if __name__ == "__main__":
    sys.exit(main())

이 스크립트는 명령줄에서 다음과 같이 사용할 수 있습니다:

# 기본 사용법
python csv_processor.py input_folder output_folder

# 날짜와 숫자 컬럼 지정
python csv_processor.py input_folder output_folder --date-cols 생일 가입일 --numeric-cols 나이 금액

# 인코딩 지정
python csv_processor.py input_folder output_folder --encoding cp949

CSV와 JSON 간 변환 유틸리티

CSV와 JSON은 모두 널리 사용되는 데이터 교환 형식입니다. 때로는 이 두 형식 간의 변환이 필요할 수 있습니다. 다음은 이를 위한 간단한 스크립트입니다:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CSV와 JSON 간 변환 유틸리티
"""

import pandas as pd
import json
import argparse
import sys
import os

def csv_to_json(csv_file, json_file, encoding='utf-8', orient='records'):
    """CSV 파일을 JSON으로 변환"""
    try:
        # CSV 파일 읽기
        df = pd.read_csv(csv_file, encoding=encoding)
        
        # 데이터 타입 최적화 (선택 사항)
        # 수치형 컬럼을 적절한 타입으로 변환
        for col in df.select_dtypes(include=['int64']).columns:
            if df[col].min() >= -128 and df[col].max() <= 127:
                df[col] = df[col].astype('int8')
            elif df[col].min() >= -32768 and df[col].max() <= 32767:
                df[col] = df[col].astype('int16')
            elif df[col].min() >= -2147483648 and df[col].max() <= 2147483647:
                df[col] = df[col].astype('int32')
        
        # JSON으로 변환
        if orient == 'records':
            # 레코드 배열 형태 (가장 일반적)
            # [{"column1": value1, "column2": value2}, {...}, ...]
            result = df.to_json(orient='records', force_ascii=False, date_format='iso')
            json_data = json.loads(result)
        elif orient == 'split':
            # 분할 형태 {"columns": [...], "index": [...], "data": [...]}
            result = df.to_json(orient='split', force_ascii=False, date_format='iso')
            json_data = json.loads(result)
        else:
            raise ValueError(f"지원하지 않는 orient 값: {orient}")
        
        # 파일에 쓰기
        with open(json_file, 'w', encoding='utf-8') as f:
            json.dump(json_data, f, ensure_ascii=False, indent=2)
        
        print(f"변환 완료: {csv_file} -> {json_file}")
        print(f"총 레코드 수: {len(df)}")
        return True
    
    except Exception as e:
        print(f"오류 발생: {str(e)}")
        return False

def json_to_csv(json_file, csv_file, encoding='utf-8'):
    """JSON 파일을 CSV로 변환"""
    try:
        # JSON 파일 읽기
        with open(json_file, 'r', encoding=encoding) as f:
            json_data = json.load(f)
        
        # JSON 구조 확인
        if isinstance(json_data, list):
            # 레코드 배열 형태
            df = pd.DataFrame(json_data)
        elif isinstance(json_data, dict) and 'data' in json_data and 'columns' in json_data:
            # 분할 형태
            df = pd.DataFrame(json_data['data'], columns=json_data['columns'])
        elif isinstance(json_data, dict):
            # 중첩된 JSON을 평탄화해서 DataFrame으로 변환
            flat_data = []
            for key, value in json_data.items():
                if isinstance(value, dict):
                    row = {'id': key}
                    row.update(value)
                    flat_data.append(row)
                else:
                    # 단일 레벨 구조
                    flat_data = [json_data]
                    break
            df = pd.DataFrame(flat_data)
        else:
            raise ValueError("지원하지 않는 JSON 구조입니다.")
        
        # CSV 파일로 저장
        df.to_csv(csv_file, index=False, encoding=encoding)
        
        print(f"변환 완료: {json_file} -> {csv_file}")
        print(f"총 레코드 수: {len(df)}")
        return True
    
    except Exception as e:
        print(f"오류 발생: {str(e)}")
        return False

def main():
    parser = argparse.ArgumentParser(description='CSV와 JSON 간 변환 유틸리티')
    parser.add_argument('input_file', help='입력 파일 경로')
    parser.add_argument('output_file', help='출력 파일 경로')
    parser.add_argument('--encoding', default='utf-8', help='파일 인코딩 (기본: utf-8)')
    parser.add_argument('--orient', default='records', choices=['records', 'split'], 
                        help='JSON 형식 (CSV->JSON 변환 시): records 또는 split (기본: records)')
    
    args = parser.parse_args()
    
    # 파일 확장자로 변환 방향 결정
    input_ext = os.path.splitext(args.input_file)[1].lower()
    output_ext = os.path.splitext(args.output_file)[1].lower()
    
    if input_ext == '.csv' and output_ext == '.json':
        return 0 if csv_to_json(args.input_file, args.output_file, args.encoding, args.orient) else 1
    elif input_ext == '.json' and output_ext == '.csv':
        return 0 if json_to_csv(args.input_file, args.output_file, args.encoding) else 1
    else:
        print("오류: 입력 또는 출력 파일 형식이 올바르지 않습니다.")
        print("입력 파일은 .csv 또는 .json 이어야 합니다.")
        print("출력 파일은 입력과 다른 형식(.csv 또는 .json)이어야 합니다.")
        return 1

if __name__ == "__main__":
    sys.exit(main())

이상으로 파이썬을 사용한 CSV 파일 처리 방법에 대해 알아봤습니다. CSV 파일은 단순해 보이지만 실제로는 다양한 형식과 예외 상황 때문에 처리가 까다로울 수 있습니다. 이 글에서 제공한 코드와 팁을 활용하면 대부분의 CSV 관련 작업을 효율적으로 처리할 수 있을 겁니다. 특히 대용량 데이터 처리나 자동화된 파이프라인에서는 여기서 다룬 최적화 방법이 큰 도움이 될 것입니다.

CSV 파일 처리 시 가장 중요한 것은 데이터 품질과 일관성을 유지하는 것입니다. 인코딩 문제, 데이터 타입 불일치, 누락된 값 등의 문제를 미리 예방하고 적절히 처리하는 습관을 들이면 많은 시간과 노력을 절약할 수 있습니다. 이제 여러분도 CSV 파일을 효율적으로 처리하는 파이썬 개발자가 되셨길 바랍니다.

아, 그리고 마지막으로 하나 더 말씀드리자면, 가능하다면 처음부터 CSV 파일 형식과 구조를 명확히 정의하고 문서화하는 것이 좋습니다. 이렇게 하면 나중에 데이터를 처리할 때 발생할 수 있는 많은 문제를 예방할 수 있습니다. 이미 존재하는 시스템과 작업할 때는 그럴 수 없겠지만, 새 프로젝트를 시작한다면 꼭 기억하세요.