강의를 듣다 보니 다음과 같은 구절이 나왔다. 여기서 triggerer가 scheduler한테 알려서 scheduler를 깨우면 scheduler는 task를 어떻게 할당할까? 이미 한 번 executor를 통해 전략이 내려진 task니까 그대로 수행하는 게 더 효율적이지 않을까라는 생각이 들었다.

결론부터 말하면, scheduler는 다시 executor를 통해 task를 queue에 넣거나 worker에 보낸다.

다시 할당하는 이유

• Airflow의 단위는 Task 코드보다 TaskInstance(특정 run의 상태) 라서, Deferred -> 재실행 시점은 사실상 새 실행으로 취급된다.
• 그 사이에 worker 수, queue state, executor의 parallelism, 우선순위, 다른 task의 대기 상태가 전부 바뀌었을 수 있기 때문에, task가 오면 가장 효율적인 worker를 다시 선택하는 것이 유연한 대처 방법이다.
• CeleryExecutor, KubernetesExecutor와 같은 분산/동적 스케일링이 있는 환경에서는 예전에 돌린 work를 고정시키는 게 비효율적이거나 불가한 경우가 많다.

Deferrable Task 기다리는 동안 잠깐 잠들었다가, 조건이 되면 다시 깨어나는 Task

• defer()라는 함수를 호출해서 알린다.
• 기다리는 동안 worker 대신 가벼운 triggerer 프로세스가 외부 조건을 계속 감시한다.

polling : 새로운 일이 생겼는지 계속 물어보는 방식

• 파일이 생겼는지
• API 작업 끝났는지
• DB가 특정 상태가 되었는지 polling은 API 호출이 아니다. polling은 주기적으로 반복해서 호출하는 패턴이고, 그 안에서 API를 사용할 뿐이다.

webhook : 서버가 알아서 먼저 연락해주는 API

• 어떤 시스템에서 특정 이벤트가 발생했을 때, 미리 정해둔 URL로 자동으로 HTTP 요청을 보내는 방식이다.
• event 기반 push 알림
• 알림을 받는 쪽 URL : [Webhook URL / Callback URL / Endpoint] 라고 부른다.

polling은 client가 변화가 있는지 계속 물어보는 방식이라 요청이 자주 발생하고 자원을 많이 쓴다. webhook은 서버가 이벤트가 생긴 순간, 1번 알려주는 방식이라 불필요한 요청이 줄어든다.

arr = list(map(int, input().split()))

라는 코드가 있었다. 정말 자연스럽게 주어진 코드였는데 문득,

arr = [map(int, input().split())]

로 코드를 작성하는 게 더 간이할 것 같은데? 라는 생각이 들었다.

a, b = map(int, input().split())

정말 많이 쓰는 코드니까 당연히 되겠지 싶었는데 []는 안된다.

map()은 주어진 코드에 따라int 형변환을 하려고 대기 중인 map 객체이다. 여기서 포인트는 대기 중인 map 객체라는 것이다.

a, b = map(int, input().split())

을 보면 대기 중인 map 객체를 a, b로 unpacking을 하면서 값을 할당한 걸 볼 수 있는데 내가 작성한 리스트 리터럴 형식의 문법은 안에서 대기 중인 값들을 unpacking할 방법이 없다. 따라서 arr에 리스트로 씌워진 map 객체 하나만 달랑 있는 것이다.

그럼 여기서 생각해볼만한 부분이 그럼 *로 unpacking 하면 되겠네?

arr = [map(int, input().split())]
print('[map(int,input().split())]', type(arr))
print(type(arr[0]))

arr_with_unpacking =[*map(int, input().split())]
print('*map(int, input().split())', arr_with_unpacking)
print(type(arr_with_unpacking[0]))

잘 되는 걸 볼 수 있다. 문득 생각이 나서 해봤는데, 좋은 접근법이었다.

Rating Class

PySpark(MLlib)에서 추천 시스템 모델(ALS) 만들 때 아주 표준적으로 사용하는 데이터 틀.

• User
• Product
• Rating

from pyspark.mllib.recommendation import Rating

# Rating(user, product, rating)
my_rating = Rating(1, 101, 5.0)

Rating 객체는 내부적으로 namedtuple과 비슷하게 동작한다. my_rating.user처럼 속성값에 쉽게 접근할 수 있다.

namedtuple이란? Python의 collections 모듈에 있는 가벼운 튜플 기반 자료형 값은 바꿀 필요 없지만, 각 칸에 이름을 붙여 읽기 쉽게 쓰고 싶을 때 사용한다. 불변한 값이고 리스트처럼 수정못 하지만, field name으로 접근할 수 있어 가독성이 좋다.

Rating class → Transformation

Transformation Process

1. Parsing : map → 문자열 줄을 Rating 객체로 변환
2. Filtering : filter → 특정 조건만 남도록 필터링
3. Extraction : map → 특정 정보만 추출

raw_data = sc.parallelize(["1, 101, 5,0", "1, 102, 3.0", "2, 101, 4.0"])

ratings_rdd = raw_data.map(lambda line: line.split(",")) \
                      .map(lambda x: Rating(int(x[0]), int(x[1]), float(x[2])))

high_ratings = ratings_rdd.filter(lambda r: r.rating >= 4.0)

RDD를 다룰 때는 Lazy Evaluation(지연 연산) 특징을 기억해야 한다. map, filter를 쓴다고 바로 계산이 일어나지 않고 나중에 collect, count와 같은 Action 명령할 때까지 계획만 세워둔다.

Lazy Evaluation으로 얻을 수 있는 이점

1. Query Optimization Spark는 명령어를 받으면 바로 실행하지 않고 DAG라는 설계도를 그린다. ex : 100만 개의 데이터 중에서 1번 유저만 필터링한 후에 5개만 가져와라. → Spark는 100만 개를 다 뒤지는 개 아니라 5개를 찾는 즉시 작업을 멈추는 최적의 경로를 찾아낸다.

   2. 메모리 효율성 데이터 즉시 변형하지 않기에, 불필요한 중간 데이터 결과를 메모리에 일일이 저장하지 않는다.

   3. 장애 복구 (Fault Tolerance) 계산 도중에 서버 한 대가 고장 나도, spark 그동안 그려온 설계도(Lineage)가 있기 때문에, 고장 난 부분만 다시 계산해서 복구할 수 있다.

ALS(Alternating Least Squares, 교차 최소 제곱법)

추천 시스템에서 가장 유명한 행렬 분해 알고리즘

데이터 : 사용자와 아이템으로 이루어진 거대한 행렬 → 사용자의 취향 행렬(U)과 아이템의 특성 행렬(I)로 행렬을 분해한다. 이 두 값을 곱하면 예측 평점이 나온다.

Alternating? 두 행렬을 동시에 맞추는 것는 난이도가 높다.

1. 사용자 취향을 고정하고, 거기에 맞는 아이템 특성을 계산한다.
2. 아이템 특성을 고정하고, 거기에 맞는 사용자 취향을 다시 계산한다.
3. 만족스러운 과정이 나올 때까지 이 두 과정을 번갈아가면서(Alternating) 반복한다.

• 확장성(Sacalability) : 데이터가 많아도 사용자/아이템별로 계산을 쪼갤 수 있어 Spark 같은 분산 처리 시스템에 적합하다.
• Cold start 완호 : 사용자의 평점이 몇 개 없어도 비슷한 취향의 데이터를 통해 예측이 가능하다.

ALS의 핵심 파라미터

1. Rank(계수)
   • 사용자나 아이템의 특징을 몇 개의 숫자로 표현할 것인가?
   • 10~200 사이에서 결정
2. Iterations(반복 횟수)
   • 사용자 행렬 ←→ 아이템 행렬을 몇 번 번갈아가며 업데이트할 것인가?
   • 10~20
3. Lambda(정규화 매개변수)
   • 모델이 너무 복잡해지지 않도록 벌금을 주는 수치
   • 0.01, 0.1, 1.0 같은 값들로 테스트한다.

이 수치들을 제각기 조절해가는 것보다는 Grid Search라는 방법을 사용한다. 각 파라미터에 리스트를 설정해두면 모든 조합을 컴퓨터가 테스트한다.

Vector & LabeledPoint

={cyan}Vector(데이터의 특징을 숫자로)= 분류 모델이 이해할 수 있도록 데이터를 숫자로 바꾼 것이 벡터이다. Spark에서는 두 가지 형태의 벡터를 지원한다.

• Dense Vector (밀집 벡터) : 모든 데이터를 다 적는 방식
  • [1.0, 0.0, 3.5] (모든 위치의 값을 다 기록)
• Sparse Vector (최소 벡터) : 0이 아주 많을 때, 0이 아닌 값이 어디에 있는지만 적어서 메모리를 아끼는 방식
  • 100개 데이터 중 5번 위치에 1.0, 10번 위치에 3.5가 있고 나머지는 다 0

={blue}LabeledPoint(정답 + data)= 벡터에 '이건 스팸이야(1), 이건 정상이야(0)'라는 정답(Label)을 붙여아 모델이 학습을 할 수 있다. 그 정답지 역할을 하는 게 LabeledPoint이다.

• 구조 : LabeledPoint(label, features)
  • label : 우리가 맞히고 싶은 정답 ( 보통 Double 타입)
  • features : Vector (데이터의 특징)

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.linalg import Vectors

lp = LabeledPoint(1.0, Vectors.dense([0.1, 0.5, 0.2]))

print(lp.label)     # 1.0
print(lp.features)  # [0.1, 0.5, 0.2]

MLlib의 많은 알고리즘은 입력값으로 반드시 LabeledPoint 형태의 RDD를 요구한다.

Map vs flatMap

map을 사용하는 상황

• 데이터의 형식만 바꾸고 싶을 때 사용한다.
• 전체 데이터의 개수(Line 수)가 변하지 않아야 할 때 적합하다.

flatMap을 사용하는 상황

• 데이터를 잘게 쪼개거나 펼치고 싶을 때 사용한다.
• 하나의 행을 여러 개의 데이터로 분리하고 싶을 때 적합하다.
• 문장을 단어 단위로 쪼갤 때, 리스트 안에 리스트가 들어있는 구조를 하나로 합치고 싶을 때

flatMap은 map + flatten의 합성어다. map으로 먼저 데이터를 쪼갠 뒤(List의 형태로), 그 리스트의 껍데기를 까서 내용물만 밖으로 꺼내는(flatten) 작업을 동시에 해주는 것이다.

Clustering

라벨 없이 유사도가 높은 그룹들을 조직하는 비지도 학습 과제

• K-means
• 가우시안 혼합
• 전력 반복 클러스터링(PIC)
• 이분 K-means
• steraming K-means

K-means clustering

• 수치형 특성들로 이루어진 데이터
• 목표 클러스터링 수 : 'K'

Project

time_of_day (4)

• 새 컬럼 생성
  • 5-12 am → morning
  • 12-6pm → afternoon
  • 6-12pm → evening
  • lower bound 포함, upper bound 미포함

product

• 'TV'를 포함하는 행 제거 (1)
• 모든 values → lowercase (2)

category

• values → lowercase (2)

purchase_state (3)

• 컬럼 추가
• ordered_from에서 State 뽑아서 만들기

Data Processing in Shell #bash

curl(Client for URLs)

서버와 데이터를 주고받기 위한 Unix command line 도구 주로 HTTP 사이트나 FTP 서버에서 데이터를 다운로드할 때 사용한다.

기본 구조 : curl [option flag] [URL]

• URL 입력은 필수

단일 파일 다운로드

• 원래 파일 이름으로 저장 : -O → curl -O [file URL]
• 다른 이름으로 저장 : -o → curl -o [file URL]

서버에 비슷한 이름의 파일 다중 다운로드 curl -O https://website.com/datafile*.txt

Globbing Parser 활용

• 연속 다운로드 → [ ] 사용
  • curl -O https://~~~/datafilename[001-100].txt
• 간격 두고 다운로드 : 10번째 파일마다 다운로드 하고 싶다면 콜론: 추가
  • curl -O https://~~~/datafilename[001-100:10].txt

선제적 트러블슈팅

• -L : 300번대 에러 코드(리다이렉트) 발생 시 자동으로 해당 URL 따라간다.
• -C : 다운로드 도중 타임아웃 발생시, 중단된 지점부터 이어서 받는다.

Wget(World Wide Web & Get)

curl과 마찬가지로 HTTP 및 FTP를 통해 파일을 다운로드할 수 있는 도구

