파이썬으로 여러 CSV 파일을 하나로 합치는 방법

데이터 분석 작업을 하다 보면 여러 개의 CSV 파일을 하나로 합쳐야 하는 경우가 종종 있습니다. 특히 기업의 ERP 시스템에서 대용량 데이터를 여러 파일로 나누어 다운로드받은 경우, 이를 다시 하나로 합쳐야 분석 작업을 수월하게 진행할 수 있습니다. 이번 글에서는 파이썬의 pandas 라이브러리를 활용해 여러 CSV 파일을 효율적으로 병합하는 방법을 소개합니다.

목차

1. CSV 파일 합치는 방법 비교
2. 개별 변수 저장 방식으로 CSV 파일 합치기
3. 반복문을 활용한 CSV 파일 합치기
4. 통합 결과 검증 방법
5. 한국어 환경의 인코딩 문제 해결
6. 최종 코드 및 실행 결과

1. CSV 파일 합치는 방법 비교

여러 CSV 파일을 합치는 방법에는 크게 다음과 같은 방법들이 있습니다:

1. pandas.concat() - 행 방향으로 여러 DataFrame을 연결
2. pandas.merge() - 특정 키를 기준으로 두 DataFrame을 병합 (SQL의 JOIN과 유사)
3. append() - DataFrame에 다른 DataFrame을 추가 (pandas 2.0부터 deprecated)
4. for 반복문과 concat - 반복문을 사용하여 여러 파일을 순차적으로 읽고 합치기

동일한 열 구조를 가진 여러 CSV 파일을 단순히 이어붙이는 목적이라면 pandas.concat()이 가장 적합합니다. 이 방법을 기준으로 파일을 합치는 두 가지 접근법을 비교해 보겠습니다.

2. 개별 변수 저장 방식으로 CSV 파일 합치기

첫 번째 방법은 각 파일을 개별 변수(df1, df2, df3, ...)에 저장한 후 이를 concat 함수로 합치는 방식입니다.

import pandas as pd
import os
import time

# 시작 시간 기록
start_time = time.time()

# 파일 경로 설정
path = r"C:\Data\Assets"

# 각 파일 경로 지정
file1 = os.path.join(path, "Asset_Data_01.csv")
file2 = os.path.join(path, "Asset_Data_02.csv")
file3 = os.path.join(path, "Asset_Data_03.csv")
# ... 필요한 파일 수만큼 계속

# 각 파일을 데이터프레임으로 읽기
print("각 파일 읽는 중...")
df1 = pd.read_csv(file1, encoding='cp949')
print(f"파일 1: {len(df1):,}행 읽음")
df2 = pd.read_csv(file2, encoding='cp949')
print(f"파일 2: {len(df2):,}행 읽음")
df3 = pd.read_csv(file3, encoding='cp949')
print(f"파일 3: {len(df3):,}행 읽음")
# ... 필요한 파일 수만큼 계속

# 모든 데이터프레임 합치기
print("\n파일 합치는 중...")
result = pd.concat([df1, df2, df3], ignore_index=True)

# 결과 저장
output_file = os.path.join(path, "통합_자산데이터.csv")
result.to_csv(output_file, index=False, encoding='cp949')

# 처리 시간 계산
elapsed_time = time.time() - start_time
print(f"처리 시간: {elapsed_time:.2f}초")

장점

• 각 파일에 대한 처리를 개별적으로 커스터마이즈할 수 있습니다.
• 특정 파일에 대한 전처리나 변환을 쉽게 추가할 수 있습니다.
• 디버깅이 쉽습니다 (각 단계가 명시적).

단점

• 파일 수가 많을수록 코드가 길어지고 반복적인 작업이 많아집니다.
• 파일 경로나 수가 변경되면 코드를 수정해야 합니다.
• 확장성이 떨어집니다.

3. 반복문을 활용한 CSV 파일 합치기

두 번째 방법은 반복문을 사용하여 디렉토리 내의 모든 해당 파일을 자동으로 찾아 합치는 방식입니다.

import pandas as pd
import os
import glob
import time

# 시작 시간 기록
start_time = time.time()

# 파일 경로 설정
path = r"C:\Data\Assets"

