[천재교육] 프로젝트 기반 빅데이터 서비스 개발자 양성 과정 9기
학습일 : 2024.10.29
📕 학습 목록
- 기능별 파일로 모듈을 분리하여 저장 → 필요 시 해당 모듈을 호출
- DB에 가상 데이터 저장
- 시각화
- 파이썬 심화
📗 기억할 내용
[데이터 파이프라인]
- 데이터의 생성에서 가공, 데이터베이스 전송, 시각화까지
- 가상 데이터 유입: 처음 단계에서 가상 데이터(Fake Data)가 생성되어 데이터베이스로 유입됨
- PostgreSQL에 데이터 저장: 생성된 가상 데이터는 PostgreSQL 데이터베이스에 저장
- 데이터 가공: Python을 사용하여 데이터 가공 작업을 수행합니다. 이 단계에서 필요한 데이터 변환이나 전처리가 이루어짐
- MySQL로 데이터 이행: 가공된 데이터는 MySQL 데이터베이스로 전송되어 저장됨
- 데이터 읽기: MySQL에 저장된 데이터는 Streamlit을 통해 읽어옴
- 시각화: 마지막으로 Streamlit에서 데이터를 시각화하여 사용자에게 데이터를 이해하기 쉽게 제공함
📘코드 실습
* 실습 내용
1. postgresql DB에 Fake 유저 정보 저장 (Batch Data)
2. Pipeline 모듈을 통한 데이터 이행 및 데이터 가공
3. POSTGRESQL DB에 가공된 datamart 테이블 저장
4. Streamlit 라이브러리를 통한 POSTGRESQL 테이블 데이터 시각화
5. n초 마다 Batch성 이행을 통해 시각화 테이블 업데이트 내용 확인
[모듈 생성 & 호출]
1. 모듈을 파일 형태로 나누어 관리
1) 디렉토리 구조 및 역할
[각 파일의 역할]
1. 최상위 파일 및 폴더
- .env: 데이터베이스나 API 키와 같은 민감한 환경 변수를 저장
- app.py: 애플리케이션의 진입점 파일로, 서버를 실행하거나 특정 기능을 실행
- controller.py: ETL 파이프라인을 관리하는 컨트롤러 파일로, 데이터를 추출, 변환, 로드하고 임시 파일을 정리
- .ipynb: Jupyter Notebook 파일로, 분석, 테스트 또는 실험 코드가 포함되어 있음
- main.py: 애플리케이션의 메인 실행 파일로, 초기 설정과 함께 전체 워크플로우를 관리
- settings.py: 애플리케이션의 설정 파일로, 데이터베이스 연결 정보, 파일 경로와 같은 주요 설정 저장
2. dataset: 데이터셋 파일을 저장하는 폴더로, 분석이나 모델 학습을 위한 데이터를 보관
3. db: 데이터베이스 관련 파일을 모아놓은 폴더
- connector.py: 데이터베이스 연결 객체를 생성. DB에 연결하는 기능을 제공
- pgsql_query.py: PostgreSQL 데이터베이스에 사용되는 쿼리들을 저장
4. fakedata: 가상 데이터를 생성하고 삽입하는 기능을 제공하는 폴더
- create.py: 가상 데이터를 생성하는 로직을 포함
- insert.py: 생성된 가상 데이터를 데이터베이스에 삽입
- process.py: 가상 데이터를 처리하는 데 필요한 추가적인 로직을 포함
5. pipeline: ETL 파이프라인을 구성하는 각 단계의 파일을 포함
- extract.py: 데이터베이스로부터 데이터를 추출
- load.py: 변환된 데이터를 데이터베이스에 로드
- remove.py: ETL 작업 후 임시 파일이나 불필요한 데이터를 삭제하는 기능을 수행
- transform.py: 데이터를 변환하는 로직을 포함
6. temp_storage: 임시 파일을 저장하는 폴더로, ETL 과정에서 사용되는 임시 데이터를 보관
2) 파일 생성
① controller.py
[tip] controller : 아래 기능들을 순차적으로 실행
1. DBconnector >> DB Connector 생성
2. postgresql_query >> queries 에서 테이블 이름 목록(table_list) 받아오기
ex) for tbl in table_list:
3. extract >> DB 조회 후 DataFrame 형태로 변환
4. transform >> 저장 경로 생성 후 임시 저장 디렉토리 아래에 dataframe 저장
5. load >> 저장소에 dataframe 파일 저장
6. remove >> 저장이 끝난 후 임시 저장 디렉토리 삭제
# controller.py
from db.connector import DBconnector
from db.pgsql_query import queries
from settings import DB_SETTINGS, TEMP_PATH
from pipeline.extract import extractor
from pipeline.transform import transformer
from pipeline.load import loader
from pipeline.remove import remover
from datetime import datetime
def controller(batch_date):
"""
ETL (Extract, Transform, Load) 파이프라인의 컨트롤러 함수.
주어진 날짜를 기준으로 데이터베이스의 데이터를 추출, 변환, 로드하고
임시 파일을 정리함
"""
# queries에 있는 키 값(테이블 이름)을 리스트로 변환하여 반복 작업에 사용
table_list = list(queries.keys())
# DB 연결 객체를 한 번만 생성 (반복적으로 연결을 생성하지 않음으로써 효율성 향상)
db_obj = DBconnector(**DB_SETTINGS["POSTGRES"])
# 테이블 목록을 순회하면서 각 테이블에 대해 데이터를 처리
for table_name in table_list:
# extractor 함수를 통해 해당 테이블의 데이터를 pandas DataFrame으로 추출
pandas_df = extractor(db_obj, table_name)
# 데이터가 존재하는 경우에만 변환 및 로드 작업을 수행
if len(pandas_df) > 0:
# transformer 함수를 통해 데이터를 변환하고 임시 저장 경로에 파일을 생성
res = transformer(TEMP_PATH, batch_date, pandas_df, table_name)
# 변환 결과가 존재하고 빈 데이터가 아닌 경우에만 데이터 로드 진행
if res is not None and not res.empty:
loader(db_obj, res, table_name) # pandas_df 대신 변환된 res를 로드합니다.
# 모든 작업이 완료되면 임시 파일을 삭제하여 디스크 공간 정리
remover(TEMP_PATH)
# 메인 스크립트로 실행될 때의 처리
if __name__ == "__main__":
# 현재 날짜와 시간을 가져와 "YYYYMMDD" 형식으로 변환하여 batch_date에 저장
_date = datetime.now()
batch_date = _date.strftime("%Y%m%d")
# controller 함수를 호출하여 ETL 작업을 실행
controller(batch_date)
② fakedata>create.py
# create.py
from faker import Faker # Faker 라이브러리 - 가상 데이터를 생성하기 위해 사용
import shortuuid # shortuuid 라이브러리 - 짧은 UUID를 생성하기 위해 사용
import pandas as pd # Pandas 라이브러리 - 데이터프레임을 사용하여 데이터 조작
def create_fakedataframe(count: int) -> pd.DataFrame: # 가상 사용자 데이터를 생성하여 Pandas DataFrame으로 반환하는 함수
# count 수만큼 가상 사용자 데이터를 생성하고 리스트에 저장
fake_data_list = [create_fakeuser() for _ in range(count)]
# 리스트를 데이터프레임으로 변환하여 반환
return pd.DataFrame(fake_data_list)
def create_fakeuser() -> dict: # 단일 가상 사용자를 생성하여 딕셔너리 형태로 반환하는 함수
fake = Faker("ko_KR") # 한국어 기반의 가상 데이터 생성을 위해 Faker 인스턴스 생성
fake_profile = fake.profile() # Faker 라이브러리를 사용해 가상 프로필 생성
key_list = ["name", "ssn", "job", "residence", "blood_group", "sex", "birthdate"] # 필요한 필드 목록
fake_dict = dict() # 가상 사용자 정보를 저장할 빈 딕셔너리 생성
# 필요한 필드를 key_list에서 가져와 fake_dict에 추가
for key in key_list:
fake_dict[key] = fake_profile[key]
fake_dict["uuid"] = shortuuid.uuid() # 사용자에게 고유한 UUID 생성하여 추가
fake_dict["birthdate"] = fake_dict["birthdate"].strftime("%Y%m%d") # birthdate를 'YYYYMMDD' 형식으로 변환
return fake_dict # 가상 사용자 정보를 담은 딕셔너리 반환
③ fakedata>insert.py
# insert.py
import pandas as pd # Pandas 라이브러리 - 데이터프레임을 사용하여 데이터 조작
from db.connector import DBconnector # DBconnector 클래스 - 데이터베이스 연결 객체를 생성하는 클래스
from settings import DB_SETTINGS # DB 설정 정보가 담긴 파일
def insert_fakedataframe(df: pd.DataFrame) -> bool: # 가상의 데이터를 담은 Pandas DataFrame을 데이터베이스에 삽입하는 함수 # 성공 시 True를 반환하고, 실패 시 False를 반환
db_connector = DBconnector(**DB_SETTINGS["POSTGRES"]) # PostgreSQL 설정을 사용하여 DB 연결 객체 생성
# with 문을 사용해 DB 연결을 관리 (연결이 자동으로 닫힘)
with db_connector as connected:
try:
orm_conn = connected.orm_connect() # ORM 방식으로 데이터베이스에 연결
# 데이터프레임을 'fake' 테이블에 삽입, 기존 데이터가 있으면 추가 (if_exists="append")
df.to_sql(name="fake", con=orm_conn, if_exists="append", index=False)
return True # 성공적으로 삽입된 경우 True 반환
except Exception as e:
print(e) # 에러 발생 시 에러 메시지를 출력
return False # 실패한 경우 False 반환
④ fakedata>process.py
# process.py
import pandas as pd
def categorize_age(age: int): # 나이를 나이대 문자열로 변환하는 함수 # 예: 25 -> "20대", 95 -> "90대 이상"
if age >= 90:
return "90대 이상"
else:
return str(age // 10 * 10) + "대" # 나이대를 10 단위로 구분하여 문자열로 반환
def create_fakedatamart(df: pd.DataFrame) -> pd.DataFrame: # 가상 데이터 프레임을 가공하여 새로운 데이터 마트 데이터프레임을 생성하는 함수
today = pd.to_datetime("today") # 현재 날짜를 가져옴
df["birthdate"] = pd.to_datetime(df["birthdate"], format="%Y%m%d") # birthdate를 날짜 형식으로 변환
# 도시 컬럼 생성 - 거주지(residence)에서 첫 번째 단어를 추출하여 city로 지정
df["city"] = df["residence"].str.split().str[0]
# 출생 연도 컬럼 생성 - birthdate에서 연도(year)만 추출하여 birth_year로 지정
df["birth_year"] = df["birthdate"].dt.year
# 나이 컬럼 생성 - 현재 연도와 출생 연도를 비교하여 나이를 계산
df["age"] = (today.year - df["birthdate"].dt.year - (
(today.month < df["birthdate"].dt.month) |
((today.month == df["birthdate"].dt.month) & (today.day < df["birthdate"].dt.day))
)).astype(int)
# 혈액형 컬럼 생성 - 혈액형(blood_group)에서 마지막 문자를 제외한 부분을 추출하여 blood로 지정
df["blood"] = df["blood_group"].str.slice(0, -1)
# 나이대 컬럼 생성 - categorize_age 함수를 사용하여 나이대를 age_category로 지정
df["age_category"] = df["age"].apply(categorize_age)
# 필요한 컬럼만 선택하여 최종 데이터 마트 데이터프레임을 생성
column_list = ["uuid", "name", "job", "sex", "blood", "city", "birthdate", "age", "age_category"]
df_datamart = df[column_list]
return df_datamart # 가공된 데이터프레임 반환
3) 이미 존재하는 컬럼(level_0) 삭제
from db.connector import DBconnector # DBconnector 클래스를 통해 데이터베이스에 연결
from settings import DB_SETTINGS # DB 설정 정보를 가져옴
# 삭제할 테이블 및 컬럼 정보
table_name = "lecture" # 컬럼을 삭제할 테이블 이름
column_name = "level_0" # 삭제할 컬럼 이름
# 컬럼 삭제 작업
try:
# DBconnector 인스턴스를 통해 데이터베이스에 연결
db_connector = DBconnector(**DB_SETTINGS["POSTGRES"])
# `postgres_connect()` 메서드를 통해 연결 설정
with db_connector.postgres_connect() as connector:
cursor = connector.conn.cursor() # 데이터베이스 커서를 생성
# 컬럼 삭제 쿼리 실행
alter_query = f"ALTER TABLE {table_name} DROP COLUMN IF EXISTS {column_name}" # 컬럼 삭제 SQL 쿼리
cursor.execute(alter_query) # 쿼리 실행
connector.conn.commit() # 변경 사항을 커밋하여 반영
print(f"'{table_name}' 테이블에서 '{column_name}' 컬럼이 삭제되었습니다.") # 성공 메시지 출력
except Exception as e:
print(f"오류 발생: {e}") # 예외 발생 시 에러 메시지 출력
finally:
# 커넥션 및 커서 닫기
if 'cursor' in locals() and cursor is not None: # 커서가 존재하는 경우 닫기
cursor.close()
if 'connector' in locals() and connector.conn is not None: # 연결 객체가 존재하는 경우 닫기
connector.conn.close()
4) PostgreSQL DB에 Fake 데이터 저장
① create.py에서 함수(create_fakeuser) 호출 → Fake user 데이터 생성
from create import create_fakeuser
create_fakeuser() # 정상적으로 데이터가 생성된 것을 확인
"""10명의 Fake user 데이터 생성 및 출력
for _ in range(10):
fake_user = create_fakeuser() # create.py의 create_fakeuser 함수 호출
print(fake_user) # 생성된 가짜 사용자 데이터 출력
"""
② create.py에서 함수(create_fakedataframe) 호출: Fake 데이터 → Fake 데이터프레임 저장
- create_fakedataframe() ⊃ create_fakeuser()
import pandas as pd
from random import randint
from create import create_fakedataframe # create.py에서 create_fakedataframe 함수 가져오기
count = randint(5, 15) # 5에서 15 사이의 랜덤한 개수 생성
df = create_fakedataframe(count) # create_fakedataframe 함수 호출로 DataFrame 생성
③ PostgreSQL 데이터베이스에 연결
from db.connector import DBconnector
from settings import DB_SETTINGS
pgsql_obj = DBconnector(**DB_SETTINGS["POSTGRES"]) # PostgreSQL 데이터베이스 연결 객체 생성
pgsql_obj.__dict__ # 데이터베이스 정보를 딕셔너리로 확인
④ 연결된 데이터베이스에 데이터 넣기(테이블 명: fake)
with pgsql_obj as connected:
sqlalchemy_conn = connected.orm_connect() # SQLAlchemy 연결 객체 생성
df = create_fakedataframe(count) # 랜덤한 개수의 가짜 데이터프레임 생성
df.to_sql(name="fake", con=sqlalchemy_conn, if_exists="replace", index=False)
# if_exists="replace" -> 기존 테이블을 대체하여 새 데이터로 저장
# if_exists="append" -> 기존 데이터에 새 데이터를 추가하여 누적 저장
print(df) # 생성된 Fake 데이터프레임 출력
⑤ 데이터가 잘 들어갔는지 유효성 검사
# 데이터가 비어 있지 않은지 확인
assert not df.empty, "DataFrame이 비어있습니다."
# 필수 컬럼이 모두 포함되어 있는지 확인
assert set(["name", "ssn", "job", "residence", "blood_group", "sex", "birthdate"]).issubset(df.columns), "필드가 누락되었습니다."
# birthdate 필드가 "YYYYMMDD" 형식인지 확인
assert df["birthdate"].str.match(r"\d{8}").all(), "birthdate 필드가 YYYYMMDD 형식이 아닙니다."
# uuid 필드가 문자열 형식인지 확인
assert df["uuid"].apply(lambda x: isinstance(x, str)).all(), "uuid 필드가 문자열이 아닙니다."
# 모든 유효성 검사가 통과된 경우 출력
print("모든 필드가 올바르게 생성되었습니다.")
5) 데이터 가공
[pandas를 통해 Data Mart 형태의 테이블로 가공]
1. 거주하는 도시 통계 -> 'residence' 전처리 및 칼럼 생성
2. 혈액형 통계 -> 'blood_group' 전처리 및 컬럼 생성
3. 남녀 통계 -> 'sex' 칼럼 사용
4. 나이테 통계 -> 'birthdate' 전처리 및 칼럼 생성
① Fake 데이터 프레임 생성(create_fakedataframe) → 데이터 마트 생성(create_fakedatamart)
from fakedata.process import create_fakedatamart # 데이터 가공 함수 가져오기
from fakedata.create import create_fakedataframe # 가짜 데이터 생성 함수 가져오기
# 10개의 가짜 데이터를 포함한 데이터프레임 생성
_df = create_fakedataframe(10)
# 생성된 데이터를 데이터마트 형식으로 가공
df = create_fakedatamart(_df)
# df 데이터프레임에서 sex 컬럼을 기준으로 그룹화하여 각 성별의 데이터 개수를 세고, 이를 새로운 데이터프레임으로 반환
# df.groupby("sex").size().reset_index(name="count")
6) 데이터 이행 모듈에 통합
① main.py
[tip] main.py 의 Batch Process
1. Click 라이브러리를 사용하여 명령줄에서 실행 가능한 스크립트를 정의
2. 가짜 데이터를 생성하여 PostgreSQL 데이터베이스에 저장
3. ETL 컨트롤러를 실행
# main.py
import click # Click 라이브러리 - 명령줄 인터페이스 구성
import random, time
from datetime import datetime, timedelta # 날짜와 시간 처리
from controller import controller # ETL 컨트롤러 함수 가져오기
from fakedata.create import create_fakedataframe # 가짜 데이터 생성 함수
from fakedata.insert import insert_fakedataframe # 가짜 데이터 삽입 함수
@click.command()
@click.option("-s", "--batch_date", default="") # batch_date 옵션, 기본값은 빈 문자열
def start(batch_date: str) -> None:
# 배치 프로세스를 시작하는 함수
if not batch_date: # batch_date 값이 주어지지 않은 경우 어제 날짜를 사용
_date = datetime.now() - timedelta(days=1)
batch_date = _date.strftime("%Y%m%d") # YYYYMMDD 형식으로 변환
for _ in range(3): # 배치 작업을 3번 반복
count = random.randint(5, 15) # 5~15 사이의 랜덤한 데이터 개수 생성
print(f"======{count}개의 row가 삽입됩니다.======") # 디버깅 출력
# 가짜 데이터 생성
fake_df = create_fakedataframe(count)
print("<<< fake_df 생성 >>>") # 가짜 데이터프레임 생성 디버깅 출력
# 생성한 가짜 데이터를 PostgreSQL에 삽입
insert_fakedataframe(fake_df)
# ETL 컨트롤러 실행
controller(batch_date)
time.sleep(3) # 각 배치 사이에 3초 대기
if __name__ == "__main__":
start() # start 함수 실행
"""
[파이썬 파일 실행 명령어]
터미널 창에, python main.py
"""
7) 시각화
① app.py
# app.py
import streamlit as st # Streamlit 라이브러리 - 웹 대시보드 개발을 위한 라이브러리
import pandas as pd # Pandas 라이브러리 - 데이터프레임을 사용하여 데이터 조작
import altair as alt # Altair 라이브러리 - 데이터 시각화를 위한 라이브러리
# Streamlit 웹 설정
st.title("사용자 데이터 시각화 대시보드") # 대시보드의 제목을 설정
st.sidebar.title("필터") # 사이드바 제목 설정
# 예시 데이터 생성
data = {
"age_category": ["10대", "20대", "30대", "10대", "10대", "30대", "30대", "40대", "50대", "10대"],
"sex": ["M", "F", "M", "F", "F", "F", "M", "M", "F", "F"],
"city": ["서울", "부산", "대구", "대구", "대구", "대구", "부산", "서울", "서울", "대구"]
}
df = pd.DataFrame(data) # 딕셔너리 데이터를 Pandas DataFrame으로 변환
# 나이대 필터
# 사용자가 사이드바에서 선택한 나이대를 기반으로 데이터를 필터링
age_category = st.sidebar.multiselect(
"나이대 선택", # 멀티셀렉트 위젯의 제목
df["age_category"].unique(), # 선택할 수 있는 나이대 목록
default=df["age_category"].unique() # 기본 선택값으로 모든 나이대를 선택
)
filtered_data = df[df["age_category"].isin(age_category)] # 선택된 나이대로 데이터 필터링
# 1. 나이대별 인원수 시각화
age_chart = (
alt.Chart(filtered_data) # 필터링된 데이터를 사용하여 Altair 차트 생성
.mark_bar() # 막대(bar) 차트 유형 설정
.encode(
x = alt.X("age_category", sort="-y", title="나이대"), # x축에 나이대, 내림차순 정렬
y = alt.Y("count()", title="인원수"), # y축에 인원수(카운트)
color = "age_category" # 나이대별 색상 구분
)
.properties(title="나이대별 인원 분포") # 차트 제목 설정
)
# 2. 성별별 인원수 시각화
gender_chart = (
alt.Chart(filtered_data) # 필터링된 데이터를 사용하여 Altair 차트 생성
.mark_bar() # 막대(bar) 차트 유형 설정
.encode(
x = alt.X("sex", title="성별"), # x축에 성별
y = alt.Y("count()", title="인원수"), # y축에 인원수(카운트)
color = "sex" # 성별별 색상 구분
)
.properties(title="성별별 인원 분포") # 차트 제목 설정
)
# 3. 지역별 인원수 시각화
city_chart = (
alt.Chart(filtered_data) # 필터링된 데이터를 사용하여 Altair 차트 생성
.mark_bar() # 막대(bar) 차트 유형 설정
.encode(
x = alt.X("city", sort="-y", title="지역"), # x축에 지역, 내림차순 정렬
y = alt.Y("count()", title="인원수"), # y축에 인원수(카운트)
color = "city" # 지역별 색상 구분
)
.properties(title="지역별 인원 분포") # 차트 제목 설정
)
# 시각화 결과 출력
st.altair_chart(age_chart, use_container_width=True) # 나이대별 인원 분포 차트를 Streamlit에 표시
st.altair_chart(gender_chart, use_container_width=True) # 성별별 인원 분포 차트를 Streamlit에 표시
st.altair_chart(city_chart, use_container_width=True) # 지역별 인원 분포 차트를 Streamlit에 표시
"""
[시각화 명령어]
터미널 창에, streamlit run app.py -> 이메일 입력
"""
② 시각화 결과
[파이썬 심화]
1. 기본 함수와 데코레이터
1) 기본 함수 정의 및 실행
- time.sleep(): 입력된 시간(초) 동안 코드 실행을 중지
- time.time(): 현재 시간을 초 단위로 측정
import time # 시간을 측정하거나 지연을 줄 수 있는 모듈
def say_hello():
# 이 함수는 "Hi!"를 출력하고, 3초간 대기한 후 "Goodbye"를 출력함
print("Hi!")
time.sleep(3) # 3초간 대기. time.sleep() 함수는 입력된 시간(초)만큼 일시적으로 코드 실행을 중지시킴
print("Goodbye")
# 실행 시작 시간을 기록하여 총 소요 시간 계산에 사용
start = time.time() # 현재 시간을 초 단위로 측정해 변수 start에 저장
say_hello() # say_hello 함수 호출
2) 데코레이터
① 정의 및 사용
- 데코레이터: 다른 함수의 동작을 감싸서 추가적인 기능을 부여할 수 있는 구조
- "he = say_hello": say_hello 함수를 he라는 변수에 참조시킨 것
- he()를 호출하면 say_hello()가 실행됨
- he 자체는 함수 객체를 가리키는 변수
def time_checker(func):
# time_checker 데코레이터는 전달된 함수(func)의 실행 시간을 측정하여 출력하는 함수
def wrapper():
start = time.time() # 시작 시간 기록
func() # 전달된 함수 실행
end = time.time() # 종료 시간 기록
print("total time: ", end - start) # 실행에 걸린 총 시간 출력
return wrapper # wrapper 함수를 반환하여 데코레이터 역할 수행
# 함수 객체 호출
he = say_hello # say_hello 함수 자체를 he 변수에 저장
he() # he()로 say_hello() 함수 호출 (실제 함수 실행)
he # 함수 객체 자체를 출력, 실행되지 않음
② 데코레이터 적용 방식
- @데코레이터 표기법: @time_checker 데코레이터가 say_hello와 say_bye에 적용됨
- 이 경우, 각 함수가 호출될 때 time_checker의 wrapper가 실행되어 해당 함수의 실행 시간을 측정하고 출력
- 함수 데코레이터: 코드의 가독성을 높여주며, 동일한 기능(여기서는 실행 시간 측정)을 여러 함수에 쉽게 적용할 수 있도록 함
@time_checker # time_checker 데코레이터를 say_hello 함수에 적용
def say_hello():
time.sleep(3) # 3초간 대기
print("Hello!")
@time_checker # time_checker 데코레이터를 say_bye 함수에 적용
def say_bye():
time.sleep(3) # 3초간 대기
print("Bye!")
say_hello() # 데코레이터가 적용된 say_hello 함수 실행
say_bye() # 데코레이터가 적용된 say_bye 함수 실행
2. 클래스와 메서드
1) 기본 클래스 정의 및 __init__ 메서드
- __init__ 메서드: 객체가 생성될 때 초기화를 담당. var1과 var2 값을 받아 인스턴스 변수를 설정
- 메서드 호출: upper_letter와 lower_letter 메서드는 var1의 대문자와 소문자 변환 버전을 각각 생성해 인스턴스 속성으로 저장하며, 출력함
- __dict__: 객체의 속성을 딕셔너리 형태로 보여주어 속성 정보를 확인할 수 있음
class TempClass:
# 클래스 초기화 메서드 (__init__) 정의. 인스턴스가 생성될 때 자동으로 호출됨
def __init__(self, var1: str, var2: int):
self.var1 = var1 # 인스턴스 변수 var1에 입력값을 저장
self.var2 = var2 # 인스턴스 변수 var2에 입력값을 저장
def upper_letter(self):
# var1을 대문자로 변환하여 upper_var1 속성에 저장하고, 변환된 값 출력
self.upper_var1 = self.var1.upper()
print(self.var1.upper())
def lower_letter(self):
# var1을 소문자로 변환하여 lower_var1 속성에 저장하고, 변환된 값 출력
self.lower_var1 = self.var1.lower()
print(self.var1.lower())
# 객체 생성 및 메서드 호출
tc = TempClass("BANANA", 50) # TempClass 인스턴스 생성
tc.lower_letter() # lower_letter 메서드 호출, "banana" 출력
print(tc.__dict__) # 객체의 모든 속성을 딕셔너리로 확인
2) dataclass 활용
- @dataclass: 클래스의 기본적인 메서드(__init__, __repr__, __eq__ 등)를 자동으로 생성해주는 데코레이터
- frozen=True로 설정하면 불변 객체가 되어 인스턴스를 생성한 후 필드를 수정할 수 없음
- __post_init__ 메서드: 데이터 클래스의 초기화 이후 추가적인 설정이 필요할 때 사용
- object.__setattr__을 통해 불변 객체의 속성을 설정할 수 있습니다.
from dataclasses import dataclass, field
# 데이터 클래스를 frozen=True로 설정하여 불변 객체로 만듦
@dataclass(frozen=True)
class TempClass:
var1: str # 필드 var1 정의
var2: int # 필드 var2 정의
upper_var1: str = field(init=False) # 초기화에서 제외할 필드 upper_var1 정의
def __post_init__(self):
# 불변 객체의 필드를 설정하기 위해 object.__setattr__ 사용
object.__setattr__(self, "upper_var1", self.var1.upper())
3) staticmethod 활용
- @staticmethod: 인스턴스가 아닌 클래스 자체에서 호출할 수 있는 메서드를 정의
- 여기서 add_int는 클래스의 인스턴스 없이도 호출할 수 있음
- staticmethod의 용도: 클래스와 독립적인 기능을 정의할 때 사용
- 예를 들어, 인스턴스가 필요 없는 단순한 계산 작업 등을 처리할 때 유용
class TempClass:
# 클래스 초기화 메서드
def __init__(self, var1: str, var2: str):
self.a = var1 # 인스턴스 변수 a 설정
self.b = var2 # 인스턴스 변수 b 설정
self.upper_letter() # upper_letter 메서드 호출하여 대문자 설정
def upper_letter(self):
# var1을 대문자로 변환하여 upper_var1 속성에 저장
self.upper_var1 = self.a.upper()
@staticmethod
def add_int(age: int):
# 전달된 age 값을 두 배로 출력
print(age * 2)
# 클래스 메서드 호출
TempClass.add_int(5) # 정적 메서드 add_int 호출, 10 출력
3. 복잡한 데이터 구조와 dataclass
1) DataPipeline 클래스와 __post_init__ 메서드 활용
- DataPipeline 클래스: 복잡한 데이터를 다룰 때 사용
- catalog_name 딕셔너리와 schema_name 문자열을 매개변수로 받아 초기화 됨
- __post_init__ 메서드: dataclass의 __init__ 실행 후 추가 속성을 설정할 때 사용
- 여기서는 catalog_name의 "index" 값을 사용해 index 속성을 리스트로 초기화함
from dataclasses import dataclass
# DataPipeline 클래스 정의. catalog_name과 schema_name 필드를 받음
@dataclass
class DataPipeline:
catalog_name: dict # 딕셔너리 필드
schema_name: str # 문자열 필드
def __post_init__(self):
# catalog_name의 "index" 값을 리스트 형태로 새로운 index 속성에 추가
self.index: list[int] = [self.catalog_name["index"]]
def sample(self):
# sample 메서드 실행시 "sample" 출력
print("sample")
2) getattr 활용
- getattr 함수: 객체에서 속성이나 메서드를 동적으로 접근할 수 있게 해주는 함수
- 여기서는 DataPipeline 객체의 sample 메서드를 getattr을 통해 동적으로 호출함
- 딕셔너리 언패킹: pp = DataPipeline(**night)는 night 딕셔너리의 키-값 쌍을 DataPipeline 클래스의 매개변수로 전달해 객체를 생성하는 방식
- ** 표기법: 키-값 쌍을 해체하여 전달하는 역할
# DataPipeline 클래스 객체 생성 및 sample 메서드 호출
pipeline = DataPipeline(
catalog_name=night["catalog_name"], schema_name=night["schema_name"]
)
print("index 속성: ", pipeline.index) # index 속성 확인
pipeline.sample() # sample 메서드 호출
pp = DataPipeline(**night) # 딕셔너리 언패킹으로 객체 생성
# getattr로 DataPipeline 객체의 메서드 동적 호출
getattr(DataPipeline(**night), "sample")() # sample 메서드 동적으로 호출
📙 내일 일정
- PySpark 실습 및 문제 풀이
'TIL _Today I Learned > 2024.10' 카테고리의 다른 글
| [DAY 72] Data Pipeline 및 PySpark 시험 (0) | 2024.10.31 |
|---|---|
| [DAY 71] PySpark (0) | 2024.10.30 |
| [DAY 69] SQLAlchemy ORM (1) | 2024.10.28 |
| [DAY 68] 데이터 엔지니어링 (1) | 2024.10.25 |
| [DAY 67] AWS 아키텍처 그리기 (0) | 2024.10.24 |
