Python - 데이터수집처리 (적재) 3

데이터수집처리 1탄 : https://pabeba.tistory.com/276

데이터수집처리 2탄 : https://pabeba.tistory.com/277

 

오늘은 마지막으로 데이터를 적재해 보겠습니다.

앞서 진행된 데이터 추출(Extract) 및 변환(Transform) 단계를 거쳐 생성된 최종 데이터를 영구적인 저장 매체(파일, 데이터베이스 등)에 보관하는 적재(Load) 단계에 대한 가이드입니다.

파이썬의 데이터 분석 라이브러리인 pandas를 활용하면 리스트 형태의 데이터를 표(Table) 형태의 DataFrame으로 변환할 수 있으며, 내장 메서드를 통해 CSV 파일이나 RDBMS(관계형 데이터베이스)로 간편하게 적재할 수 있습니다.

---

1. 데이터프레임(DataFrame) 변환

파이썬의 딕셔너리를 요소로 가지는 리스트([{}, ...])를 pandas DataFrame으로 변환합니다.

딕셔너리의 키(Key)는 컬럼(Column)명으로, 값(Value)은 행(Row) 데이터로 자동 매핑됩니다.

import pandas as pd

# final_exchanges는 추출/변환 단계에서 생성된 딕셔너리 리스트로 가정합니다.
df = pd.DataFrame(final_exchanges)

# 데이터 구조 및 상위 5개 행 확인
print(df.head())

---

2. CSV 파일로 적재

변환된 DataFrame은 to_csv() 메서드를 사용하여 로컬 파일로 저장할 수 있습니다.

csv_file_path = "exchange_rates.csv"

# index=False: DataFrame의 기본 인덱스(0, 1, 2...)를 파일에 쓰지 않음
# encoding="utf-8-sig": 엑셀에서 한글이 깨지는 현상을 방지하기 위해 BOM(Byte Order Mark)을 포함한 UTF-8로 저장
df.to_csv(csv_file_path, index=False, encoding="utf-8-sig")

---

3. 데이터베이스(DB)로 적재

데이터베이스 적재는 따로 해보지는 않았는데 나중에 프로젝트 진행할 때 겁나 할 것 같으니 그때 해보고 오류가 있으면 글 수정하겠습니다! 여러분도 해보시고 안 되는 것이 있으면 문의 남겨주세요 ㅎ

3.1. SQLite 적재 (로컬 환경)

SQLite는 파이썬에 기본 내장되어 있어 별도의 서버 구축 없이 즉시 사용 가능한 경량 데이터베이스입니다. sqlite3 모듈과 DataFrame의 to_sql() 메서드를 사용합니다.

import sqlite3

# 1. DB 연결 객체 생성 (파일이 없으면 자동 생성됨)
db_path = "exchange_rates.db"
conn = sqlite3.connect(db_path)

# 2. DataFrame 데이터를 DB의 'exchange_list' 테이블에 적재
# if_exists 옵션:
#  - 'append': 기존 테이블 데이터 하단에 새 데이터를 추가 (주기적 수집에 적합)
#  - 'replace': 기존 테이블을 삭제하고 덮어쓰기
#  - 'fail': 테이블이 존재하면 아무 작업도 하지 않음 (기본값)
df.to_sql('exchange_list', conn, if_exists='append', index=False)

# 3. DB 연결 종료
conn.close()

3.2. MySQL 적재

MySQL과 같은 외부 RDBMS에 적재하기 위해서는 데이터베이스 통신 엔진인 sqlalchemy와 MySQL 접속 드라이버인 pymysql 패키지가 필요합니다.

# 필수 패키지 설치
pip install pandas sqlalchemy pymysql

 

[주의 및 수정 사항: 인증 정보 하드코딩 방지] 제시된 요건 중 DB_PASSWORD = '비밀번호'와 같이 소스 코드 내부에 데이터베이스 접속 정보를 직접 기입(하드코딩)하는 것은 심각한 보안 취약점을 유발합니다.

정보가 외부 저장소(Git 등)에 유출될 위험이 있으므로, 반드시 환경 변수(Environment Variables)를 통해 안전하게 주입받아야 합니다. 아래 코드는 이를 정정한 구현 방식입니다.

import os
import pandas as pd
from sqlalchemy import create_engine

# 1. 환경 변수에서 MySQL 접속 정보 획득 (사전 설정 필요)
DB_USER = os.getenv('DB_USER', 'root')
DB_PASSWORD = os.getenv('DB_PASSWORD') 
DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
DB_PORT = os.getenv('DB_PORT', '3306')
DB_NAME = os.getenv('DB_NAME', 'exchange_db')

if not DB_PASSWORD:
    raise ValueError("데이터베이스 비밀번호가 환경 변수에 설정되지 않았습니다.")

# 2. SQLAlchemy 엔진 생성 (MySQL 연결 문자열 구성)
# charset=utf8mb4: 한글 및 특수문자 깨짐 방지
db_url = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
engine = create_engine(db_url)

table_name = 'exchange_list'

try:
    # 3. DataFrame 데이터를 MySQL 테이블에 적재
    df.to_sql(name=table_name, con=engine, if_exists='append', index=False)
except Exception as e:
    print(f"적재 중 오류 발생: {e}")
finally:
    # 데이터베이스 엔진 풀(Pool) 반환 및 종료
    engine.dispose()

---

 

4. 추가 고려 사항 (성능 및 안정성 최적화)

대용량 데이터를 다루거나 프로덕션 레벨의 파이프라인을 구축할 때 다음 사항을 반영해야 합니다.

• 대용량 데이터 분할 적재 (chunksize): DataFrame의 크기가 클 경우, 한 번에 DB로 전송하면 메모리 초과나 패킷 크기 제한(MySQL의 max_allowed_packet 등) 오류가 발생할 수 있습니다. to_sql(..., chunksize=1000)과 같이 파라미터를 설정하여 지정된 행(Row) 단위로 쪼개어 적재해야 합니다.
• 명시적 데이터 타입 매핑 (dtype): pandas가 데이터베이스 컬럼의 데이터 타입을 자동으로 추론할 때, 텍스트가 긴 컬럼을 지나치게 큰 타입(예: TEXT)으로 설정하여 DB 성능 저하를 일으킬 수 있습니다. sqlalchemy.types를 임포트 한 뒤, dtype={'컬럼명': sqlalchemy.types.VARCHAR(255)} 형태로 명시적 데이터 타입을 지정하는 것이 안전합니다.