# 파일 패턴 설정
file_pattern = os.path.join(path, "Asset_Data_*.csv")

# 파일 목록 가져오기
all_files = glob.glob(file_pattern)

# 파일 순서대로 정렬
all_files.sort()

# 결과를 저장할 빈 DataFrame 리스트 생성
dfs = []

# 각 파일 읽어서 리스트에 추가
for file in all_files:
    print(f"파일 읽는 중: {os.path.basename(file)}")
    df = pd.read_csv(file, encoding='cp949')
    dfs.append(df)
    print(f"  - 레코드 수: {len(df)}")

# 모든 DataFrame 합치기
result = pd.concat(dfs, ignore_index=True)

# 결과 저장
output_file = os.path.join(path, "통합_자산데이터.csv")
result.to_csv(output_file, index=False, encoding='cp949')

# 처리 시간 계산
elapsed_time = time.time() - start_time
print(f"처리 시간: {elapsed_time:.2f}초")

장점

• 코드가 간결하고 유지보수가 쉽습니다.
• 파일 수가 변경되어도 코드 수정이 필요 없습니다.
• 파일 이름 패턴만 맞으면 모든 파일을 자동으로 처리합니다.
• 확장성이 뛰어납니다.

단점

• 개별 파일마다 다른 처리를 적용하기 어렵습니다.
• 모든 파일이 동일한 구조와 형식을 가져야 합니다.

4. 통합 결과 검증 방법

CSV 파일들을 성공적으로 합쳤다면, 통합된 결과가 올바른지 검증해야 합니다. 다음은 일반적인 검증 방법입니다:

# 검증 코드
# 1. 각 파일의 행 수 합계와 통합 파일의 행 수 비교
total_rows_original = sum(len(pd.read_csv(file, encoding='cp949')) for file in all_files)
total_rows_merged = len(result)

print("\n===== 검증 결과 =====")
print(f"원본 파일 총 행 수: {total_rows_original:,}")
print(f"통합 파일 행 수: {total_rows_merged:,}")
print(f"일치 여부: {'일치' if total_rows_original == total_rows_merged else '불일치'}")

# 2. 각 파일별 중요 통계치 검증 (예: 금액의 합)
if '금액' in result.columns:
    total_amount_original = sum(pd.read_csv(file, encoding='cp949')['금액'].sum() for file in all_files)
    total_amount_merged = result['금액'].sum()

    print(f"원본 파일 금액 합계: {total_amount_original:,.0f}")
    print(f"통합 파일 금액 합계: {total_amount_merged:,.0f}")
    print(f"금액 일치 여부: {'일치' if abs(total_amount_original - total_amount_merged) < 0.01 else '불일치'}")

# 3. 중복 레코드 확인
duplicate_count = result.duplicated().sum()
print(f"중복 레코드 수: {duplicate_count:,}")

주요 검증 포인트:

1. 행 수 검증: 원본 파일들의 총 행 수와 통합 파일의 행 수가 일치해야 합니다.
2. 주요 필드 합계 검증: 금액이나 수량과 같은 주요 필드의 합계가 일치해야 합니다.
3. 중복 레코드 확인: 의도하지 않은 중복 레코드가 없는지 확인합니다.

5. 한국어 환경의 인코딩 문제 해결

한국어 윈도우 환경에서 CSV 파일을 다룰 때는 인코딩 문제가 자주 발생합니다. 일반적으로 한글 Windows에서 생성된 CSV 파일은 'cp949' 또는 'euc-kr' 인코딩을 사용합니다. 다음은 인코딩 문제를 해결하는 함수입니다:

# 인코딩 찾기 시도 함수
def try_encodings(file_path):
    encodings = ['cp949', 'euc-kr', 'utf-8', 'ISO-8859-1']
    for encoding in encodings:
        try:
            # 파일의 첫 몇 줄만 읽어서 인코딩 테스트
            with open(file_path, 'r', encoding=encoding) as f:
                f.readline()
                return encoding
        except UnicodeDecodeError:
            continue
    return None

