[Python] ETL (Extract, Transform, Load)

ETL(Extract, Transform, Load) 은 데이터 처리의 핵심 과정으로, 데이터를 다양한 소스에서 추출하고, 변환하여 최종적으로 저장소에 로드하는 방법을 설명합니다.

1. ETL의 프로세스

ETL 과정은 데이터를 효과적으로 처리하고 분석하기 위해 필수적인 단계입니다.
추출(Extract) 단계에서는 데이터를 소스에서 가져오고, 변환(Transform) 단계에서는 데이터를 분석 및 보고서 작성을 위한 형식으로 가공하며, 로드(Load) 단계에서는 변환된 데이터를 최종 저장소에 저장합니다.

1. 추출 (Extract)

추출(Extract) 단계는 데이터를 다양한 소스 시스템에서 가져오는 과정입니다. 데이터 소스는 데이터베이스, 파일, API 등 다양할 수 있으며, 이 단계의 목표는 원본 데이터에서 필요한 정보를 수집하는 것입니다.

# 샘플 코드

import csv

# CSV 파일에서 데이터 추출
with open('month_temp.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    for line in rdr:
        print(line)

2. 변환 (Transform)

변환(Transform) 단계는 추출된 데이터를 분석이나 보고서 작성에 적합한 형태로 가공하는 과정입니다. 이 단계에서 데이터 정제, 필터링, 집계, 계산 등이 이루어집니다.

# 예제 1: 특정 날짜의 기온 출력

import csv

# 10월 20일 기온 출력
with open('month_temp.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    for line in rdr:
        if '2019-10-20' in line:
            print(line[2])

# 예제 2: 10월 1일부터 10일까지 평균 기온 계산

import csv

# 10-01 ~ 10-10일까지 10일간 평균기온의 평균 출력
with open('month_temp.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    next(rdr)  # 헤더 건너뛰기

    temp_sum = 0
    count = 0

    for line in rdr:
        date = line[1]
        if '2019-10-01' <= date <= '2019-10-10':
            temp_sum += float(line[2])
            count += 1

    avg = round(temp_sum / count, 2)
    print(f'10일간의 평균기온= {avg}')

# 예제 3: 일교차 계산

import csv

# 일자별 기온 일교차 (최고기온 - 최저기온) 출력
with open('month_temp.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    next(rdr)  # 헤더 건너뛰기

    for line in rdr:
        max_temp = float(line[4])
        min_temp = float(line[3])
        gap = round(max_temp - min_temp, 2)
        date = line[1]
        print(f'일시: {date}, 일교차: {gap}')

3. 로드 (Load)

로드(Load) 단계는 변환된 데이터를 최종 저장소에 저장하는 과정입니다. 저장소는 데이터베이스, 데이터 웨어하우스, 파일 시스템 등 다양할 수 있습니다.

# 샘플 코드

import json

# 변환된 데이터를 JSON 파일에 저장
user = {
    "name": "홍길동",
    "id": 152352,
    "history": [
        {"date": "2015-03-11", "item": "iPhone"},
        {"date": "2016-02-23", "item": "Monitor"}
    ]
}

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

2. 약국 데이터 검색 예제

1. 특정 약국의 주소 검색

이 코드는 pharm_2019.csv 파일에서 신대원약국의 주소를 찾습니다.

import csv

# 경주시 신대원약국의 주소 검색
with open('pharm_2019.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    next(rdr)  # 헤더 건너뛰기

    tgt_pharm = '신대원약국'
    tgt_city = '경주시'
    for line in rdr:
        if line[0] == tgt_pharm and line[1] == tgt_city:
            address = line[2]
            print(f'{tgt_pharm}의 주소 : {address}')

2. 특정 건물에 입점된 약국 검색

이 코드는 용인시 수지구의 로얄스포츠 건물에 입점된 약국의 이름을 출력합니다.

import csv

# 용인시 수지구 로얄스포츠 건물에 입점된 약국 검색
with open('pharm_2019.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    next(rdr)  # 헤더 건너뛰기

    for line in rdr:
        if '로얄스포츠' in line[2] and '용인시 수지구' in line[2]:
            print(line[0])

3. 경상북도의 2010년 약국 목록 출력

이 코드는 경상북도에 위치하고 2010년에 개설된 약국 목록을 출력합니다.

import csv

# 경상북도 2010년 1년간 개설 약국 목록 출력
with open('pharm_2019.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    header = next(rdr)  # 헤더 추출
    print('번호', header[0], header[1], header[2], header[4], header[5], sep=',')
    
    count = 1
    for line in rdr:
        if '경상북도' in line[2] and (int(line[3]) >= 20100101 and int(line[3]) <= 20101231):
            print(f'{count}. {line[0]}/{line[1]}/{line[2]}/{line[4]}/{line[5]}')
            count += 1

4. 최근 5년 이내 개설된 용인시 약국 수

이 코드는 최근 5년 이내에 개설된 용인시 약국의 수를 계산하고 출력합니다.

import csv
from datetime import datetime
from dateutil.relativedelta import relativedelta

today = datetime.now()  # 오늘 날짜
before_date = today - relativedelta(years=5)  # 5년 전 날짜

with open('pharm_2019.csv', 'r', encoding='utf-8') as f:
    rdr = csv.reader(f)
    next(rdr)  # 헤더 건너뛰기

    count = 0
    recent = 0
    for line in rdr:
        address = line[2]
        open_date = datetime.strptime(line[3], '%Y%m%d')  # 개업일 서식 맞추기
        if '용인시' in address:
            count += 1
            if before_date <= open_date <= today:
                recent += 1

    print(f'용인시 약국 수: {count}개')
    print(f'최근 용인시 5년 이내 개설 약국 수: {recent}개')