본문 바로가기
Python/Data analysis

데이터 저장

by curious week 2025. 10. 7.

1. 학습개요

  • 데이터 분석은 수집 → 정제/변환 → 저장 → 재사용의 순환입니다.
  • 저장 단계는 단순 보관이 아니라 분석 목적에 맞는 형식으로, 다양한 시스템에서 공유 가능하도록 만드는 관리 과정입니다.
  • 이번 차시는 파일 기반 저장(CSV/JSON/Excel) vs 데이터베이스(DB)의 특징과 선택 기준을 비교하고, pandas DataFrame을 생성·가공·저장하는 방법을 실습합니다.
  • 공공데이터포털(Open API)을 활용해 실제 수집→변환→저장 흐름을 체험합니다.

2. 학습목표 매핑

  1. DataFrame으로 수집 데이터를 처리: 스키마(컬럼/타입) 정리, 결측/이상치 처리, 인덱싱.
  2. DataFrame을 다양한 형식으로 저장: to_csv, to_json, to_excel, to_sql 등.
  3. 수집 과정 문제 인식·해결: 인코딩, 누락·중복, 타임존, API 제한(Rate limit), 재시도 등 대응.

3. 데이터 저장의 이해

3.1 왜 저장이 중요한가

  • 재현성/추적성: 언제/어디서/어떻게 수집했는지 메타데이터와 함께 보관.
  • 공유/연동성: 협업·시스템 간 교환을 위한 범용 포맷 선택.
  • 성능/비용: 크기·접근 패턴(배치 vs 실시간)·보안 요구에 맞춘 저장소 선택.

3.2 저장 방식 개요(선택 기준)

  • 파일(CSV/JSON/Excel): 가볍고 범용. 소규모·교환·보고서에 유리. 대용량/동시성/트랜잭션엔 불리.
  • 데이터베이스(RDBMS/NoSQL): 질의, 인덱스, 동시성, 무결성, 접근 제어에 강점. 초기 설정·운영이 필요.

4. 파일 기반 저장

4.1 CSV (Comma-Separated Values)

  • 장점: 가장 범용, 가벼움, 버전 관리 용이.
  • 주의: 데이터 타입 정보가 없으므로 스키마 정의를 별도 관리(예: YAML/JSON 스키마, 코드).
  • 권장: index=False, encoding="utf-8" 또는 utf-8-sig(Excel 호환), na_rep로 결측 표현.

4.2 JSON (JavaScript Object Notation)

  • 장점: 계층/중첩 구조, 웹/API 표준과의 친화성.
  • 주의: 중첩이 깊으면 용량 증가·가독성 저하. 교환용은 orient="records" 권장, 로그/스트림은 JSON Lines(lines=True).

4.3 Excel (.xlsx)

  • 장점: 실무 호환, 시트/서식/차트.
  • 주의: 대용량에 무겁고, 자동 타입 추론으로 깨질 수 있음(특히 날짜/우편번호). 보고서 산출용에 한정.

5. 데이터베이스 기반 저장

5.1 관계형(RDBMS: SQLite, PostgreSQL 등)

  • 장점: SQL 질의, 인덱스, 조인, 트랜잭션, 무결성 제약.
  • 권장: 스키마(타입/널/PK/FK) 명시, 배치 삽입 시 청크/트랜잭션 사용.

5.2 비관계형(NoSQL: MongoDB 등)

  • 장점: 유연 스키마, 문서/키값/그래프 등 다양한 모델.
  • 주의: 질의/인덱싱 전략을 데이터 액세스 패턴에 맞게 설계.

6. Pandas DataFrame 핵심

  • 구조: 2차원 테이블(행/열, 인덱스). 내부는 NumPy 배열로 고속 연산.
  • 인덱싱: set_index, loc/iloc로 빠른 행 접근.
  • 입력: read_csv, read_json, read_excel, read_sql 등.
  • 저장: to_csv, to_json, to_excel, to_sql, to_parquet(대용량/컬럼지향 권장).

7. 실습: 수집 → 변환 → 저장 (pandas 중심)

아래 코드는 바로 실행 가능하도록 주석과 매개변수 설명을 포함했습니다.

# 요구 패키지:
# pip install pandas requests sqlalchemy openpyxl

import os
import time
import json
import math
import pandas as pd
import requests
from sqlalchemy import create_engine