curl보다 더 다목적이다.

• 단일 폴더, 폴더 전체, 웹페이지 자체를 다운로드 가능
• 여러 파일을 재귀적으로 다운로드 가능

option

• -b : background에서 실행
• -q : Wget의 실행 로그 출력을 끈다
• -c : 이전에 중단된 다운로드를 이어서 받는다.(Wget이 아닌 다른 프로그램으로 받던 파일도 가능)

preview the log file → cat wget-log

Wget으로 여러 파일 다운로드 다운로드하려는 모든 URL이 url_list.txt에 있을 때, -i → Wget에게 로컬 파일에서 URL 읽어오도록 지시 wget -i url_list.txt

• -i url_list.txt 사이에 어떤 옵션도 들어가서는 안된다. -i 앞에 옵션 위치시킬 것

대용량 파일을 위한 다운로드 제한 설정 파일 다운로드가 네트워드 대역폭 전체를 점유하여 다른 작업을 방해하지 않도록 제한을 걸어야 할 때가 있다. --limit-rate option

• 숫자 입력 시 기본적으로 초당 바이트로 계산된다.
• wget --limit-rate=200k -i url_list.txt
  • 초당 다운로드 속도가 200KB를 넘지 않도록 제한한다.

소용량 파일을 위한 다운로드 제한 설정 작은 파일들을 여러 개 받을 때는 대역폭 제한보다 서버에 과부하를 주지 않는 것이 중요하다. 이때는 파일 다운로드 사이에 강제적인 대기 시간을 두는 --wait 옵션을 사용

• 시간 단위는 '초'
• wget --wait=2.5 -i url_list.txt : 파일 다운로드 시마다 2.5초의 휴식시간

csvkit

Bash 명령어에 부족한 데이터 핸들링 기능을 보완하기 위해 python 라이브러리에 의존한다. csvkit은 이러한 간극을 메워주기 위한 파이썬을 기반으로 개발한, 데이터 변환, 처리, 정제 기능을 모아놓은 suite이다.

in2csv : 파일을 CSV로 변환

• in2csv SPotifyData.xlsx > Spotify.csv
• > (리다이렉트 연산자)를 사용하지 않고 xlsx 파일만 입려하면 데이터가 터미널에 출력만 될 뿐, 파일로 저장되지 않는다.

엑셀 파일에 여러 시트 있을 때, 특정 시트 변환하기

• in2csv -n Spotify.xlsx (또는 --names) : 모든 시트 목록 출력
• 특정 시트 변환 in2csv --sheet "work-one_popularity" Spotify.xlsx > Spotify_population.csv
• 시트 이름에 공백이나 특수문자 있다면 따옴표("")로 감싸줘야 한다.

in2csv는 실행 시 별도의 로그를 남기지 않는다. ls를 통해 파일 생성을 확인해라.

csvlook : 데이터 미리보기 cat, less를 쓰면 터미널에서 데이터 형식이 깨져 보이기 쉽다. csvlook을 사용하면 마크다운 호환 방식의 고정폭 테이블로 예쁘게 정렬하여 보여준다. csvlook Sptify_population.csv

csvstat : 기술 통계 확인 describe()와 유사한 기능을 한다. 평균, 중앙값, 고유값 개수 등 주요 통계 수치를 요약해서 보여준다. csvstat Spotify_population.csv

RDD(Resilient Distributed Dataset)

회복 탄력성이 있는 분산 데이터셋

요즘은 잘 사용하지 않고 DataFrame을 사용하지만, 그 근간이 되는 low-level API이다.

주요 특징

• Resilient(회복력) : 데이터 손실이 일어나도, RDD는 데이터 생성 과정을 기록한 Lineage를 가지고 있어 자동으로 데이터를 복구한다.
• Distributed(분산)
• Dataset(데이터셋)

RDD의 주요 동작 방식

1. Transformation 기존 RDD에서 새로운 RDD를 만드는 과정 Lazy Evaluation(지연 연산) → 변환 명령을 내려도 즉시 실행되지 않고 기록만 해둔다.
   • map(), filter(), flatMap(), distinct()
   • flatmap() : 입력 문자열을 단어로 나누는 함수
2. Action 실제로 계산을 수행하거나 결과 반환, 저장하는 과정 Action이 호출되는 순간, 쌓여있던 Transformation들이 최적화되어 한꺼번에 실행된다.
   • collect(), count(), take(), saveAsTextFile(), first()
   • collect() : 모든 요소를 배열로 반환

       cluster 곳곳에 흩어져 있는 데이터를 Driver 프로그램의 메모리로 수집하여 파이썬의 List 형태로 반환한다.

   • take(N) : 앞의 N개 요소를 배열로 반환

data = [1, 2, 3, 4, 5]
rdd = sc.parallelize(data)  # list -> RDD

# Transformation
trasformed_rdd = rdd.filter(lambda x: x % 2 == 0).map(lambda x: x * 10)

# Action
result = transformed_rdd.collect()

SparkSession - DataFrame API 진입점

• SparkContext : RDD 생성을 위한 기본 진입점
• SparkSession : Spark DataFrame과 상호작용하는 단일 진입점
• SparkSession으로 DataFrame 생성, 등록, SQL 쿼리 실행을 수행

→ KeyError 발생

Boolean Indexing을 사용할 때는 대괄호 [] 안에 들어가는 데이터의 형태가 중요하다.

1. 대괄호 [] 내부의 데이터 타입

   sales_df[...] 내부에는 True 또는 False로 이루어진 리스트가 들어가야 한다.

PySpark #pyspark

Apache Spark : 대규모 데이터를 빠르게 처리하도록 설계된 오픈 소스 분산 컴퓨팅 시스템

• pyspark → apache spark의 파이썬 인터페이스

python workflow에서 병렬 계산으로 대용량 데이터셋을 효율적으로 처리하며, 배치 처리, 실시간 스트리밍, 머신러닝, 데이터 분석, SQL query에 적합하다.

• 대규모 데이터 분석 : spark의 inmemory 연산을 활용한 분산 데이터 처리
• 대규모 데이터셋 머신러닝
• ETL 및 ELT pipeline : 다양한 소스의 대량 원시 데이터를 구조화된 형식으로 변환

pyspark dataframe은 다른 DF와 유사하지만 PySpark에 최적화되어 있다.

• spark.read.csv(file_name, [column_names])
• .printSchema() : DataFrame의 구조 확인
• .count() : DataFrame 행 개수 세기
• .groupBy(), agg() → SQL 유사 집계
• filter() : SQL의 where처럼 동작
• select() : SQL의 select

pandas는 단일 compute instance에서 동작하는 반면, PySpark는 여러 인스턴스에 데이터를 분산해 처리 속도와 확장성을 확보한다.

결측치 처리