# 파일 읽기 함수
def read_csv_safe(file_path, enc):
    try:
        return pd.read_csv(file_path, encoding=enc)
    except Exception as e:
        print(f"오류 발생! {os.path.basename(file_path)} 파일을 읽는 중 문제가 발생했습니다: {str(e)}")
        # 다른 인코딩도 시도
        alt_encodings = ['cp949', 'euc-kr', 'utf-8', 'ISO-8859-1']
        for alt_enc in alt_encodings:
            if alt_enc != enc:
                try:
                    print(f"대체 인코딩 {alt_enc}으로 시도...")
                    return pd.read_csv(file_path, encoding=alt_enc)
                except:
                    continue
        # 모든 인코딩이 실패하면 엔진 변경 시도
        try:
            return pd.read_csv(file_path, encoding='cp949', engine='python')
        except:
            raise Exception(f"파일 {os.path.basename(file_path)}을 읽을 수 없습니다.")

이 함수들을 사용하면 다양한 인코딩으로 저장된 CSV 파일을 안정적으로 읽을 수 있습니다.

6. 최종 코드 및 실행 결과

아래는 인코딩 문제 해결과 결과 검증을 포함한 최종 코드입니다. 이 코드는 반복문을 사용해 여러 CSV 파일을 합치고, 결과를 철저히 검증합니다.

import pandas as pd
import os
import glob
import time
import re

# 시작 시간 기록
start_time = time.time()

# 파일 경로 설정
path = r"C:\Data\Assets"

# 파일 패턴 설정
file_pattern = os.path.join(path, "Asset_Data_*.csv")

# 파일 목록 가져오기
all_files = glob.glob(file_pattern)

# 파일 순서대로 정렬 (숫자 순으로 정렬하기 위한 함수)
def extract_number(filename):
    # 파일명에서 숫자 부분 추출
    match = re.search(r'Data_(\d+)', filename)
    if match:
        return int(match.group(1))
    return 0

all_files.sort(key=extract_number)

# 인코딩 찾기 시도 함수
def try_encodings(file_path):
    encodings = ['cp949', 'euc-kr', 'utf-8', 'ISO-8859-1']
    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                f.readline()
                return encoding
        except UnicodeDecodeError:
            continue
    return None

# 파일 읽기 함수
def read_csv_safe(file_path, enc):
    try:
        return pd.read_csv(file_path, encoding=enc)
    except Exception as e:
        print(f"오류 발생! {os.path.basename(file_path)} 파일을 읽는 중 문제가 발생했습니다: {str(e)}")
        # 다른 인코딩도 시도
        alt_encodings = ['cp949', 'euc-kr', 'utf-8', 'ISO-8859-1']
        for alt_enc in alt_encodings:
            if alt_enc != enc:
                try:
                    print(f"대체 인코딩 {alt_enc}으로 시도...")
                    return pd.read_csv(file_path, encoding=alt_enc)
                except:
                    continue
        # 모든 인코딩이 실패하면 에러 처리
        try:
            return pd.read_csv(file_path, encoding='cp949', engine='python')
        except:
            raise Exception(f"파일 {os.path.basename(file_path)}을 읽을 수 없습니다.")

# 첫 번째 파일의 인코딩 검사
if all_files:
    detected_encoding = try_encodings(all_files[0])
    if detected_encoding:
        print(f"감지된 인코딩: {detected_encoding}")
    else:
        print("인코딩을 자동으로 감지할 수 없습니다. 'cp949'를 사용합니다.")
        detected_encoding = 'cp949'
else:
    print("파일을 찾을 수 없습니다.")
    exit(1)

# 파일 정보 저장을 위한 리스트
file_info = []

# 반복문으로 모든 파일 처리
dfs = []
print(f"\n총 {len(all_files)}개 파일을 처리합니다...")

for i, file in enumerate(all_files, 1):
    filename = os.path.basename(file)
    print(f"파일 {i}/{len(all_files)} 읽는 중: {filename}")
    
    try:
        # 파일 읽기
        df = read_csv_safe(file, detected_encoding)
        rows = len(df)
        
        # 금액 합계 계산 (있는 경우)
        if '금액' in df.columns:
            amount_sum = df['금액'].sum()
            print(f"  - 레코드 수: {rows:,}, 금액 합계: {amount_sum:,.0f}")
        else:
            amount_sum = 0
            print(f"  - 레코드 수: {rows:,}")

        # 파일 정보 저장
        file_info.append({
            'filename': filename,
            'rows': rows,
            'amount_sum': amount_sum
        })
        
        # DataFrame 리스트에 추가
        dfs.append(df)
        
    except Exception as e:
        print(f"파일 {filename} 처리 중 오류 발생: {str(e)}")
        continue