# ------------------------------
# 1) 공공데이터포털 API 수집 (예: 대기오염 정보; endpoint/params는 가이드에 맞게 조정)
# ------------------------------
API_URL = "https://apis.data.go.kr/B552584/ArpltnInforInqireSvc/getCtprvnRltmMesureDnsty"
SERVICE_KEY = os.getenv("DATA_GO_KR_KEY")  # 환경변수에 API 키 저장 권장
# 매개변수(roles/types):
# - sidoName (str): 시/도명, 예: "서울"
# - returnType (str): 응답 형식, "json" 권장
# - pageNo/numOfRows (int): 페이징
params = {
    "serviceKey": SERVICE_KEY,   # 반드시 URL 인코딩된 키 사용
    "sidoName": "서울",
    "returnType": "json",
    "pageNo": 1,
    "numOfRows": 100
}

def call_api(params: dict, max_retries: int = 3, backoff: float = 1.5) -> dict:
    """API 호출(재시도 포함)
    params: API 쿼리 파라미터
    max_retries: 최대 재시도 횟수
    backoff: 지수 백오프 계수
    """
    for attempt in range(max_retries):
        r = requests.get(API_URL, params=params, timeout=10)
        if r.status_code == 200:
            return r.json()
        time.sleep(backoff ** attempt)
    r.raise_for_status()

raw = call_api(params)

# ------------------------------
# 2) JSON → DataFrame 변환/정제
# ------------------------------
# 응답 구조에 맞춰 경로 지정 (서비스별로 key가 다르므로 문서 확인)
items = raw.get("response", {}).get("body", {}).get("items", [])

df = pd.DataFrame(items)

# 스키마 정리: 컬럼명 표준화, 타입 캐스팅, 결측 처리
df.columns = [c.strip() for c in df.columns]
# 예시: 수치형 캐스팅 (오염도 지표 컬럼이 문자열일 수 있음)
num_cols = [c for c in df.columns if c.endswith("Value") or c.endswith("_val")]
for c in num_cols:
    df[c] = pd.to_numeric(df[c], errors="coerce")

# 날짜/시간 캐스팅(예: dataTime 컬럼)
if "dataTime" in df.columns:
    df["dataTime"] = pd.to_datetime(df["dataTime"], errors="coerce")

# 필수 키 결측 제거(예: 측정소명 stationName가 기본 키 역할)
if "stationName" in df.columns:
    df = df.dropna(subset=["stationName"]).drop_duplicates(subset=["stationName", "dataTime"])

# ------------------------------
# 3) 다양한 형식으로 저장 (파일 + DB)
# ------------------------------
SAVE_DIR = "artifacts"
os.makedirs(SAVE_DIR, exist_ok=True)

# 3-1) CSV
df.to_csv(f"{SAVE_DIR}/air_quality.csv", index=False, encoding="utf-8-sig", na_rep="")

# 3-2) JSON (교환용: records, 저장/적재용: lines)
df.to_json(f"{SAVE_DIR}/air_quality.json", orient="records", force_ascii=False)
df.to_json(f"{SAVE_DIR}/air_quality.jsonl", orient="records", lines=True, force_ascii=False)

# 3-3) Excel (보고용)
with pd.ExcelWriter(f"{SAVE_DIR}/air_quality.xlsx", engine="openpyxl") as writer:
    df.to_excel(writer, sheet_name="data", index=False)

# 3-4) SQLite (로컬 RDBMS; 분석 질의/조인에 유리)
engine = create_engine(f"sqlite:///{SAVE_DIR}/warehouse.db", echo=False, future=True)
# if_exists='replace'는 덮어쓰기; 누적은 'append' 사용
df.to_sql("air_quality", con=engine, if_exists="replace", index=False)

실습 포인트

  • API 키 관리: 환경변수/비밀 저장소(.env) 사용, Git에 올리지 않기.
  • 스키마 계약: 수치/날짜 캐스팅을 코드에 명시(재현성).
  • 파일 저장 규칙: 폴더(artifacts/), 파일명(snake_case), 인코딩/결측 표현 통일.
  • DB 저장: 로컬 분석은 SQLite로 시작 → 협업·운영은 PostgreSQL 등으로 이전.

