지난번 예제 코드에 이어서 다른 예제 코드들을 학습해 볼 것이다. 지금까지 배원던 것들을 토대로 코드가 작성되었으며, 주석을 통해 코드 설명을 해놓았기에 코드 흐름을 따라가며 학습하면 도움이 될 것이라고 생각한다. 실습하는 예제 코드가 공부했던 이론의 내용보다 어렵다고 느껴졌지만 Chat GPT와 같은 툴을 사용하고, 구글 같은 검색 엔진을 통해 정보를 얻는다면 그리 어렵지 않고, 더 도움이 될 수 있을 것이라 생각해서 이런 방식으로 학습을 진행하였다.
Q1. 다양한 형식의 파일을 처리할 수 있는 파일 처리기를 구현하시오.
# 과제: 파일 처리기 구현
#
# 다양한 유형의 파일(텍스트, CSV, JSON, 바이너리)을 읽고 쓸 수 있어야 합니다
# 파일이 존재하지 않거나, 권한이 없거나, 형식이 잘못된 경우 등 다양한 오류 상황을 적절히 처리
# 사용자 정의 예외 계층 구조를 설계하고 구현
# 오류 발생 시 로깅을 통해 문제를 기록
# 모든 파일 작업은 컨텍스트 매니저(`with` 구문)를 사용
import os
import csv
import logging
import json
# Logging 구현
logging.basicConfig(
filename='app.log', # 로그를 기록할 파일 이름 설정
level=logging.DEBUG, # DEBUG를 기준으로 로그 기록
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", # 로그 메시지 출력 형식 지정
encoding='utf-8' # 로그 파일 인코더 한글이 포함 가능하도록 설정
)
# 사용자 정의 예외 계층
class FileError(Exception): # 모든 사용자 정의 예외의 부모 클래스
"""모든 파일 처리 관련 예외의 기반 클래스""" # 이후 예외들이 해당 클래스를 상속받기에, except FileError 하나로 모든 파일 관련 예외를 처리할 수 있음
pass
class FileNotFound(FileError): # 파일이 존재하지 않을 때 발생하는 사용자 정의 예외
"""파일이 존재하지 않을 때 발생하는 예외"""
pass
class FilePermissionError(FileError): # 파일에 접근 권한이 없을 때 발생하는 사용자 정의 예외
"""파일의 권한의 문제가 있을 때 발생하는 예외"""
pass
class FileFomatError(FileError): # 파일의 형식이 지원하지 않는 형식일 때 발생하는 사용자 정의 예외
"""파일의 형식이 지원하지 않을 때 발생하는 예외"""
pass
# 파일 처리기
class FileHandler: # 다양한 파일 타입을 읽고 쓸 수 있는 범용 파일 처리 클래스
# 생성자 함수
def __init__(self, filepath, mode, filetype="text"):
self.filepath = filepath # 읽고 쓰기 위한 파일의 경로
self.mode = mode # open할 때 사용할 모드 ('r', 'w', 'rb', 'wb' 등)
self.filetype = filetype # 파일 타입
self.file = None # 파일 객체를 저장할 변수
# with 구문(컨텍스트 매니저) 진입 시 실행됨
# 파일을 열고, 파일 객체를 self.file에 저장하고, 자기 자신(self)를 반환
def __enter__(self): # 입력 함수
"""with 구문 진입 시 실행되는 메서드"""
try:
self.file = open( # open 함수로 파일을 열기
self.filepath, # 파일 경로
self.mode, # 사용할 모드
# 텍스트, JSON, CSV 파일을 읽기 위한 인코더 지정 / binary 파일의 경우 인코더가 없어야 함
encoding="utf-8" if self.filetype in ['text', 'json', 'csv'] else None
)
return self
except FileNotFoundError: # open() 중에 파일이 존재하지 않는 경우 예외 처리 및 로깅
logging.error(f"파일을 찾을 수 없습니다. {self.filepath}")
raise FileNotFound
except PermissionError: # 파일 접근 권한이 없을 경우 예외 처리 및 로깅
logging.error(f"파일 접근 권한 오류입니다. {self.filepath}")
raise FilePermissionError
# with 구문(컨텍스트 매니저) 종료 시 실행
def __exit__(self, exc_type, exc_value, traceback): # Python에서는 __exit__() 메서드는 반드시 3개의 인지를 받도록 되어있음 (실행은 안될 수 O)
"""with 구문 종료시 실행되는 메서드"""
if self.file:
self.file.close() # 열린 파일이 있다면 닫음
# 파일 읽기 함수
def read(self):
"""파일 읽기 가능 / 다양한 형식의 파일에 맞게 적절한 방식으로 읽기"""
try:
if self.filetype == "text": # 일반 텍스트 파일이라면
return self.file.read() # file.read()로 전체를 읽음
elif self.filetype == "json": # 만약 JSON 파일이라면
return json.load(self.file) # json.load(file)을 사용해 Python 객체로 변환함
elif self.filetype == "csv": # 만약 CSV 파일이라면
return list(csv.reader(self.file)) # csv.reader(file)로 읽고 list로 변환해서 반환
elif self.filetype == "binary": # 이진 파일 이라면
return self.file.read() # .read()로 바이트 단위로 읽음
else: # 위의 형식들이 아니라면 예외 처리
raise FileFomatError(f"지원하지 않는 파일 형식입니다: {self.filetype}")
except Exception as e: # 예외가 발생했을 때 로그 기록 후 예외 전파
logging.error(f"파일 읽기 실패: {e}")
raise
# 파일 쓰기 함수:
# 파이썬에서 다룰 수 있는 데이터를 해당 파일 형식에 맞는 실제 파일 형태로 변환해서 저장하는 기능을 수행하는 함수
def write(self, data):
"""파일 쓰기 가능 / 다양한 형식의 파일에 맞게 적절한 방식으로 입력 """
try:
if self.filetype == "text": # 텍스트라면 문자열을 텍스트 파일에 직접 작성
self.file.write(data)
elif self.filetype == "json": # Python 객체를 JSON 형식으로 변환 후 저장
json.dump(data, self.file, ensure_ascii=False, indent=2) # ensure_ascii=False: 한글 깨짐 방지 / indent=2: 들여쓰기 사용
elif self.filetype == "csv": # 리스트 형태 데이터를 CSV로 저장
writer = csv.writer(self.file)
writer.writerows(data)
elif self.filetype == "binary": # 이진 데이터는 그대로 저장
self.file.write(data)
else: # 위와 동일
raise FileFomatError(f"지원하지 않는 파일 형식입니다: {self.filetype}")
except Exception as e: # 위와 동일
logging.error(f"파일 쓰기 실패: {e}")
raise
→ 해당 코드를 작성하고 이해하며 의문이 들고, 궁금했던 부분은 학습했던 예외처리와 파일 입출력 부분이 아니라 메서드에 관련된 내용이었다. 메서드를 학습하긴 했었지만 자세히 이해하고 넘어가지 못했던 것 같다. 해당 코드에 일반 메서드와 특수 메서드가 사용되었는데 해당 부분 내용이 잘 이해되지 않아서 다시 정리하며 이해해보았다.
일반 메서드와 특수 메서드에 대한 개념 정리를 해보았다. 일반 메서드는 클래스 내에 사용자가 정의한 일반적인 함수로 우리가 일반적으로 알고 있는 함수로 생각하면 될 것 같다.
특수 메서드는 Python 내부에서 특정 동작을 자동으로 수행하기 위해 사용되는 약속된 이름의 메서드로, 직접 호출하기보다 Python이 자동으로 호출하는 프로세스를 가지고 있다. 여기서 Python이 자동으로 호출한다고 하는 말이 어떤 말인지 이해가 잘 되지 않아 조금 더 자세히 살펴보았는데, 이 말의 의미는 내가 직접 호출하지 않아도 특정 상황이 되면 Python이 알아서 호출한다는 의미라고 한다.
예를 들어, with FileHandler("data.txt", "r") as f: 코드가 있고 with 구문 내에 context = f.read()로 작성되어 있다면, 우리는 with 구문을 작성해서 사용했지만 __enter()__를 호출할 적이 없는 것이다. 이는 Python이 알아서 __enter()__와 __exit()__ 같은 특수 메서드들을 호출하는 것이라고 볼 수 있다.
이렇게 파일 입출력에 관련된 개념을 공부하는 예제 코드를 통해 여러 개념을 학습해보았다. 다음은 그 뒤의 내용인 이터레이터와 제너레이터, 비동기 처리까지 학습할 수 있는 예제 코드를 보며 학습해 볼 것이다.
Q2. 로그 파일을 한 줄씩 읽는 제너레이터 함수를 작성하시오.
# 로그 파일을 한 줄씩 읽는 제너레이터 함수 작성
# 특정 패턴(예: 'ERROR', 'WARNING' 등)이 포함된 줄만 필터링하는 제너레이터 작성
def read_log_lines(filepath):
"""로그 파일을 한 줄씩 읽어들이는 제너레이터 함수"""
try:
# open()을 통해 파일을 '읽기 모드'로 UTF-8 인코딩을 사용해 연다.
with open(filepath, 'r', encoding='utf-8') as f:
# 파일의 각 줄에 대해 반복
for line in f: # for 문을 통해 한줄 단위로 읽는다.
yield line.strip() # yield를 통해 한 줄 씩 반환 / 줄 끝 개행 제거하고 한 줄씩 반환
except FileNotFoundError:
# 파일이 존재하지 않을 경우 예외 처리
print(f"파일을 찾을 수 없습니다: {filepath}")
except Exception as e:
# 다른 예외가 발생했을 경우 에러 메시지 출력
print(f"예외 발생: {e}")
# log_lines: 로그 텍스트가 한 줄 씩 들어오는 이터러블
# patterns: 필터링할 문자열 패턴들의 튜플
def filter_log_by_level(log_lines, patterns=('ERROR', 'WARNING')): # 주어진 로그 줄 들 중
"""특정 패턴이 포함된 로그 줄만 필터링하는 제너레이터"""
for line in log_lines: # log_lines는 제너레이터 또는 리스트도 가능
# 패턴 중 하나라도 현재 줄에 포함되어 있다면
if any(pat in line for pat in patterns):
yield line # 그 줄을 반환
→ 위 예제 코드를 통해 제너레이터와 이터레이터를 사용하는 방법을 익혀보았다. 사실 제너레이터와 이터레이터가 무엇인지는 알겠는데 이걸 왜 사용해야 하는지를 모르고 있었던 것 같은데 이 예제 코드를 작성하고 학습하며, 궁금한 부분들을 해결하면서 알게 되었던 것 같다.
이터레이터와 제너레이터를 왜 사용하는지가 궁금해서 찾아보았다. 예를 들어, 만약 대용량의 로그 파일이 있다고 할 때, 제너레이터를 사용하지 않고 다른 함수나 반복문 등을 사용해서 읽는다면 메모리 상의 문제가 발생할 수 있는 가능성이 있다. 이를 통해 일단 효율적인 처리를 위해 제너레이터를 사용한다고 할 수 있을 것 같다.
그리고 이터레이터와 제너레이터가 둘 다 있는 이유가 무엇인지 궁금했다. 이터레이터만으로 구현해도 상관은 없지만 제너레이터를 사용해서 구현하는 것이 훨 간단한 것을 볼 수 있다. 예를 들어, __iter()__나 __next()__를 직접 작성하지 않아도 되기에 간편하게 사용할 수 있다.
Q3. 동시성과 병렬처리: API에 GET 요청을 보낼 때, 세 가지 방식으로 이를 구현하고 성능을 비교하시오.
# 5개의 공개 API에 GET 요청을 보냄
# 세 가지 방식으로 구현하고 성능을 비교
# - 순차 처리
# - ThreadPoolExecutor
# - asyncio와 aiohttp
import time
import requests
import aiohttp
import asyncio
from concurrent.futures import ThreadPoolExecutor
# 테스트할 API 목록
API_URLS = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
"https://jsonplaceholder.typicode.com/posts/4",
"https://jsonplaceholder.typicode.com/posts/5",
]
# 모든 URL에 대해 순차적으로 requests.get으로 HTTP 요청을 보내는 함수 / 하나의 요청이 끝나면 다음 요청 수행
def fetch_sequential(urls):
results = [] # 결과 저장 리스트 생성
for url in urls: # URL을 하나씩 순회하면서 하나씩 처리
response = requests.get(url) # 현재 URL에 대해 동기 방식의 HTTP GET 요청을 보냄 (블로킹 방식)
results.append(response.json()) # 응답 객체에서 JSON 데이터를 추출해 리스트에 추가
return results # 결과 반환
# 병렬 처리할 때 각 스레드에서 실행될 단일 URL 요청 함수
def fetch_single_url(url):
response = requests.get(url) # 주어진 URL에 대해 동기 방식으로 GET 요청을 보냄
return response.json() # 응답에서 JSON을 추출해 반환
# ThreadPoolExecutor를 사용해 여러 URL을 병렬로 요청하는 함수
def fetch_threadpool(urls):
"""병렬 작업은 내부적으로 스레드가 나누어 실행"""
results = [] # 결과를 저장할 리스트
with ThreadPoolExecutor() as executor: # ThreadPoolExecutor 컨텍스트 생성 (자동으로 스레드 풀을 관리)
# executor.map(): URL 리스트의 각 원소에 대해 fetch_single_url을 병렬로 실행
# 결과는 입력 순서대로 반환
for data in executor.map(fetch_single_url, urls):
results.append(data) # 각 요청 결과를 리스트에 저장
return results
# aiohttp의 비동기 세션을 이용한 단일 URL을 호출하는 함수 / await을 통해 비동기적으로 응답 대기
async def fetch_async(session, url):
# aiohttp의 세션을 사용해 비동기 GET 요청을 보냄
async with session.get(url) as response:
return await response.json() # 응답 데이터를 JSON으로 파싱해 반환 (await으로 비동기 대기)
# aiohttp를 사용해 여러 요청을 동시에 처리하는 함수
async def fetch_asyncio(urls):
# aiohttp 클라이언트 세션 생성 (세션 재사용으로 성능 최적화)
async with aiohttp.ClientSession() as session:
# 각 URL에 대해 fetch_async 호출 -> Task 리스트로 생성
tasks = [fetch_async(session, url) for url in urls]
# asyncio.gather(): 모든 task들을 동시에 실행하고 결과를 리스트로 반환
results = await asyncio.gather(*tasks)
return results
# 실행해서 성능을 측정하기 위한 코드
if __name__ == "__main__":
print("▶ 순차 처리")
start = time.time()
fetch_sequential(API_URLS)
print(f"⏱ 실행 시간: {time.time() - start:.2f}초\n")
print("▶ ThreadPoolExecutor 처리")
start = time.time()
fetch_threadpool(API_URLS)
print(f"⏱ 실행 시간: {time.time() - start:.2f}초\n")
print("▶ asyncio + aiohttp 처리")
start = time.time()
asyncio.run(fetch_asyncio(API_URLS))
print(f"⏱ 실행 시간: {time.time() - start:.2f}초\n")
→ 위의 동시성과 병렬처리 관련 예제 코드를 작성하며, 어떻게 처리되는지를 익힐 수 있었다. 예제 코드에 나온 세 가지 방법을 비교해보아야 했는데 세 방식에 어떤 차이가 존재하길래 이런 시간 상의 차이가 발생하는지 궁금해서 간단히 세 가지 방식을 알아보았다. 이번 예제 코드는 결과도 함께 보는 것이 더 낫다고 생각되어 출력된 결과도 함께 첨부해 보았다.
먼저 순차처리(Blocking)은 모든 요청이 완료될 때까지 기다렸다가 하나의 요청이 완료된 후에 다른 요청이 시작되는 프로세스를 가지고 있다.
그리고 ThreadPoolExecutor(멀티스레딩 + Blocking)은 각 요청을 스레드로 나누어 동시에 실행한다. 이렇게되면 응답은 동기적으로 대기하고, 스레드가 병렬로 처리하는 프로세스를 가지게 된다.
세 번째인 asyncio + aiohttp(비동기 + nonBlocking I/O)는 await을 이용해 비동기적으로 처리하는데, 이는 하나의 이벤트 루프에서 모든 요청을 동시에 시작하고, 도착하는 대로 처리하는 프로세스를 가진다.
이렇듯 왜 시간의 차이가 발생하는지 알아볼 수 있었고, 코드를 작성해 보며 조금 더 이해해 볼 수 있었던 것 같다.
다음은 파이썬을 이용해 머신러닝과 딥러닝을 학습하기 전 데이터 전처리 관련 내용에 대한 학습을 시작해 볼 예정이다. 해당 내용들이 전공에서 이미 학습했던 내용이었기에 복습하는 느낌으로 진행해보려 한다.
본 후기는 [카카오엔터프라이즈x스나이퍼팩토리] 카카오클라우드로 배우는 AIaaS 마스터 클래스 (B-log) 리뷰로 작성 되었습니다.