• na.drop() : null 값이 있는 행 삭제
• .where(col('columnName').isNotNull()) : null 값이 아닌 행만 뽑기
• .na.fill({'column': value) : null → 특정 값으로 대체

column 작업

• .withColumn() : 계산/기존 컬럼 기반 새 컬럼 추가 df = df.withColumn('age_plus_5', df['age'] + 5)
• withColumnRenamed() : column명 변경 df = df.withCOlumnRenamed('age', 'years')
• drop() : 불필요한 컬럼 제거

UNION 연산 : 구조가 같은 두 DataFrame을 위아래로 쌓아 하나로 만드는 도구이다.

• df_union = df1.union(df2)

Array와 Map

• Array : 열 내 리스트 저장에 유용 ArrayType(StringType(), False)
• Map: key-value 쌍, 딕셔너리형 데이터에 적합 MapType(StringType(), StringType()

StructType & StructField

StructType : 제목과 데이터 타입의 묶음 → DataFrame 전체의 구조 : 여러 개의 StructField를 리스트 형태로 담고 있다.

StructField

• Column name
• Data Type : IntegerType, StringType …
• Null 허용 여부 : None을 허용할 지 결정(True or False)

Pyspark는 데이터의 type를 추론(Inference)할 수 있지만, 데이터의 양이 많으면 type을 알아내기 위해 데이터를 다 흝어봐야 하기 때문에 시간이 오래 걸린다. → schema를 미리 정해줌으로써 시간을 단축한다.

RDD(Resilient Distributed Dataset)

PySpark → 병렬화를 통한 대규모 데이터 처리를 수행하는 능력이 뛰어나다.

병렬화 : 데이터를 클러스터의 여러 노드로 나누어 데이터와 연산을 분산시킨다. Spark에서 정의된 연산은 자동으로 분산되어, 대규모 데이터셋을 효율적으로 처리할 수 있다. 작업은 워커 노드에 할당되어 병렬로 데이터를 처리하고, 마지막에 결과를 합친다.

RDD : 클러스터 전반에 분산된 데이터 컬렉션을 표현하는 Spark의 핵심 빌딩 블록

• 불변 객체이므로 한 번 생성되면 변경할 수 없다.
• 대신 map(), filter()같은 연산으로 새로운 RDD를 만들 수 있다.
• RDD 연산의 결과를 가져오는 collect() 같은 액션도 지원한다.

DataFrame → SQL 실행

df.createOrReplaceTempviwe("people")

테스트를 실행하기 위해 필요한 준비물 @pytest.fixture라는 데코레이터를 사용해서 함수를 정의하면 다른 테스트 함수들이 이름을 파라미터처럼 넘겨받아 사용할 수 있다.

import pytest

@pytest.fixture
def sample_data():
    return {"name": "chan", "age": "25"}

def test_check_name(sample_data):
    # sample_data return 값이 자동으로 넘겨진다.
    assert sample_data['name'] == 'chan'

Fixture의 범위 (Scope)

fixture를 매번 새로 만들지, 아니면 한 번 만들어서 계속 쓸지를 결정하는 것을 scope라고 한다. @pytest.fixture(scope="설정")

• function (default) : 테스트 함수 실행될 때마다 매번 fixture를 새로 만든다.
• class : 같은 클래스 안에 있는 테스트 메서드끼리 fixture를 공유한다.
• module : 해당 .py 파일(module) 안에 있는 모든 테스트가 fixture를 한 번만 만들어 공유
• session : 전체 테스트 세션 동안 한 번만 만든다. → DB 연결처럼 무겁고 공통적인 작업에 사용

Setup과 Teardown(yield)

테스트를 위해 만든 임시 파일들을 정리하는 키워드가 yield이다.

fixture에서 yield를 사용하면

1. yield 전까지 코드가 실행된다.(테스트 준비 : Setup)
2. 테스트 함수가 실행되는 동안 잠시 멈춰 있는다.
3. 테스트가 끝나면 yield 다음 코드부터 다시 실행된다. (뒷정리 : Teardown)

import pytest
import os

@pytest.fixture
def temp_file():
    # setup 테스트용 파일 만들기
    f = open("test.txt", "w")
    f.wrtie("hello pytest")
    f.close()

    yield "test.txt"   # test 함수에 파일 이름 전달하고 대기

    # Teardown : 테스트가 끝나면 파일 삭제
    os.remove("test.txt")
    print("\n tmp file 삭제 완료")

만약 테스트 도중에 에러가 발생해서 테스트가 실패하더라도 pytest는 yield 뒤의 뒷정리 코드를 실행해서 테스트 환경을 깨끗이 유지할 수 있다.

autouse

자동으로 사용되는 fixture, fixture를 사용하려면 테스트 함수 인자에 fixture 이름을 적어줘야 하지만 autouse=True 설정을 하면, 이름을 적지 않아도 해당 범위 내의 모든 테스트에서 알아서 실행된다.

import pytest

@pytest.fixture(autouse=True)
def setup_log():
    print("[log] test start")
    yield
    print("[log] test finish")

def test_ex_1():
    # 인자에 setup_log 적지 않아도 자동 실행
    assert 1 == 1

def test_ex_2():
    assert 'a' == 'a'

사용하기 좋은 때

• 성능 모니터링 : 모든 테스트의 시작과 종료 시간을 기록하고 싶을 때
• 로그 기록 : 테스트마다 어떤 작업이 일어나는지 항상 남기고 싶을 때
• 환경 초기화 : 특정 폴더 항상 비워두거나, 공통 환경 변수 설정할 때

benchmark fixture

pytest-benchmark 설치 시 제공되는 benchmark라는 fixture → 측정하고 싶은 함수를 인자로 받아서 여러 번 반복 실행한 뒤, 평균 시간을 계산해 준다.

import pytest

@pytest.fixture
def big_data():
    # test 위한 대용량 데이터 준비
    return list(range(10000, 0, -1))

def test_sort_performance(benchmark, big_data):
    # benchmark fixture가 big_data 정렬하는 시간 측정
    result = benchmark(sorted, big_data)

    assert result == list(range(1, 100001))

fixture를 활용해서 benchmark를 쓰는 이유

• 데이터 준비 시간 제외 : benchmark의 핵심은 그 로직을 실행하는 시간만 재는 것이다. fixture를 쓰면 setup 시간은 측정에 포함되지 않고, benchmark() 안에 들어간 코드의 시간만 잴 수 있다.
• 다양한 환경 테스트 : fixture의 params 옵션을 사용하면 다양한 조건에서 성능이 어떻게 변하는지(시간 복잡도)를 한 번에 테스트할 수 있다.

pytest #python #pytest

기능테스트, 유닛테스트, 통합테스트, 성능테스트(pytest-benchmark)

성능 테스트

종류

• 부하 테스트(Load Testing) : 평상시 부하를 얼마나 잘 견디는지
• 스트레스 테스트(Stress Testing) : 시스템의 한계치가 어디인지 확인, 시스템 터졌을 때 어떻게 복구되는지
• 내구성 테스트(Endurance / Soak Testing) : 오랜 시간 부하 주었을 때 메모리 누수나 자원 소모 생기는지

성능 테스트 핵심 지표

1. Latency (응답 시간 / 지연 시간) : 사용자가 요청을 보내고 응답을 받을 때까지 걸리는 시간
2. Throughput (처리량) : 단위 시간당 시스템이 처리하는 요청의 양
3. Error rate (에러율) : 전체 요청 중 실패한 요청의 비율

여기서, 평균 응답 시간에 매몰되어서는 안된다. 90%의 유저는 잘 되어도 10%의 유저가 오래 걸리면 안되기 때문에 P95, P99(상위 95%, 상위 99%의 응답 시간) 같은 지표를 더 중요하게 봐야 한다.

성능 테스트 도구와 프로세스

성능 테스트 도구

• Locust : 파이썬 기반
• JMeter : 자바 기반, GUI 환경
• nGrinder : 네이버에서 만듦

테스트 진행 순서(Process)

1. 목표 설정
2. 환경 구축
3. 시나리오 작성
4. 부하 실행
5. 결과 분석 및 튜닝

• 코드 재사용

• DRY : 반복하지 말 것

  Python Class Decorator #python

python에서 decorator는 호출 가능한 객체(Callable)를 받아서 다른 객체를 반환하는 함수

• 기존 기능을 수정하지 않고 새로운 기능을 추가

class decorator는 클래스 자체를 수정하거나 확장할 때 사용

• __init__ , __call__

  class CallCounter:
    def __init__(self, func):
        self.func = func
        self.count = 0  # 호출 횟수를 저장하는 상태
  
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} 함수가 {self.count}번 호출되었습니다.")
        return self.func(*args, **kwargs)
  

@CallCounter def say_hello(): print("안녕하세요!")

say_hello() say_hello()

→ `sayl_hello`가 실행될 때마다, `self.count` 1씩 올리며 상태 유지


**class method** #python
___
python에서 일반 메서드는 첫 번째 인자로 `self`를 받는다. → `self`는 이미 만들어진 구체적인 객체(인스턴스)
- 일반 메서드 사용하려면 이미 `객체 = class()` 를 통해 이미 객체가 존재해야 한다.

**IF, 직원 데이터를 파일에서 읽어와서, 그 데이터를 바탕으로 직원을 새로 만들고 싶다면?**
- 직원을 만드려고 하는데
- 아직 직원 객체가 존재하지 않는다.
- 일반 메서드에서는 '존재하는 직원'이 수행하는 동작
- 존재하지 않는 객체에 메서드를 수행시킬 수는 없다!
- → `@classmethod` 등장

class method는 객체가 없어도 클래스(설계도)만 있으면 호출할 수 있다. 
> 자동차 회사에 전화 걸어서 직원 파일을 보낼 테니 자동차(객체)를 뽑아주세요!
> 자동차 객체는 없지만 자동차 회사 클래스는 존재한다!

```python
class Employee:
    def __init__(self, name, salary):
        self.name = name
        self.salary = salary

    # 일반 메서드는 불가능
    def from_file_instance(self, filename):
        # 이 메서드를 부르려면 이미 Employee 객체가 있어야 하는데,
        # 우리는 객체를 '만들기 위해' 이 기능을 쓰려는 거라 모순이 생겨요.
        pass

    # 클래스 메서드는 가능!
    @classmethod
    def from_file(cls, filename):
        # 파일 읽기 로직 (생략)
        name, salary = "철수", 50000 
        # cls는 Employee 클래스 그 자체이므로, 여기서 객체를 생성해서 반환합니다.
        return cls(name, salary) 

# 객체가 하나도 없는 상태에서도 호출 가능!
new_emp = Employee.from_file("info.txt")

일반 메서드는 만들어진 객체가 하는 행동이고, 클래스 메서드는 객체를 만들기 전에도 클래스가 할 수 있는 행동 → 대체 생성자(파일로 만들기, JSON으로 만들기 …)는 클래스 메서드로 만드는 것이 논리적으로 좋다.

class Person:
    CURRENT_YEAR = 2024
    def __init__(self, name, age):
        self.name = name
        self.age = age

    # Add a class method decorator
    @classmethod
    # Define the from_birth_year method
    def from_birth_year(cls, name, birth_year):
    # Create age
    age = Person.CURRENT_YEAR - birth_year
    # Return the name and age
    return cls(name, age)

bob = Person.from_birth_year("Bob", 1990)

self는 일반 메서드에서 나중에 생성될 실제 객체를 담기 위한 빈 그릇과 같다. 일반 메서드 호출 시 파이썬 내부 실행 동작은 다음과 같다.

1. emp = Employee() 객체 탄생
2. emp.general_method() 호출
3. 파이썬이 자동으로 Employee.general_method(emp)로 변환해서 실행 → 객체 emp를 self 자리에 넣어준다.

그래서, 일반 메서드는 실행되는 순간에 반드시 self 자리에 들어갈 실제 객체가 메모리에 살아있어야만 작동한다.

Employee.general_method()처럼 호출하면 self 자리에 객체를 넣어줘야 하는데, 전달된 객체가 아무것도 없기 때문에 self에 뭘 넣어서 실행할 지 모르기에 에러가 발생한다.

→그렇기 때문에, ={cyan}파일을 읽는 로직의 함수가 클래스 내부에 써 있더라도, 그 메서드 형식이 self 인자를 받는 일반 메서드라면 객체가 생겨야만 쓸 수 있는 기능=이다

class method가 가능한 이유는 self 대신 인자로 cls를 받기 때문이다.

• 일반 메서드 : self가 있어야만 기능 수행
• 클래스 메서드 : self가 아니라 cls(class 설계도)만 있으면 기능 수행 가능

그래서 파일로부터 객체를 만드는 from_file 같은 기능은 객체가 생성되기 전이기 때문에, self 인자가 아니라 class method나 정적 메서드(staticmethod)로 만들어야만 호출이 가능하다!!

static method(정적 메소드) #python

class 안에 있지만, self나 cls에 대한 정보와 관련 없이 사용한다. 인자를 정의할 때, 첫 인자로 self나 cls를 쓰지 않는다.

• 클래스 이름으로 바로 호출 가능
• 논리적으로 클래스와 관련 있는 편의 기능 묶어둘 때 사용

class Calculator:
    @staticmethod
    def add(a, b):
        return a + b

# 객체 만들지 않아도 바로 사용 가능
print(Calculator.add(10, 20))

singleton desing pattern #python

어떤 클래스의 인스턴스가 프로그램 전체에서 오직 하나만 존재하도록 보장하는 디자인 패턴

why? 여러 곳에서 동시에 접근하면 안되거나, 자원을 공유해야 하는 경우에 필요하다.

• DB 연결 객체 : 연결을 여러 개 만들면, 서버 부하가 커지기 때문에 하나만 만든다.
• Settings 관리자 : 프로그램 설정값은 하나로 통일되어야 한다.
• Logging 객체 : 로그를 기록하는 통로를 하나로 통일할 때 쓴다.

def singleton(cls):
    # 생성될 객체를 담아둘 저장소
    instances = {} 

    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)

        return instances[cls]

    return get_instance

@singleton
class Database:
    def __init__(self):
        print('connected database')

db1 = Database()
db2 = Database()

print(db1 is db2) # resutl = True

데코레이터로 구현하는 방법 이외에도, 클래스 자체의 생성 과정을 제어하는 __new__ 매직 메서드를 사용할 수 있다.

class SingletonClass:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            # 인스턴스가 없을 때만 부모 클래스의 __new__를 호출해 생성
            cls._instance = super().__new__(cls)

팩토리 패턴 #python

어떤 객체를 만들지 결정하는 로직을 별도의 클래스나 함수에 둔다.

• 유연성 : 나중에 새로운 종류의 객체가 추가되어도, 메인 코드를 수정하지 않고 팩토리 객체에 추가하면 된다.
• 복잡함 은폐 : 객체를 만드는 과정이 복잡할 때(여러 설정 거칠 때), 사용자가 이를 모른 상태로도 할 수 있도록 한다. (식당에 가서 레시피를 모르고 메뉴를 시킬 수 있는 것처럼)

class Dog:
    def speak(self):  return "wow"
class Cat:
    def speak(self): return 'yaong'

class PetFactory:
    @staticmethod
    def get_pet(pet_type):
    pets = {"dog": Dog, "cat": Cat}
    return pets.get(pet_type, Dog)()

my_pet = PetFactory.get_pet("cat")
print(my_pet.speak())

factory method patter 확장 if-else, dictionary로 객체를 찍어내는 것을 넘어서 공장 자체를 추상화하는 방식이 있다. 이를 factory method pattern이라고 한다.

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def speak(self):   pass

class JindoDog(Ainmal):
    def speak(self):   return '진돗개 멍'

class Spasal(Animal):
    def speak(self):   return '삽살개 멍'

class AnimalFactory(ABC):
    @abstractmethod
    def create_method(self):   pass

    def deliver(self):
        animal = self.create_animal()
        print(f" {animal.speak() = }")

class SeoulFacotry(AnimalFactory):
    def create_animal(self):  return JindoDog()

class BusanFactory(AnimalFactory):
    def create_animal(self):   return Sapsal()


seoul = SeoulFactory()
seoul.deliver()

→ 객체 생성 로직 분리되어 있어, 새로운 종류의 객체가 추가되어도 기존 코드를 수정하기가 편하다.

Abstract class(추상 클래스) #python

상속받는 자식 클래스들이 반드시 구현해야 할 메서드를 지정해 주는 가이드라인

• 자기 자신으로는 객체를 만들 수 없지만, 다른 클래스들이 공통적으로 가져아 할 특정을 정의할 때 사용한다.

왜 미완성 클래스를 만들고, abc module이 필요한가?

• 협업할 때, 공통적으로 지켜야 할 규칙을 강제하기 위함이다.

from abc import ABC, abstractmethod

class Animal(ABC): # ABC를 상속받으면 추상 클래스가 된다.
    @abstractmethod
    def move(self):
        """animal이라면 반드시 move 기능이 있어야 한다."""
        pass

my_animal = Animal() # 추상 클래스는 직접 객체를 만들 수 없으므로 에러 발생

class Human(Animal):
    def move(self):
        print('두 발로 걸음")

class Fish(Animal):
    def move(self):
        print('지느러미로 헤엄침')

h = Human()
h.move()

만약 move() 메서드를 정의하지 않으면 Can't instantiate abstract class Fish with abstract method move 라는 에러를 띄운다.

파이썬에는 다른 언어처럼 별도의 interface 키워드는 없지만, 추상 클래스를 이용해 그 역할을 수행한다.

• 추상 클래스 : 자식 클래스는 부모 클래스의 기능을 쓰고, 필요한 건 추가로
• 인터페이스(형태) : 강제 메서드만 존재(기능은 없고 형태만 갖춤)

from abc import ABC, abstractmethod

# python style's interface
class Remocon(ABC):
    @abstractmethod
    def turn_on(self):   pass
    def turn_off(self):  pass

# functions - implement interface
class TV(Remocon):
    def turn_on(self): print('turn on tv')
    def turn_off(self): print('turn off tv')

class Airconditioner(Remocon):
    def turn_on(self): print('turn on air')
    def turn_off(self): print('turn off air')

sort puts data in order. By default it does this in ascending alphabetical order, but the flags -n and -r can be used to sort numerically and reverse the order of its output, while -b tells it to ignore leading blanks and -f tells it to fold case (i.e., be case-insensitive). Pipelines often use grep to get rid of unwanted records and then sort to put the remaining records in order.

uniq, whose job is to remove duplicated lines. More specifically, it removes adjacent duplicated lines

container & 가상화

컨테이너화

• 운영 체제 수준의 가상화
• 애플리케이션과 그 의존성을 OS 커널이 관리하는 자체 환경의 컨테이너로 패키징하는 과정

컨테이너

• 각 애플리케이션마다 전용 OS 필요 없이, 하나의 호스트 OS에서 실행 가능

• 각 애플리케이션이 고유한 환경을 가지며 다른 application과 의존성 충돌되지 않는다.

• 격리성 제공 → 다른 ps에 간섭하지 않는다

• 높은 이식성과 재현성

컨테이너 오케스트레이션

• 선언적 프로그래밍 사용(원하는 결과를 정의하고, 그 결과에 도달하는 절차는 직접 명시하지 않는 방식)
• 필요할 때 컨테이너 손쉽게 확장 가능
• 운영 자동화

데이터 모델 : 데이터셋의 논리적 구성과 해석을 정의

• 데이터와 그 구성 요소들이 서로 어떻게 관련되는지

Data Mesh & Data Febric

배치 처리

• 배치 & 스트리밍
• 고정된 시간 처리

스트리밍 처리 (고정 시간 윈도우 & 슬라이딩 시간 윈도우)

• 윈도우 : 연속적인 데이터 스트림을 시간 또는 크기 기준으로 나눈 파티션으로, 계산과 집계를 수행하기 위해 사용된다.

비용 모델

• pay-as-you-go : 쓴 만큼 내라
  • 네트워크 트래픽
  • 저장한 바이트 수
  • 보관 기간
  • 데이터 작업
    • 이동
• 예약 용량 : 구독료

비용 최적화

• 서비스 품질 유지하면서 비용 절감

NoSQL

표 형식 NoSQL 데이터 저장소

SELECT
    title,
    price
FROM books
WHERE pirce <50.00;

열 기반 데이터베이스에서 쿼리 처리 과정

1. price 열에 가서 조건절에 맞게 필터링
2. 반환되는 행에서 title 컬럼 값들을 뽑아서 필터링 결과로 반환
3. title, price 컬럼만 처리하여 반환

non-table NoSQL

문서형 데이터베이스 : 키-값, 키-배열, 키-객체 쌍으로 구성된 유연한 반정형 형식에 데이터를 저장

snowflake -micro partition :

• 각 마이크로 파티션에는 50~500 MB 사이의 압축되지 않은 데이터가 포함
  • 데이터 항상 압축되어 저장되기에 실제로는 더 작다
• 테이블의 행 그룹은 열 방식으로 구성된 개별 마이크로 파티션에 매핑된다.
• → 초대형 테이블의 매우 세분화된 정리 가능하도록 한다

이점

• 기존의 정적 파티셔닝과 달리 Snowflake 마이크로 파티션은 자동으로 파생됩니다. 명시적으로 사전에 정의하거나 사용자가 유지 관리할 필요가 없습니다.

• 이름에서 알 수 있듯이 마이크로 파티션은 크기가 작기 때문에(압축 전 50~500 MB), 매우 효율적인 DML 및 더 빠른 쿼리를 위한 세분화된 정리가 가능합니다.

• 마이크로 파티션은 값 범위에서 겹칠 수 있으며 균일하게 작은 크기와 결합되어 왜곡을 방지하는 데 도움이 됩니다.

• 열은 종종 열 저장소 라고 하는 마이크로 파티션 내에 독립적으로 저장됩니다. 이를 통해 개별 열을 효율적으로 스캔할 수 있습니다. 쿼리에서 참조하는 열만 스캔됩니다.

• 열은 또한 마이크로 파티션 내에서 개별적으로 압축됩니다. Snowflake는 각 마이크로 파티션의 열에 대해 가장 효율적인 압축 알고리즘을 자동으로 결정합니다.

나중에 다시 보자. 아직은 제대로 이해가 안 간다.

테이블에 저장된 데이터는 자연 차원(날짜, 지리)에 따라 정렬된다. → 쿼리 성능에 큰 영향 snowflake에서는 데이터가 테이블에 삽입되면서 클러스터링 메타데이터가 수집되고 프로세스 중에 생성된 각 마이크로 파티션에 대해 기록된다. 그 다음, 이 클러스터링 정보를 활용하여 쿼리 중 마이크로 파티션의 불필요한 스캔을 방지하고 이러한 열을 참조하는 쿼리의 성능을 크게 가속화한다.

데이터 클러스터링

• 유사한 데이터 포인트를 함께 구성/그룹화
• 데이터 적재 시 자동 수행

query pruning : 조건에 맞는 데이터가 없을 것 같은 파티션•파일•컬럼•마이크로 파티션은 아예 읽지 않는 최적화 기법

데이터는 사실과 수치의 비조직적이고 맥락 없는 집합이다.

• 아날로그와 디지털 형식이 있고
• 다양한 곳에서 데이터가 수집된다.
• → 원천 시스템 문서를 읽고 그 패턴과 특이점을 이해하자
• → RDBMS를 사용한다면 그 시스템의 작동 방식을 익히고 영향을 줄 수 있는 요소들을 파악하자.

파일 : byte의 sequence → disk에 저장된다.

• 로컬 매개변수, 이벤트, 로그, 이미지, 오디오 저장 주로 보는 파일의 형식은 엑실, csv, txt, json, xml
• 정형 : excel, csv
• 반정형 : json, xml, csv
• 비정형 : txt, csv + parquet, ORC 등

API(application programming interface)

시스템 간 데이터를 교환하는 표준 방식 #API

#OLTP

짧은 지연시간과 높은 동시성 지원

#ACID : 원자성, 일관성, 독립성, 내구성

일부 분산형 데이터베이스는 최종 일관성과 같은 완화된 일관성 제약 조건을 사용하기도 한다.

원자적 트랜잭션 : 트랜잭션이 진행됨에 있어서, 모든 트랜잭션이 성공하든가 모두 실패해야 한다. → 전체 작업이 트랜잭션으로서 발생해야 한다.

#OLAP OLAP에서도 OLTP처럼 여러 쿼리문이 실행되면 리소스 경쟁이 일어난다. 하지만 OLTP는 같은 엔진/노드에서 경쟁이 일어나서 UX가 깨진다는 게 문제고, OLAP는 멀티 노드 MPP 구조로 느려지긴 해도 그렇게 큰 문제는 되지 않는다.

OLAP의 Online 부분은 시스템이 들어오는 쿼리를 지속해 수신 대기한다는 뜻으로 OLAP 시스템이 대화형 분석에 적합함을 의미한다.

그런데, 종종 원천 시스템이 아니라 DWH에서 다시 원천 시스템으로 데이터를 보내야 할 때가 있는데, 이때 역뱡향 ETL 워크플로를 OLAP 시스템이 제공할 수 있다.

#log : 최소한 누가, 무엇을, 언제 수행했는지 수집해야 한다. 인코딩 방법

• 바이너리 인코딩 로그
• 반정형 로그
• 일반 텍스트 로그

로그 해상도 : log에 캡쳐된 이벤트 데이터의 양 다 저장하면 실용적이지 않으니까 → 특정 유형의 커밋 이벤트가 발생한 사실만 기록할 수 있다.

log level : 로그 엔트리를 기록하는 데 필요한 조건, 특히 에러와 디버깅에 관한 ==조건이다==

reference) Fundamentals of Data Engineering

현대 데이터 아키텍쳐

요구사항

• 유연성과 확정성
• 클라우드로 확장
• 증가하는 데이터 처리 가능
• 핵심 비즈니스 경로
  • 청구
• 분산 도메인 통합
• 데이터 거버넌스와 보안

Airflow 연산자

EmptyOperator : 문제 해결을 위한 task나 아직 구현되지 않은 task를 표현하는 데 사용

BashOperator : 지정된 Bash 명령어나 스크립트 실행

• 워크플로우 맥락에서 의미가 있다면, Bash가 할 수 있는 거의 모든 동작을 수행할 수 있다.

실제 액션 정의

Airflow Task

• operator를 instance화한 실제 실행 단위
• Dag 내부에서 정의되고, task 간 연결 정립
• 한 task는 하나의 책임으로 명확하게 정의 내릴 것!
• 대용량 데이터를 보내는 것보다는 메타데이터만 보내고 실제 데이터는 s3와 같은 스토리지에서 사용
• 의존성 확립 중요 (Upstream, Downstream 설정, 방향을 명확하게 설정할 것!)

t1 >> t2 : task1 진행 후, task2

• t1 : upstream
• t2 : downstream

t1 >> t2 << t3 t1, t3가 끝나야 t2 진행

python operator에서 op_kwargs 키워드 인자 딕셔너리의 키와 함수의 이름은 항상 일치해야 한다.

def pull_file(URL, savepath):
    r = requests.get(URL)
    with open(savepath, 'wb') as f:
    f.write(r.content)
    # Use the print method for logging
    print(f"File pulled from {URL} and saved to {savepath}")

from airflow.operators.python import PythonOperator

# Create the task
pull_file_task = PythonOperator(
    task_id='pull_file',
    # Add the callable
    python_callable=pull_file,
    # Define the arguments
    op_kwargs={'URL':'http://dataserver/sales.json', 'savepath':'latestsales.json'}
)

python_callable는 키워드 인자로 pull_file()을 인자로 함수에 괄호를 붙여서 보내면 바로 함수가 실행이 되면서 반환값이 전달되기 때문에 에러가 뜬다. 바로 실행되지 않고 함수만 전달되도록 pull_file만 전달한다.

Airflow cron 기본 문법 (5자리)

Airflow cron은 기본적으로 유닉스 cron과 같은 5필드를 씁니다.

* * * * * 분 시 일 월 요일 각 필드 의미:

• 분(minute): 0-59
• 시(hour): 0-23
• 일(day of month): 1-31
• 월(month): 1-12 또는 JAN-DEC
• 요일(day of week): 0-6 또는 SUN-SAT (0/7 = 일요일)

자주 쓰는 패턴:

• * : 가능한 모든 값 (매 분, 매 시 등)
• , : 여러 값 지정 (예: 1,2,5)
• - : 범위 (예: 1-5 = 월~금 요일)
• */n: n 간격 (예: */5 = 5분마다)

자주 쓰는 Airflow cron 예시

Airflow DAG 정의에서:

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="example_cron",
    start_date=datetime(2025, 1, 1),
    schedule_interval="0 0 * * *",  # 매일 0시
    catchup=False,
) as dag:
    ...

매일 0시 : 0 0 * * * 매시간 정각 : 0 * * * * 매 5분 : */5 * * * *

preset 문자열도 지원

• None : 스케줄 없이 수동 / 외부 트리거로만 실행
• @once : 한 번만 실행
• @hourly : 매시간 정각
• `@daily : 매일 0시
• `@weekly : 매주 일요일 0시
• @monthly : 매달 1일 0시
• @yearly : 매년 1월 1일 0시 0 0 1 1 *

센서 (Sensor)

특정 조건이 참이 될 때까지 계속 체크(polling)만 하는 특수한 operator

• 조건이 만족되면 success로 끝나고 그 뒤의 downstream task들이 실행
• EX
  • FileSensor : S3/FTP/local에 특정 파일이 생겼는지
  • SqlSensor : table에 레코드가 생겼는지
  • ExternalTaskSensor : 다른 DAG/Task가 완려되었는지

Sensor 동작 방식 (poke vs reschedule)

공통 주요 파라미터

• poke_interval : 몇 초마다 조건을 체크할지(default = 60s)
• timeout : 최대 대기 시간(지나면 실패, 단위 : seconds)
• mode : "poke" 또는 "reschedule"
• soft_fail : 실패 시 FAILED 대신 SKIPPED로 처리할지 결정

mode 차이

• poke 모드(default)
  • sensor task가 돌아가는 동안 계속 워커 슬롯 점유
  • 지연 시간(조건 충족 후 반응하는데까지 걸리는 시간)이 짧은 대신, 워커 자원을 많이 사용
• reschedule모드
  • 체크할 때만 잠깐 점유하고 확인하고, 나머지는 자원 반납
  • 워커 자원 효율 좋지만, poke_interval 단위로 약간의 지연 존재

Example Code

from airflow import DAG
from datetime import datetime
from airflow.provides.common.sql.sensors.sql import SqlSensor
from airflow.operators.python import PythonOperator

def process_data():
    pritn("데이터 처리 시작")

with DAG(
    dag_id='example_sql_sensor',
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",
    catchup=False,
) as dag:

    wait_for_partition = SqlSensor(
        task_id="wait_for_partition",
        conn_id="postgres_default",
        sql="""
            SELECT 1
            FROM partitions
            WHERE dt = {{ ds }}
        """,
        poke_interval=60,       # 60s마다 체크
        timeout=60 * 60 * 3,    # 최대 3시간 기다리기
        mode="reschedule",      # 워카 자원 아끼기
    )

    run_processing = PythonOperator(
        task_id="run_processing",
        python_callable=process_data,
    )

    wait_for_partition >> run_processing

→ 이 DAG는 매일 1회 실행되면서, 해당 날짜의 partition이 DB에 생길 때까지 seonsor가 기다리고, 이후에 처리 Task를 실행하게 된다.

Airflow Debug & TroubleShooting

1. DAG 레벨
   • DAG가 UI에 안 보인다? → DAG parsing/import error
   • airflow dags list-import-errors로 확인 가능
2. Task 레벨
   • task 상태와 log 확인
3. scheduler/worker 레벨
   • 스케줄러가 task를 잡아 주는지, 워커가 실제로 실행되는지 확인
4. 외부 시스템(파일, DB, API) 레벨
   • sensor, hook, operator가 의존하는 S3, DB, API 쪽 문제

이 순서대로 좁히면서 버그 위치를 찾는 게 Debug 핵심!

Debugging Tool & Pattern

1. Airflow UI
2. CLI로 개별 Task 테스트
   • airflow tasks test <dag_id> <task_id> <execution_date>
   • 메타데이터 DB 상태와 상관 없이 로컬에서 그 task만 실행해 보는 용도
3. dag.test()
   • dag 파일 맨 아래 다음 추가 후, IDE/로컬에서 실행

     if __name__ == "__main__":
         dag_test()

     • 전체 dag를 하나의 프로세스에서 순서대로 실행해서, 어디에서 에러가 나는지 확인
4. DebugExecuter / 로컬 개발 환경

• AIRFLOW__CORE__EXECUTOR=DebugExecutor로 두고, SQLite+ 단일 프로세스로 디버깅용 실행

SLA(Service Level Agreement)

• task 또는 DAG가 실행에 걸려야 하는 예상 시간
• SLA Miss : task, DAG가 예상 시간 내 완료되지 못한 경우

SLA 정의하는 방법

1. task에서 sla 인자 사용

   task1 = BashOperator(
            task_id='sla_task',
            bash_command='runcode.sh',
            sla=timedelta(seconds=30),
            dag=dag)

2. default_args 딕셔너리에 설정

   default_args={
    'sla': timedelta(minutes=20),
    'start_date': datetime(2023, 2, 20)
   }
   dag = DAG('sla_dag', default_args=default_args)

Template이 적용된 BashOperator

Jinja Template

Airflow는 내부적으로 Jinja2라는 파이썬 template engine을 사용한다. 이를 통해 Bash 명령어 안에 {{}}형태의 중괄호를 사용하면, Airflow가 이를 실제 값으로 치환해준다.

자주 사용하는 템플릿 변수

BashOperator의 bash_command 내에서 가장 많이 쓰이는 변수들입니다.

macros 변수

Airflow template에서 유용한 객체나 메서드에 대한 reference를 제공

• macros.datetime ← 파이썬의 datetime.datetime 객체
• macros.timedelta ← timedelta 객체 참조
• macros.uuid ← python의 uuid 객체와 동일
• macros.ds_add 와 같은 추가 함수도 존재 ← 템플릿 안에서 날짜 계산 간단히 할 수 있도록 도와준다
  • {{ macros.ds_add('2020-05-15', 5) }} : 날짜에 일 수 더하기

python의 uuid 객체란? Universally Unique Identifier(범용 고유 식별자)의 약자로, 네트워크 상에서 서로 다른 시스템들이 독립적으로 식별자를 생성하더라도 중복될 확률이 거의 없도록 설계된 128비트 길이의 숫자이다.

DB의 기본키, 세션 ID, file name 등 절대 중복되면 안 되는 고유값이 필요할 때 사용한다!

고급 template

templated_command="""
{% for filename in params.filenames %}
    echo "Reading {{ filename }}"
{% endfor %}
"""

Jinja 구문에서 for 루프의 끝을 나타내려면 → {% endfor %}

Branch(분기)

브랜칭 : 조건부 로직 가능토록 한다. BranchPythonOperator 사용

• from airflow.operators.python import BranchPythonOperator 다음에 실행할 task id, (id 목록)을 반호나하는 python_callable을 받는다.

python_callable에서 중괄호를 안 쓰는 이유 BranchPythonOperator의 python_callable 인자는 문자열이 아니라 파이썬 함수 객체 그 자체를 전달받습니다. - 동작 방식: 이 인자에는 함수의 "이름"을 넘겨줍니다. Airflow는 이 함수를 나중에 직접 호출(Call)합니다. - 중괄호를 쓰지 않는 이유: - 함수는 문자열이 아님: {{ }}는 문자열 내부의 텍스트를 바꿀 때 쓰는 문법입니다. 함수 객체 자체에는 적용되지 않습니다. - 런타임 실행: 함수 내부에서 날짜 같은 정보가 필요하다면, Airflow는 함수를 호출할 때 context라는 딕셔너리에 모든 정보를 담아 보내줍니다. 함수 안에서 직접 꺼내 쓰면 되기 때문에 굳이 중괄호로 치환할 필요가 없습니다.

provide_context의 주요 역할 Airflow는 함수를 호출할 때 Context라는 거대한 딕셔너리를 인자로 전달한다.

• 날짜 정보 : ds, logical_date, execution_date
• 객체 정보 : dag, task
• task 간 통신 : ti, task_instance(Xcom을 사용해 다른 task의 데이터를 가져올 때 필수)

def check_weekend(**kwargs):
    dt = datetime.strptime(kwargs['execution_date'],"%Y-%m-%d")
    # If dt.weekday() is 0-4, it's Monday - Friday. If 5 or 6, it's Sat / Sun.
    if (dt.weekday() < 5):
        return 'email_report_task'
    else:
        return 'no_email_task'

branch_task = BranchPythonOperator(task_id='check_if_weekend',
    python_callable=check_weekend,
    provide_context=True,
    dag=dag)

production pipeline 구축하기

DAG 및 task 실행

• command line에서 특정 task 실행 : airflow tasks test <dag_id> <task_id> <date>
• 전체 DAG 실행 airflow dags trigger -e <date> <dag_id> → 지정한 날짜에 전체 DAG가 실행하는 것처럼 동작한다.

Operator 요약

• Bashoperator → bash_command 필요
• PythonOperator → python_callable 필요
• BranchPythonOperator → python_callable, provide_context=True 필요, 호출 함수는 **kwargs를 받아야 함.
• FileSensor → filepath 인자 필요

Building a Retail Data Pipeline(Project)

🛒 식료품 매출 데이터 처리 요구 사항

1. transform() 함수 구현

• 입력: merged_df (데이터프레임)

• 수행 작업:

  • 수치형 데이터의 결측치(Missing values)를 원하는 방식(예: 0 또는 평균값 등)으로 채웁니다. O
  • Month(월) 컬럼을 새로 추가합니다. O
  • 주간 매출(Weekly_Sales)이 $10,000를 초과하는 행만 유지합니다. O
  • 분석에 불필요한 컬럼들을 삭제합니다.
    • "Store_ID"
    • "Month"
    • "Dept"
    • "IsHoliday"
    • "Weekly_Sales"
    • "CPI"
    • ""Unemployment""

• 출력: 최종 데이터프레임을 반환하며, 결과는 clean_data라는 변수에 저장되어야 합니다.

2. avg_weekly_sales_per_month() 함수 구현

• 입력: clean_data (위에서 정제된 데이터프레임)

• 수행 작업:

  • 월별 평균 매출을 계산합니다.

  • 분석에 필요한 Month와 Weekly_Sales 컬럼만 선택합니다.

  • 메서드 체이닝(Chain operation)을 사용하여 다음 함수들을 순서대로 적용해야 합니다:

    1. groupby(): "Month" 컬럼을 기준으로 그룹화
    2. agg(): 평균 매출 계산
    3. reset_index(): 인덱스를 새로 재설정
    4. round(): 결과를 소수점 둘째 자리까지 반올림

3. load() 함수 구현

• 입력: 정제된 데이터프레임(clean_data), 집계된 데이터프레임(agg_data), 그리고 각각의 저장 경로

• 수행 작업:

  • 두 데이터프레임을 각각 clean_data.csv와 agg_data.csv 파일로 저장합니다.

  • 저장 시 인덱스(index)는 포함하지 않습니다.

4. validation() 함수 구현

• 수행 작업:

  • load() 함수를 통해 생성된 두 개의 CSV 파일이 현재 작업 디렉토리에 실제로 존재하는지 확인합니다.

참고 사항:

• 데이터베이스 연결을 위한 별도의 엔진 설정은 필요하지 않습니다.

• 제공된 SQL 코드 셀에 쿼리를 실행하면 결과가 자동으로 grocery_sales라는 이름의 Pandas 데이터프레임으로 저장되며, 이를 바로 Python 코드에서 사용할 수 있습니다.

def avg_weekly_sales_per_month(clean_data):
    df = clean_data.groupby(by="Month").agg('mean').reset_index().round(2)
    return df

error → agg(mean)이라고 작성했는데 agg('mean')이라고 작성해야 한다

agg(mean) → 파이썬은 mean이라는 이름을 가진 변수나 객체를 찾으려고 하는데, 이전에 정의해두지 않았으면 이를 찾지 못하고 NameError가 발생시킨다.

os.path.exists → csv 파일 validation 확인

정규표현식 r'(\d+\.?\d*)'

• \d : 숫자
• + : 하나 이상 반복됨
• [0-9] : \d와 같은 의미
• \.? : 마침표 + 0개 또는 1개 → 소수점이 있을수도 있고 없을 수도 있고(?)
• \d* : 숫자 + 0개 이상 → 소수점 뒤에 숫자가 붙을 수도 있고 없을 수도 있다

정규표현식에서 특정 문자열 뒤의 숫자 찾으려면? → 캡쳐 그룹 사용 : 특정문자열(\d+\.?\d*)

DAG(Directed Acyclic Graph) : 방향 비순환 그래피

• Airflow에서 워크플로를 구성하는 작업 집합
• 작업과 작업 간 의존성으로 구성
• 메타데이터와 함께 생성

  etl_dag = DAG(
    dag_id='etl_pipeline',
    default_args={"start_date":"2023-11-15"}
  )

airflow tasks test <dag_id> <task_id> [execution_date]

DAG 는 순환하지 않는다. 한 번의 실행에서 각 1회만 실행된다. Airflow DAG는 오퍼레이터, 센서 등 실행할 구성 요소들로 이루어져 있고, 보통 이를 task라고 한다.

1. from airflow import DAG
2. DAG 구성 요소에 적용할 속성들을 담은 기본 인자 딕셔너리 생성 - Airflow의 런타임 동작을 세미할게 제어
3. Python 컨텍스트 매니저를 사용해 DAG 객체를 정의한다.

command line VS python

command line

• Airflow process start
• DAG/Task 수동 실행
• Airflow 로그 정보 확인

Python

• DAG 생성
• DAG 속성 개별 편집

datetime 함수 인자로 연도, 월, 일을 각각 정수형 인자로 받아야 한다.

dag list 확인 : airflow dags list

usecols 키워드 인자 : import할 모든 열의 이름 리스트나 열 번호 리스트를 전달

• 함수를 전달해 열 선택 가능

nrows 인자 : 임포트되는 행의 수 선택

• 파일을 chunks 단위로 처리하기 위해 skiprows 인자와 결합할 때 유용하다

skiprows : 건너뛸 행 번호 list, 행 건너뛸지 결정하는 함수, 건너뛸 행의 개수를 인자로 받는다

import된 첫 번째 행을 자동으로 header(column_name)으로 만든다. → 컬럼 이름이 포함된 행을 건너뛴다면 header=None으로 지정해야만 한다.

열 이름이 없을 때, 이름을 할당하려면 read_csv의 다른 인자인 names를 사용한다. names 인자 : 사용할 열 이름 리스트를 인자로 받는다.

• 데이터의 모든 열에 대한 이름이 포함되어야 한다.
• 일부 이름을 바꾸고 싶으면 import 작업이 끝난 후 진행

열 개수가 적은 데이터셋의 경우, Data Dictionary를 참고해 직접 리스트를 만들 수도 있다.

오류 및 결측치 처리

• 데이터 타입 지정 실수 (zipcode를 문자열이 아니라 정수로 pandas가 판단하는 문제) read_csv의 키워드 인자인 dtype으로 열의 데이터 타입을 지정할 수 있다. - dictionary 형태

• 결측치 + 결측치라 판단해야 하는 값 (zipcode = 0인 경우) na_values 키워드 인자 → 단일 값, 값들의 리스트, 컬럼-결측치 값으로 구성된 딕셔너리를 전달할 수 있다.

오류가 있는 행

• on_bad_lines
• error_bad_lines : 실행 X + 에러문
• warn_bad_lines : 실행 O + 경고

pd.concat 함수는 여러 개의 pandas 객체(DataFrame이나 Series 등)를 하나로 합칠 때 사용하는 함수입니다. 이 함수의 첫 번째 인자는 반드시 DataFrame들의 리스트(또는 다른 iterable)여야 합니다.

boolean 값 설정 true_values=['Yes'], false_values=['No'] 인자 → NA가 True로 변환되는 문제는 여전하다.

+) dtype=bool보다는 dtype=boolean을 사용하면 True, False, NA를 모두 유지할 수 있어 데이터 품질 관리에 유리하다.

JSON

• record orientation : 레코드 방식의 JSON은 딕셔너리들의 리스트로 구성된다. 각 dictionary는 테이블의 한 행이 된다.
• column orientation : key-column name, value - 해당 컬럼의 데이터 리스트나 행 인덱스가 포함된 딕셔너리

실무에서는 Record 방식으로 데이터가 들어오지만 복잡하게 중첩된 경우에는 json_normalize라는 tool을 사용한다.

requests.get()의 결과값은 데이터와 메타데이터를 포함하는 response 객체이다. 여기서 실제 데이터만 뽑아내려면 .json() 메서드를 사용해야 하는데, 중요한 점은 .json() 메서드가 dictionary를 반환한다는 것이다. pd.read_json()은 문자열을 기대하기 때문에 dictionary를 직접 분석할 수 없다. → pd.DataFrmae()을 사용한다.

pd.json_normalize : dictinoary나 dictionary의 list를 받아 데이터프레임으로 반환한다. 중첩된 속성의 열 이름은 속성.하위속성을 따르지만 . 구분자는 판다스의 열 선택 문법과 충돌할 수도 있기 때문에, sep인자를 사용해 다른 구분자로 지정하는 것이 좋다.

df = json_normalize(data['job'], sep='_')

{
    "store_name": "Cafe Gemini",
    "location": "Seoul",
    "reviews": [
        {"user": "Alice", "score": 5},
        {"user": "Bob", "score": 4}
    ]
}

record_path : reviews 컬럼을 2개의 행으로 쪼개준다. 없으면 reviews 컬럼 하나에 리스트가 통째로 들어간다.

record_path는 보통 meta 인자와 함께 사용한다. reviews의 리스트를 펼치게 되면 {"user": "Alice", "score": 5}에 해당하는 상위 수준의 정보(가게이름, 위치)가 사라지기 때문에 meta를 이용해 그 정보를 다시 붙여줘야 한다.

다른 상위 정보의 하위 정보를 붙이려고 할 때는 리스트를 활용한다.

flat_cafes = json_normalize(data["businesses"],
    sep="_",
    record_path="categories",
        meta=['name',
        'alias',
        'rating',
        ['coordinates', 'latitude'],
        ['coordinates', 'longitude']],
        meta_prefix="biz_")

['coordinates', 'latitude'] : coordinates 안에 있는 latitude의 정보

cafes = pd.concat([top_50_cafes, next_50_cafes], ignore_index=True)

• pd.concat은 DataFrame의 리스트를 인자로 전달해야 한다.

#Git

git branch new_name : 브랜치 생성 git switch -c new_branch : branch 생성 후 해당 branch로 이동![[스크린샷 2026-04-10 오후 4.48.40.png]]

git branch -m old_name new_name : branch 이름 변경 git branch -d delete_branch_name : branch 삭제 병합되지 않은 채면 오류가 뜨는데, -D 플래그 ㅅ용하면 삭제 가능하다.

병합

source : 병합의 원본 브랜치 destination : 병합의 대상 브랜치

piano를 main에 병합할 때:

• piano - source
• main -destination

destination 브랜치에 이동한 후에 git merge 병합시킬_브랜치 자기가 사용 중인 브랜치를 다른 브랜치에 병합시키려면 git merge using_branch destination

게시판 기능을 만들기 위해 borad 브랜치를 만들고 병합했는데 에러가 생겼어 그러면 board 브랜치에서 고치는 게 아니라 board는 삭제하고 board_fix 브랜치를 만들고 작업 후에 다시 병합한다.

병합 충돌

서로 다른 두 파일을 병합하면 git이 어느 걸 병합해야 할 지 알 수 없다. → 충돌!

원격의 파일을 로컬 저장소와 동기화하려면 원격에서 브랜치를 가져와야 한다. 이를 위해서 원격 이름을 지정해 git fetch를 사용한다.
git fetch origin

• 원격의 모든 브랜치가 로컬 저장소에 내려온다.
• 브랜치가 원격에만 존재했다면, 로컬에도 생성된다.
• 원격의 내용을 로컬에 병합하지는 않음에 주의해라.

git fetch origin main origin 원격 저장소 중 특정 브랜치만 가지고 오고 싶을 때는 브랜치도 같이 선언해라.

브랜치를 가지고 왔으면 저장소 간 내용을 동기화해야 한다. git merge origin

• local branch를 명시하지 않으면 현재 위치한 브랜치에 병합된다. git pull origin

git pull origin dev : origin의 dev 브랜치에 pull

원격에서 pull 하기 전에 로컬 작업을 저장해둬야 한다.

git push

원격에 push 하려면

• 먼저 로컬에 변경 사항 저장
• git push remote local_branch : local_branch에서 remote로 push
• git push origin main : 로컬 main에서 origin으로 push

원격에 없는 로컬 브랜치가 있다면? git push origin local_branch_name

Python으로 배우는 software engineering principle

모듈성, 문서화, 테스트

PyPi : python package index

package 이름은 소문자로 설정

__init__.py 가 있어야 python이 우리가 만든 패키지를 인식한다.

utils.py - 서브 모듈로 불리며 pakcage_name.sub_module_name.function_name 형식으로 호출할 수 있다.

또 다른 방법으로는,

# my_package/__init__.py
form .utils import we_need_to_talk

을 설정해두면, 다른 work directory에서 작업할 때

import my_package

만 해도 import 할 수 있다. init 파일에서 import를 대신 해주기 때문. 핵심 기능을 __init__.py에서 임포트하여 바로 접근할 수 있도록 해주고 비주류 함수는 서브모듈 점 표기법으로 접근토록 한다.

package 내부 package도 설정할 수 있는데, 방법은 똑같다.

python 패키지 공유

• setup.py : package를 어떻게 설치할 지
• requirements.txt : 필요한 환경 재현하는 방법 pip install -r requirements.txt

setup.py 설정되면 터미널에서 같은 디렉토리 내 터미널에서 pip install . → 패키지 설치 된다.

DRY 원치 Don't Repeart Yourself : 반복할만한 코드 재작성하지 마라. → 상속

from .parent_class import ParentClass

class ChildClass(ParentClass):
    def __init__(self):
        ParentClass.__init__(self)
        # Add child's unique thiing

다중 상속 개념도 알아둘 것!

주석은 코드가 무엇을 하는지가 아니래 왜 코드를 그렇게 작성헀는지 써라

test doctest & pytest docstring → doctest

객체 2개를 비교할 때는 ==로 하기보다는 속성을 비교하는 방식으로 검증한다.

test.py를 작성할 때는 소스코드 제목은 시작이나 끝이 test 이어야 하고, 함수 정의할 때는 test가 앞에 붙어야 하며, assert로 검증해줘야 한다.

Sphinx로 docstring을 문서화할 수 있다.

ivar 키워드 : 인스턴스 변수

• 변수라고 하지 않고 ivar라고 하는 이유 : 선언되는 위치에 따라 이름이 달라지기 때문에

  def __init__(self, table_name):
    # 여기서 self.table_name이 ivar 변수다.
    self.table_name = table.name

Travix CI 새 코드를 추가할 때, Travis가 자동으로 테스트 실행하고, 변경으로 문제가 생기면 알려준다. 수정한 걸 push 하면 다시 테스트를 실행해 수정이 제대로 됐는지 확인해준다. 빌드 예약도 가능

Codecov : 자동 테스트가 코드의 어떤 부분을 검증하고 있는지 살펴볼 수 있게 해주는 도구

• 테스트 커버리지라고 한다.

Code Climate : 가독성 개선을 위한 코드 분석

Performing a Code Review

• 리팩토링 (DRY 원칙 적용): 시각화 함수 내에서 중복되는 라벨 생성 로직을 column_to_label() 호출로 대체합니다.

• 유닛 테스트 수정: prepare_smartphone_data() 함수가 실제로 NaN을 어떻게 처리하는지(예: dropna()를 사용하는지 등) 확인하고, 테스트 코드의 assert 문이 그 로직을 정확히 검증하도록 수정합니다.

• 최종 검증: pytest를 실행하여 결과가 ExitCode.OK(보통 0)를 반환하는지 확인합니다.

docstring에 example 추가할 지?

Python으로 ETL과 ELT

변환 검증

• .nsmallest(10, ['timestamps'] : 지정한 열 목록에서 가장 작은 값 10개
• .nlargest(10, ['timestamps']

to_csv 했을 때 저장 제대로 되었는지 확인하려면

import os
print(os.path.exists('example.csv'))

log : 파이프라인이 실행되는 동안 생성되어 기록되는 메세지

• 실패 시 성능 기록
• 실패 시 원인 파악 출발점 제공

logging - 6가지 debug, info, warning, error

• debug : 파이프라인 만드는 동안 사용 → 차원, 타입, 변수 값 제공
• info : 파이프라인 실행 전반에 걸쳐 기본 정보와 체크포인트 제공
• warning : 예기치 않은 일 - 이전에 보지 못한 데이터 타입) ex
• erro : 데이터 타입 변경, 데이터 제공 X

json → dictionary → DataFrame

isinstance(pipeline, data_type) : 객체와 해당 타입 일치하면 True

@pytest.fixture() : test data와 객체를 여러 테스트에서 공유할 수 있게 해주는 함수

logging.basicConfig는 파이썬의 표준 라이브러리인 logging 모듈에서 로그 시스템의 가장 기본적인 설정을 한 번에 마치는 함수

• level: 어떤 수준의 로그부터 출력할지 결정합니다. (DEBUG < INFO < WARNING < ERROR < CRITICAL)
• format: 로그가 출력될 모양을 지정합니다. (시간, 로그 레벨, 메시지 등)
• filename: 로그를 콘솔이 아닌 파일에 저장하고 싶을 때 파일 경로를 지정합니다.
• filemode: 파일을 열 때의 모드입니다. ('a'는 이어쓰기, 'w'는 새로 쓰기)

date() →연, 월, 일 date(2000, 10, 16), date(2000, 8, 24)

from datetime import timedelta
# timedelta : 사건 사이에 경과한 시간
td = timedelta(days=29)
print(d1 + td)

sort_values() : Dataframe이나 series에서 사용하는 메서드

list 타입을 정렬하려면 내장 함수 sorted()나 sort()를 사용해야 한다.

• sorted() : 정렬된 새로운 리스트 반환
• sort() : 리스트 자체 정렬하지만 반환값은 없다

dt.replace(tzinfo=timezone.utc))랑 dt.astimezone(timezone.utc))의 차이

dt.replace(tzinfo=timezone.utc)) : 시각은 그대로 둔 채 '이 시각은 이제부터 이 시간대야'라고 이름표만 갈아끼우는 것이다.

• datetime 객채 값은 변하지 않고, tzinfo 속성만 덮어쓴다.
• 시간대 정보가 없는 Naive 데이터에 이건 이러한 UTC 시간이야-라고 정의할 때 사용
• 이미 시간대 정보가 있는 객체에 사용하면, 실제 시점이 변해버린다. → 서울 오후 3시를 replace(tzinfo=UTC)라고 하면 영국 시간(UTC) 오후 3시가 되어서 시간이 어긋난다. 잘못 설정했을 때는 이게 맞지.

dt.astimezone(timezone.utc)) : 실제 시점을 유지하면서 다른 시간대의 시각으로 계산하는 것

• UTC 기준으로 시각을 더하거나 빼서 실제 같은 순간을 가르키는 다른 지역의 시간을 계산
• 서울 오후 3시 → 영국 시간으로 변환할 때
• Naive 객체에 사용하면, 파이썬은 이 객체를 시스템의 로컬 시간대로 간주하고 변환을 시도한다.

astimezone - datetime 객체에서 사용하는 메서드

tz_convert : Pandas의 Series나 DatetimeIndex에서 사용한다.

timezone 객체를 생성할 때는 timezone(timedelta(hours=-8)) 이라면, datetime 객체를 생성할 때는 dt = datetime(2017, 10, 1, 15, 26, 26, tzinfo=pst)

tzinfo ← datetime 객체에서

campaign.csv

economics.csv

import pandas as pd
import numpy as np
from datetime import datetime

df = pd.read_csv('bank_marketing.csv')
client = df[['client_id', 'age', 'job', 'marital','education', 'credit_default', 'mortgage']]

client['job'] = client['job'].str.replace('.','_')
client['education'] = client['education'].replace({'.':'_', 'unknown':np.NaN})
client['credit_default'] = (client['credit_default'] == 'yes').astype(int)
client['mortgage'] = np.where(client['mortgage'] == 'yes', 1, 0)

client.to_csv('client.csv')

campaign = df[['client_id', 'number_contacts', 'contact_duration', 'previous_campaign_contacts','previous_outcome', 'campaign_outcome']]

campaign['previous_outcome'] = (campaign['previous_outcome'] == 'success').astype(int)
campaign['campaign_outcome'] = (campaign['campaign_outcome'] == 'yes').astype(int)
campaign['last_contact_date'] = pd.to_datetime(campaign['last_contact_date'], format='YYYY-MM-DD')

campaign.to_csv('campaign.csv')


economics = df[['client_id', 'cons_price_idx', 'euribor_three_months']]
economics.to_csv('economics.csv')

error1 : Expected the credit_default column in the client.csv file to be bool data type.

client['credit_default'] = (client['credit_default'] == 'yes').astype(int)

# astype(bool)

wrong :

client['education'] = client['education'].str.lower().replace({'.':'_', 'unknown':np.NaN})

![[스크린샷 2026-04-09 오후 9.35.15.png]] 실제로 체크해보니까 안 바뀌었다.

str.lower().replace → str.replace()가 아니라 .replace()로 받아들여진다. 이는 값 전체가 일치할 때, 처리에 적합하다. 따라서, unknown → NaN 처리에는 적합하나, 문자열의 일부인 "."를 "_ " 로 바꾸는 데는 적합하지 않다.

문자열의 일부를 바꿀 때는 str.replace() 메서드를 활용한다.

client['education'] = client['education'].str.lower().replace('unknown', np.NaN)
client['education'] = client['education'].str.replace('.', '_')

answer

import pandas as pd
import numpy as np
from datetime import datetime

df = pd.read_csv('bank_marketing.csv')
client = df[['client_id', 'age', 'job', 'marital','education', 'credit_default', 'mortgage']]

client['job'] = client['job'].str.replace('.','_')
client['education'] = client['education'].str.lower().replace('unknown', np.NaN)
client['education'] = client['education'].str.replace('.', '_')
client['credit_default'] = (client['credit_default'] == 'yes').astype(bool)
client['mortgage'] = np.where(client['mortgage'] == 'yes', 1, 0).astype(bool)

client.to_csv('client.csv', index = False)

campaign = df[['client_id', 'number_contacts', 'contact_duration', 'previous_campaign_contacts','previous_outcome', 'campaign_outcome']]

campaign['previous_outcome'] = (campaign['previous_outcome'] == 'success').astype(bool)
campaign['campaign_outcome'] = (campaign['campaign_outcome'] == 'yes').astype(bool)
# campaign['last_contact_date'] = pd.to_datetime(campaign['last_contact_date'], format='YYYY-MM-DD')
campaign['last_contact_date'] = pd.to_datetime(
    "2022-" + df['month'].str.lower() + "-" + df['day'].astype(str)
).dt.strftime('%Y-%m-%d')
campaign.to_csv('campaign.csv', index = False)


economics = df[['client_id', 'cons_price_idx', 'euribor_three_months']]
economics.to_csv('economics.csv', index = False)

효율적인 python 코드 작성

Goal : 지연 시간과 오버헤드를 줄인다

unpack 연산자 (" * ")

nums = range(1,11,2)
n_list = list(nums)


# unpack 연산자
num_list = [*range(1,11,2)]

enumerate(iterable, start=0)

• iterable (필수) : 반복 가능한 객체(list, tuple, string, dictionary …)
• start(선택) : 인덱스 시작 번호

[enumerate(names,1)] → 함수 실행하면 리스트가 아니라 enumerate object라는 특수한 객체 반환한다. → 주소 반환 → 이거는 iterator로 호출하면 결과값을 내보일 준비가 됐다. → 결과값을 보기 위해서는 그 안의 내용물을 꺼내야 한다. unpack 연산자는 안의 결과값을 꺼내서 나열하는 연산자다.

일종의 Lazy Evaulation이다. 필요할 때까지 계산을 미루는 것! 이게 나중에 spark나 snowflake 등 모든 곳에 쓰이는 핵심 원리!

str.upper()처럼 괄호를 붙이면, str.upper 메서드를 즉시 호출하려고 시도합니다. 하지만 str.upper()는 문자열 인스턴스가 필요합니다(예: 'abc'.upper()). map() 함수에는 함수 자체(즉, 호출하지 않은 상태)를 전달해야 합니다. 즉, str.upper처럼 괄호 없이 함수 객체를 전달해야 각 요소에 대해 나중에 호출할 수 있습니다.

함수 객체와 함수 호출의 차이 괄호를 붙이면 함수가 바로 실행되고, 괄호 없이 전달하면 함수 자체를 전달하는 것이다.

numpy

numpy 배열은 동종으로, 모든 원소가 같은 타입이어야 한다.
맞추지 않으면, nupy가 알아서 변환한다. python 내장 list는 브로드캐스팅을 지원하지 않는다.

실행 시간

%timeit : 분석하고 싶은 줄 앞에 매직 커맨드를 붙이면 된다. 시간 통계를 평균으로 제공한다.(평균 + 표준편차 제공)

• -r (runs) : 실행 횟수 설정
• -n (loops) : 루프 수 설정 여러 줄 → %%timeit
• -o : %timeit의 출력을 변수에 저장할 수 있다.

코드 프로파일링

• 함수 호출 빈도와 소요 시간의 상세 통계
• 한 줄 단위 분석
• line_profiler 설치 필요

메모리 사용 코드 프로파일링 memory_profiler

• 메모리 사용량에 대한 상세 통계
• 분석하고자 하는 함수는 반드시 import 해야 한다.

효율적 결합, 세고, 반복

zip : 객체들을 하나로 맞물려 결합시킨다. → zip 객체 반환하므로, 내용을 보려면 리스트로 풀어(unpack)서 출력해야 한다. 각 항목은 원래 리스트들에서 같은 위치의 원소들을 모은 tuple이다.

differing_lengths = [*zip(names[:5], primary_types[:3])]

collections 모듈

• napedtuple : 필드명이 있는 tuple 서브클라스
• deque : 빠른 append/pop이 가능한 리스트형 컨테이너
• Counter : 해시 가능한 객체를 세는 dict
• OrderDict : 삽입 순서 유지하는 dict
• defaultdict : 누락 값에 공장 함수 호출하는 dict

집합 이론

symmetric_difference() : 대칭 차집합 .union() : 중복 없이 두 집합 원소들 결합

iterrows(), itertuples()

values: np.array 타입으로 가져올 수 있다.

1. dbt init
2. profiles.yml 파일에서 설정 정의, 업데이트
3. 데이터 모델 정의 & 사용 (dbt run)
   • 데이터 모델 : 데이터 웨어하우스에 저장된 원천 데이털르 변환한 결과
   • dbt run : 원본 SQL 코드르 프로필(배포 대상)에 맞게 변환한 뒤, 변환 과정을 수행
4. 데이터 검증 & 테스트

데이터를 어떤 형태로 저장하고 보여줄 것인가? 에 대한 전략

1. View
   • 데이터 저장 X, 호출될 때마다 SQL 쿼리 실행
   • 저장 공간 차지 X
2. Table
3. Incremental
   • 추가된 데이터만 이어 붙임
4. Ephemeral(임시)
   • DB에 어떤 것도 남기지 않는다.

Materialized View(구체화된 뷰)

• 쿼리 결과 저장 → 속도는 Table처럼 빠름
• 원본 데이터 변하면 백그라운드에서 자동으로 업데이트 시도

완전 무작위 누락(MCAR) : 결측 데이터와 다른 값 사이에 체계적인 관계가 없음

• 데이터 입력 시 입력 오류 무작위 결측(누락)(MAR) : 결측 데이터와 다른 관측된 값 사이에 체계적인 관계가 있음
• 높은 기온에 대한 오존 데이터 결측 비무작위 결측(MNAR) : 결측 데이터와 관측되지 않은 값 사이에 체계적인 관계가 있음
• 높은 기온에 대한 온도 값 결측

process.extract(비교의 기준이 되는 문자열, 비교할 대상들이 담긴 리스트나 딕셔너리, limit=유사도가 높은 순서대로 몇 개의 결과 가져올지 결정)

페어 생성하기, 블로킹 recordlinkage,compare, exact

# import recordlinkage
import recordlinkage

# create indexing object
indexer = recordlinkage.Index()

# generate pairs blocked on state
indexer.block('state')
pairs = indexer.index(column_A, column_B)

# create a compare object
compare_cl = recordlinkage.Compare()

# find exact matches for pairs of column
compare_cl.exact('column', 'column', label='label_name')

# find similar matches for pairs of surname and address_1 using string similarity
compare_cl.string('surname', 'surname', threshold=0.85, label='label_name')

# find matches
potential_matches = compare_cl.compute(pairs, column_A, column_B)

recordlinkage

서로 다른 두 데이터셋에서 같은 대상을 가르키는 행들을 찾아내는 과정

ex) A 데이터셋의 A 식당과 B데이터셋의 B 식당이 같은 식당인가? city, cuisine_type을 먼저 비교해.

Indexing(Block) : 비교할 범위를 제한 Compare(exact) : 완벽히 일치해야 하는 항목 → 같으면 True, 다르면 False Compare(string): 오타가 있어도 비슷한 항목 → Threshold 이상 = 1 Compute: 최종 점수 계산

페어 생성 → 컬럼 간 비교 → 잠재적 매치 찾기 → 중복 탐색 → 두 데이터프레임 붙이기

Python으로 데이터 정제하기 Again → Record Linkage

• Python에서 날짜와 시간 다루기
• Python 정규표현식
• Python에서 결측치 처리

sheet_name=None으로 인수를 전달하면 모든 시트가 dictionary 형태로 반환됨. sheet_name을 인수로 전달하지 않으면 첫 번째 시트만 반환된다. 각 스프레드 시트의 이름을 알고 싶다면 xls.keys()로 키값을 구해야 된다.

웹 개발에서 tag soup는 웹페이지의 html 코드가 구조나 문법 면에서 얽힌 상태를 뜻한다. BeautifulSoup는 이 tag soup를 보기 좋게 만들고 필요한 정보를 추출하는 것이다.

# error
authentication = {'name':'john@doe.com', 'password':'Warp_ExtrapolationsForfeited2'}


response = requests.get('http://localhost:3000/albums', auth=authentication)

if(response.status_code == 200):
    print("Success!")
elif(response.status_code == 401):
    print('Authentication failed')
else:
    print('Another error occurred')

auth 인자는 기본 인증을 위해 사용자 이름과 비밀번호가 들어있는 Tuple을 기대한다. 따라서 dictionary 형태로 값을 전달하면 'dict' object is not callable이라는 에러가 뜬다.

reqeusts.get() 에서 URL 매개변수로 값을 전달하는 인자는 params 다. headers에서 API 키에 대한 값을 전달할 때는

headers = {"Authorization" : 'Bearer 8apDFHaNJMxy8Kt818aa6b4a0ed0514b5d3'}

Bearer는 소유자라는 의미다. → 토큰의 소유자로서 토큰 전달

content-type header : API가 응답을 어떤 형식으로 보냈는지를 나타낼 때 사용

• 내가 보내는 것

accept header : Request 헤더에서만 주로 사용한다

• 받고 싶은 데이터의 형식

GET 요청에서 URL 쿼리 파라미터 → params 인자

json 데이터를 보낼 때(POST 요청) → json 인자

raise_for_status() : enable raising exceptions for returned error statuscodes

except HTTPError as http_err : catch error responses from the API server except ConnectionError as conn_err : catch any connections errors

kwargs.iter() → dict has no attribute 'iteration'

파이썬3에는 iter() 메서드 존재하지 않는다. items(), keys(), values()가 iterator와 유사한 view 객체를 반환한다.

kwargs.items() → can only concatenate str (not tuple) to str

kwargs.items()를 반복문으로 돌리면 각 요소는 (key, value) 형태의 튜플로 반환된다. result는 문자열인데, 문자열에 튜플을 더하려고 해서 위와 같은 에러가 발생한 것이다.

kwargs.values()를 사용해서 value만 뽑아 문자열로 사용하면 해결되는 문제였다.

valueerror : 제공된 값이 허용 가능한 범위에 있지 않을 때

• 타입은 맞았는데, 값이 잘못되서 불가함.
• float('hello') : float는 문자열을 받기도 하는데, hello를 바꿔줄 숫자 값이 없다.

git revert

• 이전 버전을 되살리고 커밋 만든다
• 해당 커밋에서 변경된 모든 파일 복원
• commit 없이 되돌리고 싶다.(staging에 가져오기) : git revert -n HEAD
  • -n : no commit
• 편집기 열리는 것을 피하고 싶다. : --no-edit

만약 단일 파일만 되돌리고 싶다면 → git checkout git checkout HEAD~1 -- report.md

• staging 파일에 올라가 있다.

staging 영역에 있는 파일을 되돌리고 싶다면? → unstaging

• 단일 파일 git restore --staged summary_statistics.csv
• 모든 파일 git restore --staged

flat file : 구조화된 관계가 없는 레코드

• 테이블 데이터를 담은 기본 텍스트 파일
• 필드나 속성으로 이루어진 행, 레코드들의 모음
• 각 필드에는 최대 한 가지 정보만 들어있다.
• 헤더가 있을 수 있다. (데이터의 열의 내용, 해당하는 속성이나 특징이 무엇인지 설명)다

공부를 하다 보니 뭘 내 직무로 삼아야 될 지 감이 안 온다. DA? 개발자? 보안? 하나를 진득하니 했어야 했는데, 최소한 누군가한테 기본기로 하루 정도는 설명할 수 있을 지식은 가지고 있어야 했다. 따라서 오늘부터 해보기로 했다. 뭘? 그냥 다! 알고리즘, 판다스, SQL, DE, DA. IT 소식과 CS도 끊임없이 채워넣겠다. 늦었다고는 생각하지 않지만 열심히 해보자.

모르는 개념, 용어가 나왔을 때 정리하고 그 이외는 넘어갈 예정입니다. 궁금한 점이 생기시면 댓글 남겨주세요. 보는 사람이 있을지는 모르겠지만서도?

일단 시작은 Datacamp로 빠르게 입문하려고 합니다. 그럼 바로 시작할께요.

Data Engineering For Beginners : https://de101.startdataengineering.com/

환경 설정

1. git version >= 2.37.1

2. Docker version >= 20.10.17 && Docker compose v2 version >= v2.10.2

3. https://github.com/josephmachado/data_engineering_for_beginners_code fork 한 후에, 자기 레포지토리에서 클론

git clone clone_address.git
cd data_engineering_for_begineers_code
docker compose up -d --build
sleep 30

MAC을 사용하는 경우

이런 에러를 확인할 수 있다. 위 사진을 보면 amd64로 spark가 설치되었다..

amd64를 arm64로 변경해준다.

docker compose down

docker compose up -d --build

컨테이너 종료하고 다시 build

http://localhost:8888 들어가서 다시 data 만들어보면 ... 성공!!

Data Schema는 아래와 같다.

각 행에서 3번째 NaN 값이 있는 컬럼을 찾아라

import numpy as np
nan = np.nan

data = [[0.04,  nan,  nan, 0.25,  nan, 0.43, 0.71, 0.51,  nan,  nan],
        [ nan,  nan,  nan, 0.04, 0.76,  nan,  nan, 0.67, 0.76, 0.16],
        [ nan,  nan, 0.5 ,  nan, 0.31, 0.4 ,  nan,  nan, 0.24, 0.01],
        [0.49,  nan,  nan, 0.62, 0.73, 0.26, 0.85,  nan,  nan,  nan],
        [ nan,  nan, 0.41,  nan, 0.05,  nan, 0.61,  nan, 0.48, 0.68]]

columns = list('abcdefghij')

df = pd.DataFrame(data, columns=columns)

isnull()은 isna()와 같다. isnull은 null 값에 익숙한 사람들을 위해 만든 alias 함수이다. df.isna().sum() - NaN의 총 개수는 구할 수 있는데 어떻게 3번째일 때를 조건절로 세울까?

cumsum()이라고 누적합을 구해주는 함수가 있다. df.loc[:, df.isna().cumsum(axis=1) == 3].idxmax()를 실행하면 Cannot index with multidimensional key라는 에러가 뜨는 것을 볼 수 있다. loc 함수는 행이나 열 자리에 조건을 넣을 때, 1줄짜리(1차원) 조건만을 수용하는데, 실제로 넣은 것은 각 행에 boolean 값이 들어가있는 df와 같은 크기의 2차원 표이기 때문에 에러가 발생한 것이다.

그러면 어떻게 할까?? df.loc을 그냥 안 쓰면 된다. (df.isna().cumsum(axis=1) == 3).idxmax(axis=1) 여기서 idxmax()??라고 할 수도 있는데, idxmax()는 최댓값이 여러 개가 있으면 가장 먼저 나온 값을 반환한다. 그렇기에 조건절로 max 값을 설정해놓고 가장 맨 첫 번째 값을 출력할 수 있다. 꿀팁이다.

27. A DataFrame has a column of groups 'grps' and column of integer values 'vals'. For each group, find the sum of the three greatest values. You should end up with the answer as follows:

a.b.c 컬럼에서 각각 가장 큰 3개의 값 합 구하시오.

grps
a    409
b    156
c    345

# 내 풀이
df['rank'] = df.groupby(by='grps')['vals'].rank(ascending=False)
df.loc[df['rank']<=3, ['grps','vals']].sort_values(by=['grps']).groupby('grps').sum()

# 해설
df.groupby('grps')['vals'].nlargest(3).sum(level=0)

일단 sort_values를 썼으면 굳이 rank가 없었어도 됐다. sql을 생각해보면 group by절 사용 후에 특정값을 뽑아내려고 하면 실패하는 경험들이 있을 것이다. 이처럼 pandas에서도 groupby('grps')를 실행하는 순간, 그룹화된 보따리 안에 있다고 생각해야 된다. .sort_values() 함수는 DataFrame 상태일 때만 사용할 수 있는 전용 함수이기에 그룹화한 후에 정렬을 하려고 하면 정렬 기능 실행 못한다고 에러가 발생되는 것이다!

df['rank']를 선언하고 sort_values를 사용했던 것은 의도가 아니었는데 앞으로는 지식을 가지고 사용하자. 다시 돌아가서!

그룹화한 후에 정렬 못하면 어떻게 하냐? GroupBy 객체 상태에서 상위 N개 추출 전용 함수가 있다. nlargest라는 함수다.

nlargest(3)을 하고 sum을 하면 그냥 grps가 a,b,c일 때 3번째까지 큰 수를 그냥 다 더한 것 밖에 되지 않는다.

그룹화를 한 것에 주의해야 하는데, 그룹화를 하고 그 안에서 3번째로 큰 수까지 구했다는 건 grps와 vals를 가르는 경계가 있는 것을 생각해볼 수 있다. 이때 밖을 level=0, 안을 level=1이라고 하는데 sum(level=0)이라고 설정하면 레벨 0 기준으로 묶어서 합을 구할 수 있다.

그런데, 에러가 발생할 수도 있어서 요즘은 이렇게 사용하지 않고 df.groupby(by='grps')['vals'].nlargest(3).groupby(level=0).sum() 과 같이 사용한다.

갑자기 공식 문서 보다가 궁금해진 함수들

set_flags() pandas는 컬럼 이름이나 인덱스(행 이름)이 중복되는 것을 허용한다. 다른 사람들이 만든 데이터들을 합칠 때, 다른 의미의 데이터가 이름은 같다면 병합되어 데이터의 가치가 없어질 것이다.

그럴 때, set_flags() 로 중복된 이름표가 들어올 수 없도록 할 수 있다. df_strict = df.set_flags(allows_duplicate_labels = False) -> 원래는 덮어씌우거나 추가되지만, 중복 라벨 허용 불가 옵션을 달면 에러가 뜬다.

row : animal, column : 방문횟수, 평균나이

pivot table

pandas.pivot_tabe(data 
                , index = None # 각 행은 무엇으로 정의할지
                , columns = None # 각 열 정의
                , values = None # 각 Cell 어떤 숫자로 계산할지
                , aggfunc = 'mean', 'sum', 'nunique', 'std' # 계산 방법
                # 추가 옵션
                , fill_value,  margins, drop_na, margins_name, observed
)

pivot_table로 table 확인하는데 원하지 않는 데이터가 많을 때, query 함수로 원하는 데이터만 조회할 수 있다. query('컬럼명 == 원하는 조건') 활용 예시

r1.query('age == [ 2, 3]').pivot_table(index = ['animal'], columns = ['visits'], values = ['age'], aggfunc = [np.mean])

answer

df.pivot_table(
    index='animal', columns = 'visits', values = 'age', aggfunc = 'mean')

22. 중복되지 않는 값 출력하기?

df.drop_duplicates(subset=None, keep='first', inplace=False, ignore_index=False) Return DataFrame with duplicates rows removed. keep : determines which duplicates to keep (first, last, false - drop all) subset : only consider certain columns for identifying duplicates

answer

# 01. drop_duplicates
df.drop_duplicates(subset='A')

# 02. loc 함수, shift 함수
df.loc[df['A'].shift() != df['A']]

02번 풀이가 뭔가 싶을 수 있는데, shift()의 period defalt 값이 1로 shift 함수는 값을 한 칸씩 아래로 미는 함수다. 1칸씩 밑으로 밀었을 때, 서로 다른 값들이 있는 행만 선택해서 출력하면 중복값이 필터링된다. 이는 값들이 정렬되어 있기에 가능한 풀이다.

23. given a dataframe of numeric values, how do you subtract the row mean from each element in the row?

행의 각 요소에 행 평균값을 빼봐라

df = pd.DataFrame(np.random.random(size=(5,3)))

df.sub((df.mean(axis = 1) , axis = 0))

혹시 axis가 헷갈리다면

• axis = 0(default) : 행을 따라 아래로 계산(위에서 아래)
• axis = 1 : 열을 따라 옆으로 계산(왼쪽에서 오른쪽으로)

24. suppose you have dataframe with 10 columns of real numbers. which column of numbers has the smallest sum? return that column's label.

# 합이 가장 작은 컬럼의 값 
df.sum(axis = 0).min(axis = 0)

# 01. loc으로 위에서 찾은 값을 조건으로 설정하여 column label 불러올 수 있다.
df.loc[:, df.sum() == df.sum().min()].columns[0]

# 02. idxmin()
df.sum().idxmin()

25. how do you count how many unique rows a DataFrame has (i.e. ignore all rows that are duplicates)?

unique한 행의 개수?

len(df.drop_duplicates(keep=False))

unique한 행을 찾는 것이기 때문에 중복되는 행은 전부 삭제해야 한다. -> keep=False len()을 DataFrame에 사용하면 df의 행 개수를 반환한다. df.shape[0]을 사용해서 행 개수를 구할 수도 있다.