# 모든 DataFrame 합치기
if dfs:
    print("\n모든 파일을 하나로 합치는 중...")
    result = pd.concat(dfs, ignore_index=True)
    
    # 결과 저장
    output_file = os.path.join(path, "통합_자산데이터.csv")
    print(f"결과 저장 중: {output_file}")
    result.to_csv(output_file, index=False, encoding=detected_encoding)
    
    # 검증
    total_rows_original = sum(info['rows'] for info in file_info)
    total_rows_merged = len(result)
    
    if '금액' in result.columns:
        total_amount_original = sum(info['amount_sum'] for info in file_info)
        total_amount_merged = result['금액'].sum()
    else:
        total_amount_original = 0
        total_amount_merged = 0
        
    duplicate_count = result.duplicated().sum()
    
    # 결과 출력
    end_time = time.time()
    elapsed_time = end_time - start_time
    
    print("\n===== 처리 결과 =====")
    print(f"처리된 파일 수: {len(dfs)}")
    print(f"총 레코드 수: {total_rows_merged:,}")
    print(f"처리 시간: {elapsed_time:.2f}초")
    
    print("\n===== 검증 결과 =====")
    print(f"원본 파일 총 행 수: {total_rows_original:,}")
    print(f"통합 파일 행 수: {total_rows_merged:,}")
    print(f"행 수 일치 여부: {'일치' if total_rows_original == total_rows_merged else '불일치'}")
    
    if '금액' in result.columns:
        print(f"원본 파일 금액 합계: {total_amount_original:,.0f}")
        print(f"통합 파일 금액 합계: {total_amount_merged:,.0f}")
        print(f"금액 일치 여부: {'일치' if abs(total_amount_original - total_amount_merged) < 0.01 else '불일치'}")
    
    print(f"중복 레코드 수: {duplicate_count:,}")
    
    # 데이터 요약
    print("\n===== 데이터 요약 =====")
    print(result.info())
    print("\n데이터 샘플:")
    print(result.head())
else:
    print("처리할 파일이 없거나 모든 파일 처리에 실패했습니다.")

실행 결과 예시

감지된 인코딩: cp949

총 20개 파일을 처리합니다...
파일 1/20 읽는 중: Asset_Data_01.csv
  - 레코드 수: 50,000, 금액 합계: 123,456,789
파일 2/20 읽는 중: Asset_Data_02.csv
  - 레코드 수: 45,000, 금액 합계: 98,765,432
...

모든 파일을 하나로 합치는 중...
결과 저장 중: C:\Data\Assets\통합_자산데이터.csv

===== 처리 결과 =====
처리된 파일 수: 20
총 레코드 수: 950,000
처리 시간: 45.23초

===== 검증 결과 =====
원본 파일 총 행 수: 950,000
통합 파일 행 수: 950,000
행 수 일치 여부: 일치
원본 파일 금액 합계: 1,987,654,321
통합 파일 금액 합계: 1,987,654,321
금액 일치 여부: 일치
중복 레코드 수: 0

결론

이 글에서는 파이썬의 pandas를 사용하여 여러 CSV 파일을 하나로 합치는 두 가지 방법을 살펴보았습니다. 개별 변수 저장 방식과 반복문을 활용한 방식 모두 장단점이 있으며, 상황에 맞게 선택하면 됩니다.

일반적으로 처리할 파일이 많거나 파일 구조가 동일하다면 반복문을 활용한 방식이 더 효율적입니다. 인코딩 문제가 발생하는 경우에는 제공된 read_csv_safe 함수를 활용하여 안정적으로 파일을 읽을 수 있습니다.

마지막으로, 통합 결과를 검증하는 단계는 매우 중요합니다. 행 수, 주요 필드의 합계, 중복 레코드 여부 등을 확인하여 데이터가 올바르게 병합되었는지 확인해야 합니다.

이 방법을 활용하면 대용량 CSV 파일을 효율적으로 관리하고 분석하는 데 도움이 될 것입니다.