8. 수집 과정의 흔한 문제와 해결

  • 인코딩 깨짐: encoding="utf-8-sig"/cp949 확인, 가급적 공급자 문서의 인코딩 명시 확인.
  • 타입 혼합: 숫자·문자 혼재 시 to_numeric(..., errors="coerce")로 정규화, 스키마 단언(assert) 추가.
  • 결측/중복: dropna, drop_duplicates + 비즈니스 키(예: (stationName, dataTime) 조합) 정의.
  • 타임존/시각화 오류: tz_localize/convert로 표준화, 저장 시 ISO8601 권장.
  • API Rate limit: 페이지네이션, 지수 백오프, 캐시(파일/DB)로 불필요 호출 억제.
  • 대용량 파일: CSV 대신 Parquet(df.to_parquet) 같은 컬럼 지향 포맷 고려(압축·스키마·속도 우수).

9. 형식 선택 가이드

  • 교환/간단 적재: CSV
  • 계층/웹 서비스 친화: JSON / JSON Lines
  • 보고/비기술 협업: Excel
  • 반복 질의/조인/거버넌스: RDBMS
  • 대규모 분석/속도: Parquet(+데이터 레이크/Spark/Polars 등)

10. 약어 해설

  • API (Application Programming Interface): 소프트웨어 간 상호 작용 규약
  • CSV (Comma-Separated Values): 콤마 구분 텍스트 표 형식
  • JSON (JavaScript Object Notation): 경량 계층 데이터 형식
  • SQL (Structured Query Language): 관계형 DB 질의 언어
  • RDBMS (Relational Database Management System): 관계형 DBMS

오픈소스 기반 데이터 분석 5강 — 데이터 저장

 

Data-Analysis-with-Open-Source/오픈소스_데이터_분석_5강.ipynb at main · mors119/Data-Analysis-with-Open-Source

Data Analysis with Open Source. Contribute to mors119/Data-Analysis-with-Open-Source development by creating an account on GitHub.

github.com

5-1 CSV 형식 저장

# --- DataFrame → CSV 저장 기본 ---
# 역할: 메모리상의 표(DataFrame)를 CSV 텍스트 파일로 저장
# 자주 쓰는 옵션:
# - index: 행 인덱스 저장 여부(False 권장)
# - encoding: 'utf-8', 'cp949'(윈도 엑셀 호환), 'utf-8-sig'(엑셀에서 한글 깨짐 방지)
# - sep: 구분자(기본 ',') → TSV는 '\t'
# - na_rep: 결측치 표기 문자열
# - quoting: 인용부호 처리 (csv.QUOTE_MINIMAL/ALL/NONNUMERIC)
# - compression: 'zip', 'gzip' 등으로 압축 저장
import pandas as pd

data = {
    'student_id': [101, 102, 103, 104, 105],
    'database_score': [85, 76, 92, 63, 88],
    'cloudcomputing_score': [78, 82, 95, 70, 84],
    'python_score': [92, 78, 85, 75, 91],
    'watch_rate': [0.95, 0.87, 0.99, 0.80, 0.93],
}
df = pd.DataFrame(data)

# 오타 수정: 'stduent_analysis.csv' → 'student_analysis.csv'
df.to_csv(
    'student_analysis.csv',
    index=False,
    encoding='utf-8',
    # sep=',',
    # na_rep='',
    # quoting=0,                 # [import csv] → csv.QUOTE_MINIMAL
    # compression='zip',         # 'student_analysis.csv.zip'
)

# 대안:
# - to_csv('파일', encoding='utf-8-sig')  # 엑셀 한글 호환
# - [import pyarrow] 후 Parquet 권장(대용량/속도) → df.to_parquet('file.parquet')

5-2 JSON 형식 저장

# --- 파이썬 객체 → JSON 파일 저장 ---
# 역할: dict/list를 JSON 포맷으로 직렬화하여 저장
# 옵션:
# - ensure_ascii=False: 한글 그대로 저장
# - indent: 보기 좋게 들여쓰기
# - sort_keys=True: 키 정렬
import json
import pandas as pd

data = {
    "이름": "홍길동",
    "나이": 25,
    "거주지": "서울",
    "관심사": ["프로그래밍", "데이터 분석", "여행"],
}

with open('output.json', mode='w', encoding='utf-8') as f:
    json.dump(data, f, indent=4, ensure_ascii=False)

# --- DataFrame → JSON 저장 ---
# 역할: 표를 JSON으로 저장(교환/웹전송에 유용)
# orient에 따라 모양이 완전히 달라짐:
# - 'records': [{...}, {...}] (권장, 표준적)
# - 'columns': {col: {index: value, ...}}
# - 'index': {index: {col: value, ...}}
# - 'split': {"index":[], "columns":[], "data":[...]}
df = pd.DataFrame([data])  # 주의: 단일 dict는 행 1개로 만들려면 리스트로 감싸기
df.to_json(
    'output_df.json',
    orient='records',
    indent=4,
    force_ascii=False,  # ensure_ascii=False와 동일 의미
)

# 대안:
# - jsonlines(JSONL) 형태: df.to_json('file.jsonl', orient='records', lines=True)
# - [import orjson] 고성능 직렬화 라이브러리

5-3 Pandas DataFrame 데이터 추출

# --- DataFrame의 메타와 값 추출 ---
# 역할: 인덱스/컬럼/값 배열을 가져와서 다른 연산에 활용
import pandas as pd

data = {
    "이름": ["김철수", "이영희", "박민수", "최지훈", "정소희"],
    "학년": [1, 2, 3, 4, 2],
    "학점": [4.2, 3.8, 4.5, 3.9, 3.5],
    "학과": ["컴퓨터학", "경영학", "농학", "교육학", "영어영문학"],
    "동아리": ["프로그래밍", "독서토론", "로봇공학", "봉사활동", "음악감상"],
}
df = pd.DataFrame(data)

print(df.index)         # RangeIndex(start=0, stop=5, step=1)
print(df.columns)       # Index([...], dtype='object')

# 값 추출: 넘파이 배열 기반
print(df.values.tolist())  # 파이썬 리스트 중첩 2차원
print(df.values)           # numpy.ndarray
print(df.values.flatten()) # 1차원 평탄화(복사)

# 대안/권장:
# - df.to_numpy()           # dtype 제어 가능
# - df.to_dict('records')   # [{col:val, ...}, ...]
# - df.itertuples(index=False)  # 빠른 행 반복
# - [import numpy] 후 np.ravel(df.to_numpy())  # 평탄화(뷰/복사 차이 인지)

5-4 DataFrame 생성 방법

# --- 다양한 소스 → DataFrame 생성 ---
# 역할: dict, CSV, JSON 등 여러 입력으로부터 표를 만든다.
import pandas as pd

data = {
    "이름": ["김철수", "이영희", "박민수", "최지훈", "정소희"],
    "학년": [1, 2, 3, 4, 2],
    "학점": [4.2, 3.8, 4.5, 3.9, 3.5],
    "학과": ["컴퓨터공학", "경영학", "전자공학", "의학", "심리학"],
    "동아리": ["프로그래밍", "독서토론", "로봇공학", "봉사활동", "음악감상"],
}

df1 = pd.DataFrame.from_dict(data)   # 또는 pd.DataFrame(data)

# 외부 파일에서 읽기(동일한 5x5로 가정)
df2 = pd.read_csv('students.csv')    # sep/encoding/parse_dates 등 옵션 활용
df3 = pd.read_json('students.json')  # orient/lines 확인 필요

print(df1.shape)  # (5, 5)
print(df2.shape)  # (5, 5)
print(df3.shape)  # (5, 5)

# 대안:
# - [import pyarrow] CSV→Arrow→Parquet 파이프라인(대용량/속도)
# - [import sqlite3] SQL 쿼리로 생성: pd.read_sql_query('SELECT ...', conn)
# - [import openpyxl] 엑셀에서 읽기: pd.read_excel('students.xlsx', engine='openpyxl')

5-5 DataFrame 저장 방법

# --- 여러 형식으로 저장 ---
# 역할: 결과물을 다양한 채널(CSV/JSON/HTML 등)로 내보내기
import pandas as pd

data = {
    "이름": ["김철수", "이영희", "박민수", "최지훈", "정소희"],
    "학년": [1, 2, 3, 4, 2],
    "학점": [4.2, 3.8, 4.5, 3.9, 3.5],
    "학과": ["컴퓨터공학", "경영학", "전자공학", "의학", "심리학"],
    "동아리": ["프로그래밍", "독서토론", "로봇공학", "봉사활동", "음악감상"],
}
df = pd.DataFrame(data)

# CSV/JSON
df.to_csv('students_scores.csv', index=False, encoding='utf-8')
df.to_json('students_scores.json', orient='records', force_ascii=False, indent=2)

# HTML(간단한 테이블)
html_table = df.to_html(
    index=False,
    # classes=['table','table-striped'],  # 스타일 클래스(렌더링 환경 필요)
    # border=0,
    # escape=True,                        # HTML 이스케이프 처리
)
with open('students_scores.html', 'w', encoding='utf-8') as f:
    f.write(html_table)

# 대안:
# - Excel: df.to_excel('file.xlsx', index=False)  # [import openpyxl] 또는 [import xlsxwriter]
# - Parquet: df.to_parquet('file.parquet')        # [import pyarrow] 또는 [import fastparquet]
# - Feather: df.to_feather('file.feather')        # [import pyarrow] (속도↑, 파이썬↔R 호환)

5-6 실습 시나리오: 데이터 수집 → DataFrame 생성 → 다양한 형식 저장

# --- 1) 공공 API에서 JSON 수집(서울시 평균대기질 예시) ---
# 역할: REST API 호출 → JSON 구조 확인 → 필요한 배열('row')만 추출
# 주의:
# - API Key는 코드에 하드코딩하지 말고 환경변수/비밀관리 사용
# - 응답 스키마가 바뀔 수 있으므로 KeyError 대비
import os
import requests

api_key = os.getenv('SEOUL_API_KEY', 'YOUR_API_KEY')  # 데모: 환경변수 우선
date = '20130615'
district = '종로구'
url = f"http://openAPI.seoul.go.kr:8088/{api_key}/json/TimeAverageAirQuality/1/5/{date}/{district}"

try:
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    payload = resp.json()
    print(url)
    print(payload)  # 구조 먼저 확인
except requests.exceptions.RequestException as e:
    raise SystemExit(f"API 호출 실패: {e}")

# --- 2) JSON → DataFrame 구성 ---
import pandas as pd

try:
    data = payload['TimeAverageAirQuality']['row']  # 리스트
except (KeyError, TypeError):
    raise SystemExit("예상한 키가 없습니다. 응답 스키마를 확인하세요.")

print(type(data), len(data), data[0])  # 구조 점검
df = pd.DataFrame(data)

# 옵션: 자료형/파싱 정리(예: 날짜 컬럼 문자열 → datetime)
# MSRDT 예시: '201306152300' → YYYYmmddHHMM
df['MSRDT'] = pd.to_datetime(df['MSRDT'], format='%Y%m%d%H%M', errors='coerce')

# 대안:
# - [import pandas] json_normalize로 평탄화
#   from pandas import json_normalize
#   df = json_normalize(payload['TimeAverageAirQuality']['row'])

# --- 3) 다양한 형식으로 저장 ---
# CSV
df.to_csv('air_pollution.csv', index=False, encoding='utf-8')

# JSON
df.to_json('air_pollution.json', orient='records', indent=2, force_ascii=False)

# Excel
# [import openpyxl] (기본 엔진) 또는 [import xlsxwriter]
df.to_excel('air_pollution.xlsx', index=False)

# HTML
df.to_html('air_pollution.html', index=False)

# 대안/팁:
# - Parquet(권장): df.to_parquet('air_pollution.parquet')  # [import pyarrow]
# - 압축: df.to_csv('air_pollution.csv.gz', index=False, compression='gzip')
# - 컬럼명/순서 정리: df = df[['MSRDT','MSRSTE_NM','NO2','O3','CO','SO2','PM10','PM25']]
# - 결측치 처리: df.fillna({'PM25': 0}, inplace=True)
저작자표시 비영리 동일조건 (새창열림)

'Python > Data analysis' 카테고리의 다른 글

데이터 전처리 2  (0) 2025.10.10
데이터 전처리 1  (0) 2025.10.10
데이터 수집  (0) 2025.10.07
언패킹, 예외처리, 함수형 프로그래밍  (0) 2025.10.03
시퀀스 슬라이싱과 컴프리헨션, 문자열 형식 지정, 컨텍스트 관리  (1) 2025.10.03

관련글

티스토리툴바