1. 한 줄 정의

Azure SQL Database의 CPU 사용률이 일정 임계값(예: 평균 80%)을 넘으면 Azure Monitor를 통해 자동으로 이메일 알림이 오도록 Alert Rule을 구성하는 실습을 진행했다.

2. 이걸 왜 하는지 (핵심 개념)

• Signal / Threshold / Action Group 구조가 핵심이다. Signal은 감시할 지표(CPU %), Threshold는 위험 판단 기준(평균 80% 초과), Action Group은 위험할 때 실행할 조치(이메일 알림)를 의미한다.
• 문제가 생긴 후에 로그를 뒤져서 원인을 찾는 것과, 문제가 생기기 전에 미리 알림을 받아 대응하는 것은 다르다. 알림 체계는 후자를 가능하게 해준다.
• Aggregation Type을 Average로 설정하는 이유는, 순간적인 CPU 스파이크(쿼리 하나가 잠깐 튀는 것)와 지속적인 부하(계속 높은 상태가 유지되는 것)를 구분하기 위함이다. 실제로 위험한 건 후자이기 때문에 평균값 기준으로 판단한다.
• Elastic Pool을 쓰는 환경에서는 여러 DB가 CPU/메모리를 공유 자원 풀로 나눠 쓰기 때문에, DB 하나의 부하가 다른 DB의 성능에도 영향을 줄 수 있다. 이런 구조에서는 어느 DB가 원인인지 특정하기 위해 DB별 CPU 알림이 더 중요해진다.

3. 내 프로젝트와 연결

방역로 프로젝트에서 Azure Functions(fn_weather, fn_kahis, fn_disinfection, fn_migratory)로 매일 자동으로 데이터를 수집해 적재하는 파이프라인을 구축했다. 이런 정기 배치 구조를 실제 운영 환경으로 확장한다면, 데이터 양이 늘어나거나 특정 배치 작업이 무거워질 때 DB에 지속적인 부하가 걸릴 수 있다. 이번 실습에서 배운 CPU 알림을 적용하면 이런 상황을 사후에 로그로 확인하는 게 아니라 사전에 감지하고 대응할 수 있다고 생각했다.

4. 예상 꼬리질문 대비

Q. Static threshold 말고 Dynamic threshold는 언제 써요? A. 트래픽 패턴이 일정하지 않고 가변적인 경우, 고정된 임계값(예: 80%)은 false positive를 자주 발생시킬 수 있다. Dynamic threshold는 ML 기반으로 정상 패턴을 학습해 임계값을 자동 조정하기 때문에, 트래픽이 시간대별로 크게 변하는 서비스에 적합하다. 다만 정확도를 위해 14일 이상의 메트릭 히스토리가 필요하다.

Q. Elastic Pool에서는 이 알림이 왜 더 중요해요? A. Pool 밖에서는 각 DB가 자원을 독점하기 때문에 한 DB의 부하가 다른 DB에 영향을 주지 않지만, Pool 안에서는 자원을 공유하기 때문에 한 DB의 부하 급증이 다른 DB들의 성능 저하로 이어질 수 있다. 이런 연쇄 영향을 조기에 감지하고 원인 DB를 특정하기 위해 DB별 알림이 더 중요하다.

Q. 알림을 받으면 그다음엔 뭘 해요? A. 원인 쿼리를 확인해 최적화하거나, 지속적으로 부하가 높다면 더 높은 컴퓨팅 티어로 스케일업을 검토한다. 자동화까지 고려한다면 Action Group에 Webhook이나 Logic App을 연결해 임계값 초과 시 자동 스케일업 같은 대응도 구성할 수 있다.

5. 헷갈렸던 부분 (복습용 메모)

• CPU 80%는 물리적 CPU 성능의 80%가 아니라, 내가 할당받은 컴퓨팅 티어(DTU 또는 vCore) 대비 사용 비율이다.
• 논리 서버(Logical Server)는 실제 컴퓨터가 아니라 로그인/방화벽 등을 관리하는 껍데기이고, CPU 같은 실제 자원은 서버가 아니라 그 안의 개별 DB(또는 Elastic Pool) 단위로 할당된다.

13 Query Optimization

📌 핵심 개념: Catalyst 옵티마이저

Spark는 코드를 실행하기 전에 Catalyst라는 최적화 엔진이 쿼리 계획을 자동으로 다듬어줌.

비유: 내가 요리 순서를 비효율적으로 짜도, 주방장(Catalyst)이 알아서 더 효율적인 순서로 바꿔서 실행하는 것

1. explain() — 쿼리 계획 확인

df.explain(True)

출력 순서:

== Parsed Logical Plan ==      ← 내가 쓴 코드 그대로
== Analyzed Logical Plan ==    ← 컬럼/타입 검증 후
== Optimized Logical Plan ==   ← Catalyst가 최적화한 버전  ← 여기가 핵심
== Physical Plan ==            ← 실제 실행될 계획

실무에서는 Optimized Logical Plan과 Physical Plan을 주로 봄.
"내가 짠 코드가 실제로 어떻게 실행되나?" 확인할 때 사용.

2. 논리적 최적화 (Logical Optimization)

2-1. filter 자동 병합

filter를 여러 번 체이닝해도 Catalyst가 하나로 합쳐서 실행함.

# 내가 쓴 코드 (filter 8번)
limit_events_df = (df
    .filter(col("event_name") != "reviews")
    .filter(col("event_name") != "checkout")
    .filter(col("event_name") != "register")
    ...8개...
)

# Catalyst가 실제로 실행하는 것 (filter 1번으로 합쳐짐)
# → Optimized Plan에서 확인 가능

그래서 이렇게 써도 성능 차이 없음:

# 한 번에 쓴 버전 — 가독성은 더 좋음
better_df = (df
    .filter(
        (col("event_name") != "reviews") &
        (col("event_name") != "checkout") &
        (col("event_name") != "register") &
        ...
    )
)

실무 팁: 성능은 같으니까 가독성 기준으로 선택하면 됨.
조건이 많으면 한 번에 쓰는 게 더 보기 좋음.

2-2. 중복 filter 자동 제거

같은 조건을 실수로 여러 번 써도 Catalyst가 중복을 제거하고 한 번만 실행함.

# 이렇게 실수로 써도
stupid_df = (df
    .filter(col("event_name") != "finalize")
    .filter(col("event_name") != "finalize")
    .filter(col("event_name") != "finalize")
    .filter(col("event_name") != "finalize")
    .filter(col("event_name") != "finalize")
)

# Catalyst가 filter 1번으로 줄여서 실행

비유: "문 잠갔어? 잠갔어? 잠갔어?" 라고 5번 물어봐도 실제로 문 확인은 한 번만 하는 것

3. 조건자 푸시다운 (Predicate Pushdown)

조건자(Predicate) = filter 조건
푸시다운(Pushdown) = 가능한 한 데이터 소스에 가까운 쪽으로 밀어 내리는 것

개념

일반적 실행 순서:
파일에서 전체 데이터 읽기 → 메모리에 올림 → filter 적용

푸시다운 후:
파일을 읽을 때 filter를 같이 적용 → 처음부터 필요한 데이터만 읽음

비유: 도서관에서 책 1000권 다 꺼낸 후 골라내기 vs. 처음부터 원하는 책만 꺼내기

언제 작동하나?

데이터 소스가 푸시다운을 지원할 때 자동으로 작동:

explain()으로 확인하는 법

df.filter(col("event_name") == "purchase").explain(True)

Physical Plan에서 PushedFilters가 보이면 푸시다운 작동 중:

== Physical Plan ==
*(1) Filter (isnotnull(event_name) AND (event_name = purchase))
+- *(1) ColumnarToRow
   +- FileScan parquet [...]
      PushedFilters: [IsNotNull(event_name), EqualTo(event_name, purchase)]  ← 이게 보이면 OK

푸시다운이 작동 안 하는 경우

UDF(사용자 정의 함수)를 filter에 쓰면 푸시다운 불가:

from pyspark.sql.functions import udf

@udf("boolean")
def is_purchase(name):
    return name == "purchase"

# 이렇게 쓰면 푸시다운 안 됨 — Spark가 UDF 내부를 모르기 때문
df.filter(is_purchase(col("event_name")))

# 이렇게 써야 푸시다운 됨
df.filter(col("event_name") == "purchase")

실무 팁: filter 조건에는 가능하면 내장 함수를 쓰고, UDF는 최후의 수단으로.

4. 실무에서 explain() 활용하는 패턴

쿼리가 느릴 때 체크리스트

# 1. 쿼리 계획 확인
df.filter(...).join(...).groupBy(...).explain(True)

# 체크 포인트:
# - Physical Plan에 PushedFilters 있는가? (푸시다운 작동 여부)
# - Filter가 Join보다 앞에 있는가? (작은 데이터로 join해야 빠름)
# - 불필요한 컬럼을 끝까지 들고 가고 있지 않은가?

filter는 최대한 앞에

# 느린 패턴: join 후 filter
df1.join(df2, "user_id").filter(col("event_name") == "purchase")

# 빠른 패턴: filter 후 join (Catalyst가 자동으로 해주기도 하지만 명시적으로 쓰는 게 안전)
df1.filter(col("event_name") == "purchase").join(df2, "user_id")

select로 필요한 컬럼만 미리 추리기

# 컬럼 20개짜리 DataFrame을 끝까지 들고 다니지 말고
df.select("user_id", "event_name", "event_timestamp")  # 필요한 것만 먼저 추리기
  .filter(...)
  .join(...)

5. 빠른 참조

1. 날짜/시간 함수 (9 Datetimes)

1-1. 타임스탬프로 변환 — cast()

비유: cast는 "형변환 캐스팅" — 배우(데이터)를 다른 역할(타입)로 바꾸는 것

원시 데이터의 타임스탬프는 보통 마이크로초(μs) 단위 Long 숫자로 저장됨.
초 단위로 바꾸려면 1e6(= 1,000,000)으로 나눈 뒤 cast.

# 방법 A: 문자열로 타입 지정
timestamp_df = df.withColumn("timestamp", (col("timestamp") / 1e6).cast("timestamp"))

# 방법 B: TimestampType 객체로 지정 (동일한 결과)
from pyspark.sql.types import TimestampType
timestamp_df = df.withColumn("timestamp", (col("timestamp") / 1e6).cast(TimestampType()))

1-2. 날짜 포맷 지정 — date_format()

from pyspark.sql.functions import date_format

df.withColumn("date string", date_format("timestamp", "MMMM dd, yyyy"))
  .withColumn("time string", date_format("timestamp", "HH:mm:ss.SSSSSS"))

주요 날짜 패턴 문자

1-3. 날짜/시간 속성 추출

타임스탬프에서 연/월/일/분/초를 숫자 컬럼으로 뽑아냄

from pyspark.sql.functions import year, month, dayofweek, minute, second

df.withColumn("year",      year(col("timestamp")))
  .withColumn("month",     month(col("timestamp")))
  .withColumn("dayofweek", dayofweek(col("timestamp")))  # 1=일요일, 7=토요일
  .withColumn("minute",    minute(col("timestamp")))
  .withColumn("second",    second(col("timestamp")))

모두 동일한 형태: 함수명(컬럼) → 정수 반환

1-4. 날짜 타입으로 변환 — to_date()

타임스탬프(날짜+시간) → 날짜만 남김 (시간 정보 제거)

from pyspark.sql.functions import to_date

date_df = timestamp_df.withColumn("date", to_date(col("timestamp")))
# 결과: "2023-07-15 14:30:00" → "2023-07-15"

1-5. 날짜 더하기 — date_add()

from pyspark.sql.functions import date_add

df.withColumn("plus_two_days", date_add(col("timestamp"), 2))

1-6. 기타 유용한 날짜 함수 (참고용)

2. 복합 타입 함수 (10 Complex Types)

2-1. 문자열 함수

from pyspark.sql.functions import split, lower, regexp_replace  # 필요한 것만 import

split() 예시

# email에서 @ 기준으로 분리
df.select(split(df.email, '@', 0).alias('email_handle'))
# 결과: ["user", "gmail.com"] 형태의 배열 컬럼

2-2. 컬렉션 함수 (배열 조작)

비유: 배열이 담긴 컬럼을 다루는 함수들. 마치 리스트를 다루는 파이썬 메서드처럼.

explode() 핵심

# items 컬럼이 배열이면, 각 원소가 별도 행으로 "터져 나옴"
sales_exploded_df = df.withColumn("items", explode("items"))

비유: 달걀 한 판(배열)을 하나씩 꺼내서 각각 새 줄에 올려놓는 것

실전 패턴 — explode → select → split

details_df = (df
              .withColumn("items", explode("items"))          # 배열 → 행 분리
              .select("email", "items.item_name")             # 필요 컬럼만
              .withColumn("details", split(col("item_name"), " "))  # 문자열 → 배열
             )

2-3. 집계 함수 (배열 생성)

# 이메일별로 주문한 매트리스 크기 종류 수집
size_df = mattress_df.groupBy("email").agg(collect_set("size").alias("size options"))

2-4. DataFrame 합치기 — union vs unionByName

비유:

• union = 위치 기준 합치기 (컬럼 순서가 같아야 안전)
• unionByName = 이름 기준 합치기 (컬럼 순서 달라도 OK)

# union: 컬럼 순서가 완전히 같을 때
df1.union(df2)

# unionByName: 컬럼명이 같으면 순서 달라도 됨 (더 안전)
df1.unionByName(df2)

⚠️ 둘 다 중복 행을 제거하지 않음 (SQL의 UNION ALL과 같음)

3. 추가 함수 (11 Additional Functions)

3-1. col() — 컬럼 객체 참조

from pyspark.sql.functions import col

# 문자열 컬럼명 대신 col 객체를 쓰면 메서드 체이닝 가능
df.filter(col("email").endswith("gmail.com"))

왜 쓰나?: "email" 문자열은 그냥 이름이지만, col("email")은 컬럼 객체라서 .endswith(), .isNull() 같은 메서드를 붙일 수 있음

3-2. lit() — 고정값으로 컬럼 만들기

from pyspark.sql.functions import lit

# 모든 행에 True 값을 가진 "gmail user" 컬럼 추가
df.select("email", lit(True).alias("gmail user"))

비유: 모든 행에 도장 찍기. "이 데이터는 전부 gmail 유저다" 같은 플래그 컬럼 만들 때 유용

3-3. Null 처리 — df.na

접근 방법: df.na.함수명() 형태로 사용

# null 있는 행 제거
df.na.drop()

# null을 "NO COUPON"으로 채우기
df.na.fill("NO COUPON")

# 행 수 비교로 null 존재 확인
print(df.count())          # 전체
print(df.na.drop().count()) # null 제거 후 → 숫자 다르면 null 있음

⚠️ 주의: 배열 컬럼(items 같은) 안에 null이 있으면 na.drop()으로 안 잡힘.
explode() 먼저 해서 펼친 뒤 확인해야 함.

3-4. DataFrame 조인 — join()

df1.join(other=df2, on=조인조건, how=조인방식)

조인 방식 (how)

예시 패턴

# 같은 이름의 컬럼으로 inner join
joined_df = df1.join(other=df2, on="email", how="inner")

# 여러 컬럼으로 join
df1.join(df2, ["name", "age"])

# 컬럼명이 다를 때
df1.join(df2, df1["customer_name"] == df2["account_name"], "left_outer")

4. 전체 함수 빠른 참조표

날짜/시간

문자열

컬렉션/배열

기타

비유: Reader는 "데이터를 가져오는 택배 수령", Writer는 "데이터를 보내는 택배 발송"

1. DataFrameReader — 데이터 읽기

기본 패턴

df = (spark
      .read
      .option("옵션이름", 값)   # 옵션 추가 (체이닝 가능)
      .schema(스키마)           # 스키마 직접 지정 (선택)
      .csv("파일경로")          # 파일 형식 메서드로 마무리
     )

1-1. CSV 읽기

방법 A: .option() 체이닝

users_df = (spark
            .read
            .option("sep", "\t")        # 구분자 (기본값은 쉼표 ",")
            .option("header", True)     # 첫 줄을 컬럼명으로 사용
            .option("inferSchema", True) # 타입 자동 추론
            .csv("파일경로")
           )

방법 B: csv() 메서드에 직접 인자 전달

users_df = (spark
            .read
            .csv("파일경로", sep="\t", header=True, inferSchema=True)
           )

두 방법은 동일한 결과. 취향에 따라 선택

주요 option 인자 정리

⚠️ inferSchema=True는 전체 데이터를 한 번 스캔하므로 느림. 대용량이면 스키마 직접 지정 권장

1-2. JSON 읽기

events_df = (spark
             .read
             .option("inferSchema", True)  # 타입 자동 추론
             .json("파일경로")
            )

1-3. 스키마 직접 정의 — StructType 방식

비유: 테이블의 설계도를 미리 그려서 "이 건물은 이렇게 지어야 해"라고 Spark에게 알려주는 것

Import 먼저

from pyspark.sql.types import (
    LongType, StringType, DoubleType, IntegerType,
    ArrayType, StructType, StructField
)

기본 구조

스키마 = StructType([
    StructField("컬럼명", 데이터타입(), nullable여부),
    StructField("컬럼명", 데이터타입(), nullable여부),
    ...
])

주요 데이터 타입

사용 예시 (단순 CSV용)

schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("user_first_touch_timestamp", LongType(), True),
    StructField("email", StringType(), True)
])

df = spark.read.option("sep", "\t").option("header", True).schema(schema).csv("경로")

중첩 구조 예시 (JSON용 — StructType 안에 StructType)

schema = StructType([
    StructField("device", StringType(), True),
    StructField("geo", StructType([           # ← 중첩 struct
        StructField("city", StringType(), True),
        StructField("state", StringType(), True)
    ]), True),
    StructField("items", ArrayType(           # ← 배열 안에 struct
        StructType([
            StructField("item_id", StringType(), True),
            StructField("quantity", LongType(), True)
        ])
    ), True)
])

1-4. 스키마 직접 정의 — DDL 문자열 방식

더 짧고 간단한 대안. SQL의 CREATE TABLE 문법과 유사

ddl_schema = "user_id string, user_first_touch_timestamp long, email string"

df = spark.read.option("sep", "\t").option("header", True).schema(ddl_schema).csv("경로")

1-5. toDDL 트릭 — 스키마 자동 생성 (개발용)

목적: 복잡한 JSON의 스키마를 처음부터 손으로 쓰기 어려울 때, Spark가 자동 추론한 스키마를 DDL 문자열로 뽑아주는 편법

⚠️ 개발/탐색 단계에서만 사용. 운영 환경에서는 금지 (inferSchema는 전체 데이터를 읽어서 느림)

# Step 1: Python에서 경로를 Spark config로 공유
spark.conf.set("com.scope.events_path", "파일경로")

// Step 2: Scala 셀에서 실행 (%scala 마법 명령어 필요)
val path = spark.conf.get("com.scope.events_path")
val schema = spark.read.option("inferSchema", true).json(path).schema.toDDL
println(schema)

# Step 3: 출력된 DDL 문자열을 Python 변수에 붙여넣기
events_schema = "`device` STRING, `event_name` STRING, ..."  # 복사한 값

df = spark.read.schema(events_schema).json("경로")

2. DataFrameWriter — 데이터 쓰기

기본 패턴

(df
 .write
 .option("옵션이름", 값)   # 선택
 .mode("쓰기 모드")        # 중요!
 .저장메서드("경로 또는 테이블명")
)

2-1. 파일로 저장 — Parquet

(users_df
 .write
 .option("compression", "snappy")  # 압축 방식
 .mode("overwrite")                # 기존 파일 덮어쓰기
 .parquet("저장경로")
)

# 또는 메서드에 직접 인자 전달
(users_df
 .write
 .parquet("저장경로", compression="snappy", mode="overwrite")
)

mode() 옵션 정리

2-2. 테이블로 저장 — saveAsTable

events_df.write.mode("overwrite").saveAsTable("테이블명")

createOrReplaceTempView vs saveAsTable 차이

2-3. Delta 테이블로 저장

(events_df
 .write
 .format("delta")     # 저장 형식을 Delta로 지정
 .mode("overwrite")
 .save("저장경로")
)

3. Delta Lake — 왜 써야 하나?

비유: 일반 Parquet은 그냥 파일 보관함, Delta Lake는 "버전 관리되는 구글 드라이브"

Databricks에서는 거의 항상 Delta Lake 사용 권장

4. 결과 확인 방법

# 스키마 확인
df.printSchema()

# 데이터 미리보기
display(df)         # Databricks 전용 (테이블 형태)
df.show(5)          # 일반 Spark (텍스트 형태)

# 저장된 파일 목록 확인
display(dbutils.fs.ls("저장경로"))

5. 전체 흐름 요약

CSV/JSON 파일
    ↓  spark.read.option().schema().csv() / .json()
DataFrame (메모리 내 처리)
    ↓  df.write.mode().parquet() / .saveAsTable() / .format("delta").save()
Parquet 파일 / 테이블 / Delta 테이블

🗓️ 학습일: 2026.06
📚 출처: Databricks 수업 자료 (5_Aggregation.ipynb)

🎯 오늘 배운 것 한 줄 요약

"데이터를 그룹으로 묶고(groupBy), 그 그룹에 집계를 적용한다(agg / count / sum / avg …)"

1. 집계(Aggregation)란?

집계는 여러 행(row)을 하나의 요약 값으로 압축하는 작업이다.

비유 🧾
엑셀에서 피벗 테이블을 만들 때를 떠올려보자.
"지역별 매출 합계"를 구할 때, 수백 개의 행을 지역으로 묶고, 매출을 더하지 않나?
PySpark의 groupBy + 집계 함수가 바로 그 역할이다.

2. 핵심 흐름

DataFrame
  └─ .groupBy("열이름")       ← 그룹화 (GroupedData 객체 생성)
       └─ .count()            ← 집계 적용
       └─ .avg("다른열")
       └─ .sum("다른열")
       └─ .agg(함수1, 함수2)  ← 여러 집계를 한 번에

3. groupBy — 그룹화하기

groupBy는 DataFrame을 특정 열 기준으로 묶어 GroupedData 객체를 만든다.
이 시점에서는 아직 계산이 일어나지 않는다. 집계 메서드를 붙여야 실제 결과가 나온다.

# 단일 열 기준
df.groupBy("event_name")

# 여러 열 기준 (조합)
df.groupBy("geo.state", "geo.city")

💡 중첩 열(nested column) 접근
"geo.state" 처럼 점(.)으로 구분해서 중첩된 구조의 필드에 접근할 수 있다.

4. GroupedData 집계 메서드

코드 예시

# 이벤트 이름별 행 수
event_counts_df = df.groupBy("event_name").count()
display(event_counts_df)

# 주(state)별 평균 구매 수익
avg_state_purchases_df = df.groupBy("geo.state").avg("ecommerce.purchase_revenue_in_usd")
display(avg_state_purchases_df)

# 주 + 도시 조합별 총 수량 & 수익 합계
city_purchase_quantities_df = df.groupBy("geo.state", "geo.city").sum(
    "ecommerce.total_item_quantity",
    "ecommerce.purchase_revenue_in_usd"
)
display(city_purchase_quantities_df)

5. 내장 함수(Built-in Functions)와 agg

pyspark.sql.functions 모듈에는 더 다양한 집계 함수가 있다.
이 함수들은 agg()와 함께 사용하며, 결과 열에 별칭(alias) 도 붙일 수 있다.

주요 집계 내장 함수

❓ approx_count_distinct vs count(distinct …)
정확한 고유값 수를 세려면 전체 데이터를 다 봐야 해서 느리다.
approx_count_distinct는 HyperLogLog 알고리즘으로 빠르게 근사치를 구한다.
대용량 데이터에서는 이 함수가 훨씬 실용적이다.

코드 예시

from pyspark.sql.functions import sum

# 단일 집계 + alias
state_purchases_df = df.groupBy("geo.state").agg(
    sum("ecommerce.total_item_quantity").alias("total_purchases")
)
display(state_purchases_df)

from pyspark.sql.functions import avg, approx_count_distinct

# 여러 집계를 한 번에
state_aggregates_df = (df
    .groupBy("geo.state")
    .agg(
        avg("ecommerce.total_item_quantity").alias("avg_quantity"),
        approx_count_distinct("user_id").alias("distinct_users")
    )
)
display(state_aggregates_df)

6. 수학 내장 함수 (보너스)

집계 외에도 열(column) 단위 수학 연산 함수도 있다.

from pyspark.sql.functions import cos, sqrt

display(
    spark.range(10)          # 0~9 범위의 id 열을 가진 DataFrame 생성
         .withColumn("sqrt", sqrt("id"))
         .withColumn("cos", cos("id"))
)

7. 헷갈리기 쉬운 포인트 정리

Q. groupBy 메서드의 집계(avg, sum)와 내장 함수의 집계(avg, sum)의 차이는?

A.
둘 다 같은 결과를 내지만, 내장 함수 방식(agg + 내장 함수)이 더 강력하다.

Q. alias는 왜 쓰는가?

집계를 하면 열 이름이 avg(ecommerce.total_item_quantity) 처럼 길고 못생기게 된다.
alias("avg_quantity") 로 읽기 좋은 이름을 붙여주는 것이다.

8. 오늘의 Q&A

(질문을 하면 이 섹션에 추가됩니다)

9. 전체 구조 한눈에 보기

[데이터 집계 전략]

① 단순 집계
   df.groupBy("A").count()
   df.groupBy("A").avg("B")
   df.groupBy("A", "B").sum("C", "D")

② agg + 내장 함수 (추천 ⭐)
   from pyspark.sql.functions import avg, approx_count_distinct
   df.groupBy("A").agg(
       avg("B").alias("평균B"),
       approx_count_distinct("C").alias("고유C수")
   )

③ 열 단위 수학 함수
   from pyspark.sql.functions import sqrt, cos
   df.withColumn("sqrt_id", sqrt("id"))

✏️ 초안 상태입니다. 질문 후 Q&A 섹션과 설명이 보강될 예정입니다.

컬럼 선택, 추가, 수정, 행 필터링, 정렬까지 DataFrame을 다루는 핵심 문법 정리

읽기 전에 — 컬럼을 참조하는 3가지 방법

DataFrame에서 컬럼을 가리킬 때 세 가지 방법이 있다. 결과는 모두 동일하다.

from pyspark.sql.functions import col

events_df.device        # 점(.) 표기법
events_df["device"]     # 딕셔너리 표기법
col("device")           # col() 함수 (가장 많이 씀)

언제 col()을 써야 하냐면:

# 단순 컬럼 이름만 쓸 땐 문자열도 됨
events_df.select("user_id", "device")

# 연산, 메서드 체이닝, 중첩 구조 접근할 땐 col() 필요
col("ecommerce.purchase_revenue_in_usd") * 100   # 연산
col("event_timestamp").desc()                     # 메서드
col("geo.city").alias("city")                     # 중첩 구조

1. 컬럼 연산자 및 메서드

col()로 가져온 컬럼에 다양한 연산과 메서드를 붙일 수 있다.

# 수학 연산
col("price") * 100
col("quantity") + col("extra_quantity")

# 비교 연산 (Python에서 == 는 ===와 다름에 주의)
col("device") == "Android"
col("price") >= 200

# 메서드
col("event_timestamp").desc()                          # 내림차순 정렬용
col("price").cast("int")                               # 타입 변환
col("email").isNotNull()                               # NULL 아닌 것
col("email").isNull()                                  # NULL인 것
col("device").isin("iOS", "Android")                   # 특정 값 포함 여부
col("ecommerce.purchase_revenue_in_usd") * 100         # 중첩 구조 접근

실제 사용 예시:

rev_df = (events_df
  .filter(col("ecommerce.purchase_revenue_in_usd").isNotNull())
  .withColumn("purchase_revenue",
      (col("ecommerce.purchase_revenue_in_usd") * 100).cast("int"))
  .withColumn("avg_purchase_revenue",
      col("ecommerce.purchase_revenue_in_usd") / col("ecommerce.total_item_quantity"))
  .sort(col("avg_purchase_revenue").desc())
)
display(rev_df)

2. 컬럼 선택 (열 부분 집합)

select() — 원하는 컬럼만 고르기

# 문자열로 컬럼 이름만 지정
devices_df = events_df.select("user_id", "device")

# col()로 중첩 구조 접근 + alias로 이름 변경
from pyspark.sql.functions import col

locations_df = events_df.select(
    "user_id",
    col("geo.city").alias("city"),    # geo 구조체 안의 city를 꺼내서 "city"로 이름 변경
    col("geo.state").alias("state")
)

selectExpr() — SQL 표현식으로 컬럼 선택

# SQL 문법을 문자열로 그대로 쓸 수 있음
apple_df = events_df.selectExpr(
    "user_id",
    "device in ('macOS', 'iOS') as apple_user"  # SQL IN 절 → true/false 컬럼 생성
)
display(apple_df)

select()와 차이:

select()      → col() 객체나 컬럼 이름 문자열 사용
selectExpr()  → SQL 표현식 문자열 그대로 사용 (더 직관적인 경우 있음)

drop() — 특정 컬럼 제거

# 문자열로 여러 컬럼 한번에 제거
anonymous_df = events_df.drop("user_id", "geo", "device")

# col()로도 가능
no_sales_df = events_df.drop(col("ecommerce"))

3. 컬럼 추가 / 수정

withColumn() — 새 컬럼 추가 or 기존 컬럼 덮어쓰기

# 새 컬럼 추가 (mobile 컬럼이 없으면 추가)
mobile_df = events_df.withColumn(
    "mobile",
    col("device").isin("iOS", "Android")   # iOS나 Android면 true, 아니면 false
)

# 기존 컬럼 덮어쓰기 (같은 이름으로 지정하면 교체됨)
purchase_quantity_df = events_df.withColumn(
    "purchase_quantity",
    col("ecommerce.total_item_quantity").cast("int")  # double → int로 타입 변환
)
purchase_quantity_df.printSchema()

withColumn 규칙:

같은 이름 지정  →  기존 컬럼 덮어씀
다른 이름 지정  →  새 컬럼 추가됨

withColumnRenamed() — 컬럼 이름만 바꾸기

# "geo" 컬럼 이름을 "location"으로 변경
location_df = events_df.withColumnRenamed("geo", "location")

4. 행 필터링 (행 부분 집합)

filter() / where() — 조건에 맞는 행만 남기기

filter()와 where()는 완전히 동일하다.

# 문자열 SQL 표현식으로 필터링
purchases_df = events_df.filter("ecommerce.total_item_quantity > 0")

# col()로 필터링
revenue_df = events_df.filter(col("ecommerce.purchase_revenue_in_usd").isNotNull())

# 여러 조건 동시에 (AND는 &, OR는 |, 각 조건은 괄호로 묶어야 함)
android_df = events_df.filter(
    (col("traffic_source") != "direct") & (col("device") == "Android")
)

주의: Python에서 AND/OR 쓸 때

# 틀린 방법 (and, or 쓰면 에러)
.filter(col("a") == 1 and col("b") == 2)   # ❌

# 맞는 방법 (&, | 사용 + 각 조건 괄호 필수)
.filter((col("a") == 1) & (col("b") == 2))  # ✅ AND
.filter((col("a") == 1) | (col("b") == 2))  # ✅ OR

dropDuplicates() / distinct() — 중복 행 제거

# 모든 컬럼 기준으로 완전히 중복된 행 제거
display(events_df.distinct())

# 특정 컬럼 기준으로 중복 제거 (dropDuplicates만 가능)
distinct_users_df = events_df.dropDuplicates(["user_id"])
#  → user_id가 같은 행이 여러 개면 첫 번째만 남김

distinct()와 dropDuplicates() 차이:

distinct()                 → 모든 컬럼 기준, 완전히 같은 행만 제거
dropDuplicates(["컬럼"])   → 지정한 컬럼 기준으로 중복 제거 (다른 컬럼 값은 달라도 됨)

limit() — 상위 n개 행만 남기기

limit_df = events_df.limit(100)

5. 행 정렬

sort() / orderBy() — 정렬 (둘은 완전히 동일)

# 오름차순 (기본값)
increase_df = events_df.sort("event_timestamp")

# 내림차순 (.desc() 붙이기)
decrease_df = events_df.sort(col("event_timestamp").desc())

# 여러 컬럼 기준 정렬 (리스트로)
multi_df = events_df.orderBy(["user_first_touch_timestamp", "event_timestamp"])

# 여러 컬럼, 각각 다른 방향으로
mixed_df = events_df.sort(
    col("user_first_touch_timestamp").desc(),
    col("event_timestamp")             # 이건 오름차순
)

전체 메서드 한눈에 보기

컬럼 관련

행 관련

SQL과 Python DataFrame API, 두 가지 방법으로 같은 쿼리를 표현하는 방법 정리

읽기 전에 — 핵심 개념 정리

Q. 내용은 PySpark인데 왜 제목이 Spark SQL이야?

Spark SQL은 "SQL만 쓰는 것"이 아니라 구조화된 데이터를 처리하는 Spark 모듈 전체의 이름이다.

Spark SQL 모듈
├── SQL 방식          (%sql 셀에서 SELECT * FROM ...)
└── DataFrame API    (Python으로 spark.table(...).select(...))

둘 다 같은 Spark SQL 엔진을 쓰고, 결과도 동일하다. PySpark는 Python으로 Spark를 쓰는 것이고, DataFrame API가 PySpark의 핵심이다.

Q. SQL에서는 Spark를 쓰는지 어떻게 분간해?

Databricks에서 %sql 셀에 쓰는 SQL은 전부 Spark SQL이다.

Databricks 노트북
  %sql 셀         → 무조건 Spark SQL
  %python 셀      → spark. 붙으면 Spark 사용 중

Databricks 밖
  MySQL에서 SELECT ...      → 일반 SQL (Spark 아님)
  spark.sql("SELECT ...")   → Spark SQL (명시적으로 표시됨)

Databricks 안에서는 "항상 Spark SQL이다"라고 이해하면 된다.

1. 두 가지 방법으로 같은 쿼리 쓰기

Spark SQL은 SQL과 DataFrame API, 두 가지 방법으로 동일한 쿼리를 표현할 수 있다.

방법 1: SQL

%sql
SELECT name, price
FROM products
WHERE price < 200
ORDER BY price

방법 2: DataFrame API (Python)

display(spark
        .table("products")
        .select("name", "price")
        .where("price < 200")
        .orderBy("price")
       )

둘 다 완전히 동일한 결과를 반환한다. 내부적으로 같은 쿼리 계획으로 변환되어 실행된다.

2. SparkSession

SparkSession은 DataFrame API를 사용하는 Spark의 모든 기능에 대한 입구다.

Databricks 노트북에서는 자동으로 생성되어 spark 변수에 저장된다.

spark  # 이미 만들어져 있음

주요 메서드:

# SQL 실행 결과를 DataFrame으로 받기
result_df = spark.sql("""
SELECT name, price
FROM products
WHERE price < 200
ORDER BY price
""")
display(result_df)

3. DataFrame

Q. DataFrame은 왜 만들어? 테이블이랑 같은 거야?

비슷하지만 다르다.

테이블                          DataFrame
──────────────────              ──────────────────
메타스토어에 등록됨               Python 변수에 저장됨
세션 끝나도 유지됨                세션 끝나면 사라짐
SQL에서 이름으로 조회 가능         Python 코드에서만 사용

DataFrame을 만드는 이유는 Python 코드 안에서 데이터를 변수처럼 다루기 위해서다.

DataFrame 생성 방법

① spark.table() — 테이블을 DataFrame으로 가져오기

products_df = spark.table("products")

② spark.sql() — SQL 실행 결과를 DataFrame으로 받기

result_df = spark.sql("""
  SELECT name, price
  FROM products
  WHERE price < 200
  ORDER BY price
""")
display(result_df)

③ spark.read — 파일에서 DataFrame 읽기

# CSV 파일에서 읽기
df = spark.read.csv("/path/to/file.csv", header=True)

# Parquet 파일에서 읽기
df = spark.read.parquet("/path/to/file.parquet")

# JSON 파일에서 읽기
df = spark.read.json("/path/to/file.json")

④ spark.createDataFrame() — 테스트용 DataFrame 직접 만들기

# 튜플 리스트로 직접 생성
data = [("Alice", 25), ("Bob", 30), ("Carol", 22)]
df = spark.createDataFrame(data, ["name", "age"])
display(df)

⑤ spark.range() — 숫자 범위로 DataFrame 만들기

# 0부터 9까지 숫자 DataFrame
df = spark.range(10)
display(df)

DataFrame 변수처럼 다루기

# 테이블에서 가져와서 변환 후 변수에 저장
budget_df = (spark
             .table("products")
             .select("name", "price")
             .where("price < 200")
             .orderBy("price")
            )

display(budget_df)  # 결과 출력

스키마 확인

# 스키마 속성으로 보기
budget_df.schema

# printSchema()로 보기 (더 읽기 쉬움)
budget_df.printSchema()

# 출력 예시:
# root
#  |-- name: string (nullable = true)
#  |-- price: double (nullable = true)

4. 변환(Transformation) vs 액션(Action)

DataFrame 메서드는 두 종류로 나뉜다.

변환 — 실행되지 않고 계획만 세움 (지연 평가)

# 이 코드는 실행해도 계산이 안 됨!
(products_df
  .select("name", "price")
  .where("price < 200")
  .orderBy("price"))
# → "이렇게 할 거야" 라는 계획만 만들어둠

액션 — 실제 계산을 실행시키는 메서드

# .show()가 붙어야 비로소 실행됨
(products_df
  .select("name", "price")
  .where("price < 200")
  .orderBy("price")
  .show())           # ← 여기서 실제 계산 시작

왜 지연 평가를 쓰냐면:

변환 10개를 연결해도 Spark는 실행 안 함
        ↓
액션 호출 시 변환 10개를 한꺼번에 최적화해서 실행
        ↓
불필요한 중간 계산 없이 한 번에 처리 → 빠름

5. DataFrame ↔ SQL 변환

DataFrame을 SQL에서 쓰고 싶으면 임시 뷰로 등록하면 된다.

# DataFrame → 임시 뷰 등록
budget_df.createOrReplaceTempView("budget")

-- 이제 SQL에서 budget 테이블처럼 쓸 수 있음
SELECT * FROM budget

# Python에서도 SQL처럼 쓰고 싶으면
display(spark.sql("SELECT * FROM budget"))

전체 흐름 정리

같은 쿼리를 두 가지로 표현 가능
  SQL (%sql 셀)                    DataFrame API (Python)
  SELECT * FROM products    ↔     spark.table("products")
  WHERE price < 200         ↔     .where("price < 200")
  ORDER BY price            ↔     .orderBy("price")

DataFrame 쓸 때 흐름
  spark.table() or spark.sql()     ← 생성
    ↓
  .select() .where() .orderBy()    ← 변환 (지연 평가, 실행 안 됨)
    ↓
  .show() .count() .collect()      ← 액션 (여기서 실제 실행)

DataFrame ↔ SQL 오가기
  df.createOrReplaceTempView("이름")  → SQL에서 테이블처럼 사용 가능
  spark.sql("SELECT * FROM 이름")     → SQL 결과를 DataFrame으로

Databricks 데이터 정리, 고급 SQL 변환, SQL UDF

지저분한 데이터를 정리하고, 복잡한 구조를 다루고, 재사용 가능한 함수를 만드는 방법 정리

1. 데이터 정리

NULL 값 파악하기

count(컬럼)은 NULL을 건너뜀. count(*)은 NULL이 있는 행도 포함.

-- 각 컬럼에 NULL이 몇 개인지 확인
SELECT
  count_if(user_id IS NULL) AS missing_user_ids,
  count_if(email IS NULL) AS missing_emails,
  count_if(updated IS NULL) AS missing_updates
FROM users_dirty

-- 총 행 수 vs NULL 제외 행 수 비교
SELECT count(*), count(user_id) FROM users_dirty
--      ↑ NULL 포함    ↑ NULL 제외

중복 파악하기

-- 전체 행 수
SELECT count(*) FROM users_dirty

-- 고유 행 수
SELECT count(DISTINCT(*)) FROM users_dirty

-- 특정 컬럼 기준 고유 값 수
SELECT count(DISTINCT user_id) FROM users_dirty

주의: DISTINCT는 NULL이 있는 행을 제외하지 않는다. 모든 값이 NULL인 행도 "하나의 고유한 행"으로 처리된다.

특정 컬럼 기준으로 중복 제거

-- user_id + user_first_touch_timestamp 기준으로 중복 제거
-- email은 여러 행 중 NULL이 아닌 값을 가져오기 위해 max() 사용
CREATE OR REPLACE TEMP VIEW deduped_users AS
SELECT user_id, user_first_touch_timestamp,
       max(email) AS email,
       max(updated) AS updated
FROM users_dirty
WHERE user_id IS NOT NULL          -- NULL 행 제거
GROUP BY user_id, user_first_touch_timestamp;

왜 max(email)을 쓰냐면: 같은 user_id가 여러 행이면 GROUP BY 후 email이 여러 개 존재. 그 중 NULL이 아닌 값을 가져오기 위해 max() 사용.

데이터 검증

-- user_id가 중복 없는지 확인 (true면 중복 없음)
SELECT max(row_count) <= 1 AS no_duplicate_ids FROM (
  SELECT user_id, count(*) AS row_count
  FROM deduped_users
  GROUP BY user_id
)

날짜 형식 변환 및 정규 표현식

SELECT *,
  date_format(first_touch, "MMM d, yyyy") AS first_touch_date,   -- 사람이 읽기 좋은 날짜
  date_format(first_touch, "HH:mm:ss") AS first_touch_time,      -- 시간
  regexp_extract(email, "(?<=@).+", 0) AS email_domain           -- 이메일에서 도메인 추출
FROM (
  SELECT *,
    CAST(user_first_touch_timestamp / 1e6 AS timestamp) AS first_touch  -- Unix 타임스탬프 변환
  FROM deduped_users
)

2. 고급 SQL 변환

JSON 데이터 다루기

Kafka 같은 시스템에서 오는 데이터는 JSON이 바이너리로 인코딩된 경우가 많다.

-- 바이너리를 문자열로 변환해서 읽기
CREATE OR REPLACE TEMP VIEW events_strings AS
  SELECT string(key), string(value)
  FROM events_raw;

JSON 안의 중첩 데이터 접근하기:

-- : 구문으로 JSON 필드 접근
SELECT value:device, value:geo:city
FROM events_strings
--    ↑ 최상위 필드  ↑ 중첩 필드 (geo 안의 city)

JSON 문자열을 구조체로 파싱하기:

-- schema_of_json으로 스키마 자동 추론 → from_json으로 파싱
CREATE OR REPLACE TEMP VIEW parsed_events AS
  SELECT from_json(value, schema_of_json('{...샘플JSON...}')) AS json
  FROM events_strings;

-- 구조체를 개별 컬럼으로 평면화
CREATE OR REPLACE TEMP VIEW new_events_final AS
  SELECT json.*       -- ← * 로 펼치기
  FROM parsed_events;

중첩 데이터 구조 접근하기

구조체(struct) 타입은 . 으로 접근:

-- ecommerce 구조체 안의 purchase_revenue_in_usd 접근
SELECT ecommerce.purchase_revenue_in_usd
FROM events
WHERE ecommerce.purchase_revenue_in_usd IS NOT NULL

배열 다루기

-- 배열 크기 필터링 (items가 2개 초과인 행만)
SELECT user_id, items
FROM events
WHERE size(items) > 2

explode — 배열을 행으로 펼치기:

-- items 배열의 각 요소를 별도 행으로 분리
SELECT user_id, explode(items) AS item
FROM events

Before: user_1 | [item_A, item_B, item_C]
After:  user_1 | item_A
        user_1 | item_B
        user_1 | item_C

collect_set, flatten, array_distinct — 배열 수집 및 정리:

Q. 이 세 함수가 뭔지 코드만 보고는 이해가 안 됐다.

단계별로 뭘 하는지 데이터로 보면 이해가 된다.

예시 데이터:

user_id | event_name | items
user_1  | click      | [A, B]
user_1  | purchase   | [B, C]
user_1  | click      | [A, D]

① collect_set(event_name)
   → 그룹 안의 값들을 배열로 모으기 (중복 제거)
   → user_1: ["click", "purchase"]  ← 중복 "click" 하나로 합쳐짐

② collect_set(items.item_id)
   → 배열 안의 값을 모으면 배열들의 배열이 됨
   → user_1: [ [A,B], [B,C], [A,D] ]

③ flatten([ [A,B], [B,C], [A,D] ])
   → 배열들의 배열을 하나의 배열로 펼치기
   → [A, B, B, C, A, D]  ← 하나로 합쳐졌지만 중복 있음

④ array_distinct([A, B, B, C, A, D])
   → 배열에서 중복 제거
   → [A, B, C, D]

세 개를 합치면:

SELECT user_id,
  collect_set(event_name) AS event_history,
  array_distinct(flatten(collect_set(items.item_id))) AS cart_history
  -- → 이 유저가 장바구니에 담았던 모든 상품 ID (중복 없이)
FROM events
GROUP BY user_id

JOIN — 조건에 맞는 행끼리 컬럼 합치기

Q. JOIN은 두 테이블의 컬럼을 합치는 거야?

거의 맞다. 정확히는 "조건에 맞는 행끼리 연결해서 컬럼을 합치는 것".

테이블 A (sales)       테이블 B (item_lookup)
order_id | item_id     item_id  | item_name
1        | M_STAN_K    M_STAN_K | Standard King Mattress
2        | P_FOAM_S    P_FOAM_S | Small Foam Pillow

JOIN ON a.item_id = b.item_id  ← item_id가 같은 행끼리 연결

결과:
order_id | item_id  | item_name
1        | M_STAN_K | Standard King Mattress
2        | P_FOAM_S | Small Foam Pillow

Q. ON에 사용한 컬럼명이 다르면 합쳐지지 않고 둘 다 컬럼으로 적히는 거 맞지?

맞다.

컬럼명 같음 (item_id = item_id)  →  하나로 합쳐져서 컬럼 1개
컬럼명 다름 (order_item_id = product_id)  →  둘 다 컬럼으로 남아서 컬럼 2개

이름이 달라도 값은 같기 때문에, 컬럼명이 다를 때 SELECT * 하면 거의 똑같은 값의 컬럼 두 개가 나란히 생겨서 지저분해 보인다.

-- explode 후 조회 테이블과 JOIN
CREATE OR REPLACE VIEW sales_enriched AS
SELECT *
FROM (
  SELECT *, explode(items) AS item
  FROM sales) a
INNER JOIN item_lookup b
ON a.item.item_id = b.item_id;

JOIN 종류:

INNER JOIN  → 양쪽 다 있는 행만 (가장 많이 씀)
LEFT JOIN   → 왼쪽 테이블 기준, 오른쪽에 없으면 NULL
RIGHT JOIN  → 오른쪽 테이블 기준, 왼쪽에 없으면 NULL

A: [1, 2, 3]  B: [2, 3, 4]
INNER → [2, 3]      공통인 것만
LEFT  → [1, 2, 3]   A 기준
RIGHT → [2, 3, 4]   B 기준

집합 연산자

-- UNION: 두 쿼리 결과 합치기
SELECT * FROM events
UNION
SELECT * FROM new_events_final

-- INTERSECT: 두 쿼리에 공통으로 있는 행만
SELECT * FROM events
INTERSECT
SELECT * FROM new_events_final

-- MINUS: 첫 번째에는 있고 두 번째에는 없는 행
SELECT * FROM events
MINUS
SELECT * FROM new_events_final

PIVOT — 행을 컬럼으로 변환

Q. PIVOT 사용법을 몰라서 문제를 못 풀었다.

PIVOT이 하는 일:

Before (행으로):               After (열로):
user_id | event_name           user_id | click | purchase | view
user_1  | click            →   user_1  | 3     | 1        | 2
user_1  | click
user_1  | purchase
user_1  | view
user_1  | view

"event_name 값들을 각각 컬럼으로 만들고, 그 안에 집계값을 넣어라"

PIVOT 문법 구조:

SELECT * FROM (
  SELECT 필요한_컬럼들        -- ① 재료 준비 (PIVOT에 필요한 컬럼만)
  FROM 테이블
) PIVOT (
  집계함수(집계할_컬럼)        -- ② 각 칸에 뭘 넣을지 (count, sum 등)
  FOR 피벗할_컬럼 IN (        -- ③ 어떤 컬럼을 기준으로 펼칠지
    '값1', '값2', '값3'       -- ④ 열로 만들 값들 (미리 알고 있어야 함)
  )
)

IN 안의 값을 어떻게 아냐면:

-- 먼저 이걸 실행해서 어떤 값들이 있는지 확인
SELECT DISTINCT event_name FROM events

문제 풀이 — "user_id로 그룹화하고 event_name을 피벗하여 각 이벤트 횟수를 열로":

"각 이벤트 유형의 개수"  →  집계함수 = count(event_name)
"event_name을 피벗"     →  FOR event_name IN (...)
"user_id로 그룹화"      →  재료에 user_id만 남기면 자동으로 그룹화

SELECT * FROM (
  SELECT user_id, event_name    -- user_id랑 event_name만 남김
  FROM events
) PIVOT (
  count(event_name)             -- 각 칸에 이벤트 개수
  FOR event_name IN (
    'cart', 'login', 'main',
    'finalize', 'payment_review',
    -- SELECT DISTINCT event_name FROM events 로 확인한 값들
    'logout', 'home', 'warehouse'
  )
)

고차 함수 — 배열을 직접 조작

일반 함수는 배열에 못 씀. 고차 함수를 쓰면 배열 안을 직접 조작 가능.

FILTER — 조건에 맞는 요소만 남기기:

SELECT
  order_id,
  FILTER(items, i -> i.item_id LIKE "%K") AS king_items
--        ↑배열  ↑반복변수  ↑ 조건 (K로 끝나는 것만 남김)
FROM sales

EXISTS — 배열 안에 조건을 만족하는 요소가 있는지 확인 (true/false 반환):

실습에서 직접 풀었던 문제: items 배열에서 item_name이 "Mattress"로 끝나면 true, "Pillow"로 끝나면 true인 부울 열 생성

CREATE OR REPLACE TABLE sales_product_flags AS
SELECT
  items,
  EXISTS(items, x -> x.item_name LIKE "%Mattress") AS mattress,
  EXISTS(items, x -> x.item_name LIKE "%Pillow") AS pillow
FROM sales;

결과:

items                                      | mattress | pillow
[{item_id: "M_PREM_K", item_name: "Premium King Mattress", ...}] | true  | false
[{item_id: "P_FOAM_S", item_name: "Standard Foam Pillow", ...}]  | false | true
[{item_id: "M_STAN_F", item_name: "Standard Full Mattress",...}] | true  | false

FILTER vs EXISTS 차이:

FILTER  → 조건에 맞는 요소만 남긴 배열 반환
          [item_A, item_B, item_C] → [item_B]

EXISTS  → 조건을 만족하는 요소가 하나라도 있으면 true, 없으면 false 반환
          [item_A, item_B, item_C] → true / false

TRANSFORM — 배열의 각 요소 변환:

SELECT
  order_id,
  TRANSFORM(king_items, k -> CAST(k.item_revenue_in_usd * 100 AS INT)) AS item_revenues
--           ↑배열       ↑반복변수  ↑ 각 요소에 적용할 변환
FROM king_size_sales

3. SQL UDF (사용자 정의 함수)

UDF가 뭔데?

자주 쓰는 SQL 로직을 함수로 등록해서 이름만으로 재사용하는 것.

엑셀의 사용자 정의 함수랑 같은 개념
내가 자주 쓰는 수식을 함수로 저장 → 필요할 때 함수 이름만 호출

UDF 만들기

CREATE OR REPLACE FUNCTION yelling(text STRING)
RETURNS STRING                        -- ← "STRING 타입을 반환할 거야" 라는 선언 (약속)
RETURN concat(upper(text), "!!!")     -- ← 실제로 반환하는 코드 (실행)

Q. RETURNS랑 RETURN이 둘 다 있던데 무슨 차이야?

RETURNS = 메뉴판에 "아메리카노 - 커피입니다" 라고 적어둔 것 (반환 타입 선언)
RETURN  = 실제로 커피를 내어주는 것 (실제 반환 코드)

RETURNS는 어떤 타입을 반환할지 미리 선언, RETURN은 실제로 반환하는 코드.

-- 사용하기
SELECT yelling(food) FROM foods
-- 결과: BEEF!!!, BEANS!!!, POTATOES!!!

UDF 정보 확인

DESCRIBE FUNCTION yelling           -- 기본 정보
DESCRIBE FUNCTION EXTENDED yelling  -- Body 필드에서 실제 SQL 로직 확인 가능

UDF의 특징

• 메타스토어에 저장됨 → 노트북, DBSQL, 작업 등 어디서든 재사용 가능
• 세션이 끝나도 유지됨 → TEMP VIEW와 달리 영구적
• 권한 관리 가능 → USAGE, SELECT 권한 부여 필요

CASE/WHEN — 조건 분기

SELECT *,
  CASE
    WHEN food = "beans" THEN "I love beans"
    WHEN food = "potatoes" THEN "My favorite vegetable is potatoes"
    WHEN food <> "beef" THEN concat("Do you have any good recipes for ", food, "?")
    ELSE concat("I don't eat ", food)
  END
FROM foods

CASE/WHEN을 UDF 안에 넣기

CREATE FUNCTION foods_i_like(food STRING)
RETURNS STRING
RETURN CASE
  WHEN food = "beans" THEN "I love beans"
  WHEN food = "potatoes" THEN "My favorite vegetable is potatoes"
  WHEN food <> "beef" THEN concat("Do you have any good recipes for ", food, "?")
  ELSE concat("I don't eat ", food)
END;

SELECT foods_i_like(food) FROM foods

전체 흐름 정리

6강 데이터 정리
  → NULL 파악 (count_if) → 중복 제거 (GROUP BY + max) → 검증 → 날짜/문자열 변환

7강 고급 SQL 변환
  → JSON 파싱 (from_json, schema_of_json)
  → 배열 조작 (explode, collect_set, flatten, array_distinct)
  → 고차 함수 (FILTER, EXISTS, TRANSFORM)
  → 테이블 결합 (JOIN, UNION, INTERSECT)
  → 형태 변경 (PIVOT)

8강 SQL UDF
  → 반복 로직을 함수로 등록 (CREATE FUNCTION)
  → RETURNS: 반환 타입 선언 / RETURN: 실제 반환 코드
  → CASE/WHEN과 결합해서 제어 흐름 함수 만들기
  → 메타스토어에 저장 → 어디서든 재사용

Spark SQL로 파일 직접 쿼리하는 것부터 Delta 테이블 생성, 외부 소스 연결, 쓰기 방법까지 정리

읽기 전에 — 가장 중요한 개념 구분

이걸 모르면 수업 내내 헷갈린다.

Databricks의 테이블 종류

① 외부 테이블 (= 외부 소스 테이블)
   → 데이터는 밖(CSV, JSON, 외부 DB 등)에 그대로
   → Databricks는 "저기 가면 데이터 있어"라는 정보만 저장
   → USING + LOCATION 키워드가 있으면 외부 테이블

② Delta 테이블
   → 데이터가 Databricks 안으로 들어온 상태
   → 버전 관리, 빠른 쿼리, 자동 최신화 등 모든 기능 사용 가능
   → CTAS (CREATE TABLE AS SELECT) 로 만드는 게 일반적

"JSON 파일에서 외부 테이블을 생성하라" = 데이터는 JSON 파일에 그대로 두고, 그 파일을 테이블처럼 읽을 수 있도록 Databricks에 등록하라는 뜻. Delta 테이블 아님.

Delta 테이블 vs 외부 테이블 vs JDBC

코드로 구분하기

-- 관리형 Delta 테이블
-- USING도 없고 LOCATION도 없음
CREATE TABLE item_lookup AS
SELECT * FROM parquet.`/path`

-- 외부 테이블 (파일)
-- USING + LOCATION 있음
CREATE TABLE sales_csv
USING CSV
OPTIONS (header="true", delimiter="|")
LOCATION "/path/to/file"

-- JDBC (외부 DB)
-- USING JDBC + url + dbtable
CREATE TABLE users_jdbc
USING JDBC
OPTIONS (
  url = "jdbc:mysql://서버주소:3306",
  dbtable = "users"
)

한눈에 비교

문제 풀 때 판단하는 순서

"어떤 테이블 만들어라" 문제가 나오면

1. "관리형" or "Delta" 언급 → USING/LOCATION 없는 CTAS
2. "외부" or 파일 경로 + 옵션 필요 → USING + LOCATION
3. DB 연결 (MySQL, PostgreSQL 등) → USING JDBC + OPTIONS

1. 파일 직접 쿼리하기 (테이블 없이 그냥 읽기)

Databricks에서는 테이블을 만들지 않아도 파일을 바로 쿼리할 수 있다.

-- 단일 파일 쿼리
SELECT * FROM json.`/path/to/001.json`

-- 디렉터리 전체 쿼리 (같은 형식/스키마인 경우)
SELECT * FROM json.`/path/to/directory`

-- Parquet도 동일하게
SELECT * FROM parquet.`/path/to/file`

경로는 반드시 백틱(`) 으로 감싸야 한다. 작은따옴표(')가 아님에 주의.

문법 구조: 파일형식.경로

json.`/path`      -- JSON 파일
parquet.`/path`   -- Parquet 파일
csv.`/path`       -- CSV 파일

뷰와 CTE로 더 편리하게 참조하기

-- 임시 뷰로 등록 (세션 끝나면 사라짐)
CREATE OR REPLACE TEMP VIEW kafka_events AS
SELECT * FROM json.`${DA.paths.kafka_events}`

-- CTE로 사용
WITH events AS (
  SELECT * FROM json.`/path/events`
)
SELECT * FROM events WHERE topic = 'clickstream'

raw 파일 내용 확인 (디버깅용)

-- 텍스트로 읽기
SELECT * FROM text.`/path/to/file`

-- 바이너리로 읽기 (파일 메타데이터 포함)
SELECT * FROM binaryFile.`/path/to/file`

2. 외부 소스 테이블 (= 외부 테이블)

외부 테이블이 뭔데? (≠Delta 테이블)

비유로 먼저 이해하자.

도서관 책 대출 시스템

책(데이터)은 도서관 서가(외부 경로)에 그대로 있음
대출 카드(외부 테이블 정의)만 시스템에 등록
"이 책은 3층 A구역 12번 선반에 있어" 라는 정보만 저장

→ 읽을 때마다 서가에서 꺼내서 읽음
→ 책 자체를 복사해서 들고오지 않음

"외부 테이블"이라는 걸 코드에서 어떻게 알아보나?

딱 두 가지 키워드가 있으면 외부 테이블이다.

CREATE TABLE sales_csv
  (order_id LONG, email STRING, ...)
USING CSV          -- ① Delta가 아닌 형식 (CSV, JSON, JDBC 등)
OPTIONS (
  header = "true",
  delimiter = "|"
)
LOCATION "/path"   -- ② 데이터가 외부 경로에 있음

반대로 Delta 테이블은 이렇게 생겼다:

-- USING도 없고 LOCATION도 없음
CREATE TABLE sales AS
SELECT * FROM parquet.`/path`
-- 데이터가 Databricks 안으로 들어옴

CSV 파일로 외부 테이블 만들기

CSV 파일을 직접 쿼리하면 이런 문제가 생긴다:

헤더가 데이터 행으로 들어옴
모든 컬럼이 하나로 합쳐짐  ← 구분자가 | 인데 Spark가 모름
마지막 컬럼이 잘림

그래서 옵션을 줘서 외부 테이블로 등록한다:

CREATE TABLE sales_csv
  (order_id LONG, email STRING, transactions_timestamp LONG,
   total_item_quantity INTEGER, purchase_revenue_in_usd DOUBLE,
   unique_items INTEGER, items STRING)
USING CSV
OPTIONS (
  header = "true",     -- 첫 행은 헤더야
  delimiter = "|"      -- | 로 구분돼 있어
)
LOCATION "${DA.paths.sales_csv}"

이렇게 하면 SELECT * FROM sales_csv 로 깔끔하게 읽을 수 있다.

왜 외부 소스를 써? 그냥 데이터 가져오면 되는 거 아니야?

현실에서 "그냥 가져오는 것"이 안 되는 경우가 많다:

실무에서는 외부 소스로 원본 연결 → 가공 후 Delta로 저장 흐름이 일반적. 이게 메달리온 아키텍처의 Bronze 단계가 하는 일이다.

외부 테이블의 치명적 한계 — 캐시 문제

-- 외부에서 파일에 데이터 추가됨
SELECT COUNT(*) FROM sales_csv  -- → 숫자 그대로 (캐시 때문에 못 봄)

REFRESH TABLE sales_csv  -- 수동으로 캐시 갱신 필요

SELECT COUNT(*) FROM sales_csv  -- → 이제 최신 숫자

Delta 테이블이면 자동으로 최신 데이터를 읽지만, 외부 테이블은 수동 갱신이 필요하다.

3. JDBC — 외부 DB 연결

JDBC가 뭔데?

우리 집(Databricks)에서
이웃 집(MySQL, PostgreSQL, SQLite 등 외부 DB)에 있는 데이터를
전화선(JDBC 연결)으로 당겨와서 읽는 것

쉽게 말해 Databricks 밖에 있는 데이터베이스 서버에 연결하는 방법이다.

외부 DB가 왜 따로 있냐?

회사에서 이런 상황이 흔하다:

영업팀이 오래전부터 쓰던 Oracle DB
  → 주문 데이터, 고객 데이터가 거기 다 있음
  → Databricks로 이사오기엔 너무 크고 복잡함
  → 그냥 거기 두고 Databricks에서 연결해서 읽음

JDBC로 외부 DB 연결하기

CREATE TABLE users_jdbc
USING JDBC                          -- JDBC 방식으로 연결
OPTIONS (
  url = "jdbc:sqlite:/path/to/db",  -- DB 서버 주소
  dbtable = "users"                 -- 읽어올 테이블 이름
)

주소 형식: jdbc:DB종류://서버주소:포트

jdbc:mysql://192.168.0.1:3306       -- MySQL 연결
jdbc:postgresql://myserver:5432     -- PostgreSQL 연결
jdbc:sqlite:/path/to/file.db        -- SQLite 연결 (파일 기반)

만들고 나면 그냥 일반 테이블처럼 쓸 수 있다:

SELECT * FROM users_jdbc  -- 외부 DB에서 데이터 가져옴

단, 데이터가 Databricks 안으로 복사되는 게 아니라 쿼리할 때마다 외부 DB에서 가져오는 것이라 대용량이면 느릴 수 있다.

4. Delta 테이블 생성

CTAS (Create Table As Select)

"SELECT 결과물로 Delta 테이블 만들기"

CREATE OR REPLACE TABLE sales AS
SELECT * FROM parquet.`/path/sales-historical`

스키마(컬럼 구조)를 직접 안 써도 SELECT 결과에서 자동으로 추론한다.

Parquet, JSON처럼 스키마가 명확한 파일에서 특히 유용하다.

CSV처럼 옵션이 필요한 파일은 CTAS 전에 임시 뷰를 먼저 만들어야 한다:

-- 1단계: CSV 읽기 옵션을 담은 임시 뷰 생성
CREATE OR REPLACE TEMP VIEW sales_tmp
  (order_id LONG, email STRING, ...)
USING CSV
OPTIONS (path = "/path", header = "true", delimiter = "|");

-- 2단계: 임시 뷰로 Delta 테이블 생성
CREATE TABLE sales_delta AS
  SELECT * FROM sales_tmp;

CRAS vs INSERT OVERWRITE

-- CRAS: 스키마도 바꿀 수 있는 강력한 덮어쓰기
CREATE OR REPLACE TABLE events AS SELECT ...

-- INSERT OVERWRITE: 스키마 변경 불가, 더 안전한 덮어쓰기
INSERT OVERWRITE sales SELECT * FROM parquet.`/path`

생성된 열 (Generated Column)

CREATE OR REPLACE TABLE purchase_dates (
  id STRING,
  transaction_timestamp LONG,
  price STRING,
  date DATE GENERATED ALWAYS AS (
    cast(cast(transaction_timestamp/1e6 AS TIMESTAMP) AS DATE))
    COMMENT "generated based on transactions_timestamp column"
)

다른 열의 값을 기반으로 자동 계산되는 열. 삽입 시 직접 값을 안 써도 Delta가 알아서 채워준다.

테이블 제약 조건

-- 조건을 위반하는 데이터가 들어오면 쓰기 에러 발생
ALTER TABLE purchase_dates ADD CONSTRAINT valid_date CHECK (date > '2020-01-01');

Clone — 테이블 복제

Shallow Clone (얕은 복제)

CREATE OR REPLACE TABLE purchases_shallow SHALLOW CLONE purchases

비유:

도서관 책은 그대로 두고
"이 책은 3층 A구역 12번에 있어" 라는 메모(Delta Log)만 복사

읽을 때 → 메모 보고 원본 위치 찾아가서 읽음
수정할 때 → 원본은 건드리지 않고, 내 공간에 새 파일 생성
원본 삭제되면 → 메모는 있는데 책이 없으니 에러

수정해도 원본이 안 바뀌는 이유: Spark는 파일을 절대 덮어쓰지 않는다. 수정 = 기존 파일 건드리는 게 아니라 새 파일을 만드는 것. 그 새 파일이 원본 경로가 아닌 Clone 자체 경로에 생기기 때문에 원본은 그대로다.

Deep Clone (깊은 복제)

CREATE OR REPLACE TABLE purchases_clone DEEP CLONE purchases

비유:

책을 통째로 복사해서 내 책상에 올려두기

원본이 어떻게 되든 내 복사본은 독립적으로 존재
원본 삭제돼도 내 복사본은 멀쩡함
단, 복사하는 데 시간이 오래 걸림

한눈에 비교

5. 테이블에 쓰기 — 4가지 방법

INSERT OVERWRITE — 전체 덮어쓰기

INSERT OVERWRITE sales
SELECT * FROM parquet.`/path`

기존 데이터를 통째로 교체. 스키마(컬럼 구조)가 바뀌면 에러남. 기존 테이블 구조를 지키면서 데이터만 갈아끼울 때.

INSERT INTO — 행 추가

INSERT INTO sales
SELECT * FROM parquet.`/path`

기존 데이터 유지하고 새 행만 추가. 중복 방지 기능 없음. 같은 셀 두 번 실행하면 데이터 두 번 들어감.

MERGE INTO — 똑똑한 병합

MERGE INTO users a
USING users_update b
ON a.user_id = b.user_id
WHEN MATCHED AND a.email IS NULL AND b.email IS NOT NULL THEN
  UPDATE SET email = b.email, updated = b.updated
WHEN NOT MATCHED THEN INSERT *

조건에 따라 UPDATE / INSERT / DELETE를 한 트랜잭션으로 처리.

같은 user_id가 있고 email이 비어있으면  → UPDATE
같은 user_id가 없으면                   → INSERT

INSERT INTO의 중복 문제를 해결할 수 있다. WHEN NOT MATCHED만 쓰면 중복 없는 삽입만 수행:

MERGE INTO events a
USING events_update b
ON a.user_id = b.user_id AND a.event_timestamp = b.event_timestamp
WHEN NOT MATCHED AND b.traffic_source = 'email' THEN
  INSERT *

COPY INTO — 증분 로드

COPY INTO sales
FROM "/path/sales-30m"
FILEFORMAT = PARQUET

이미 가져온 파일은 건너뛰고 새 파일만 가져온다. 매일 새 파일이 쌓이는 폴더에서 반복 실행할 때 유용.

한눈에 비교

전체 흐름 정리

외부 데이터 원본                외부 테이블 등록              Delta 테이블 생성
──────────────                 ────────────────              ────────────────
CSV 파일                        USING CSV                     CTAS로 변환
JSON 파일          →            USING JSON          →         Delta 최적화 적용
외부 DB (MySQL 등)              USING JDBC                    ↓
                                LOCATION 지정                 쓰기
                                (데이터는 밖에 그대로)         (MERGE / COPY INTO)

  [Bronze: 원본 연결]             [Bronze 단계]                [Silver/Gold 단계]

실무에서는 외부 테이블(Bronze) → CTAS로 Delta 변환(Silver) → 정제/집계(Gold) 순서로 메달리온 아키텍처를 구성한다.

오픈소스!

스키마 진화

• 데이터 변경에 따라 테이블 스키마가 자동으로 조정!

스키마 강제

• 입력되는 데이터가 정의된 스키마와 일치하는지 확인

🌟시간 여행 : 아래에서 자세히 다룸

Delta Lake

버전 관리, 최적화, VACUUM 핵심 정리

Databricks 실습 중 헷갈렸던 개념들을 직접 질문하며 정리한 노트

1. DESCRIBE HISTORY — 트랜잭션 로그 조회

DESCRIBE HISTORY beans

Delta Lake는 테이블을 수정하는 모든 작업을 버전으로 기록한다.

• operationParameters : WHERE 조건 등 상세 파라미터 확인 가능
• operationMetrics : 추가/삭제된 행, 파일 수 확인 가능

2. Time Travel — 이전 버전 조회 및 복원

-- 버전으로 조회
SELECT * FROM beans VERSION AS OF 1
SELECT * FROM beans@v1   -- 단축 문법

-- 타임스탬프로 조회
SELECT * FROM beans TIMESTAMP AS OF '2024-01-01'

-- 임시 뷰로 등록
CREATE OR REPLACE TEMP VIEW pre_delete_vw AS
SELECT * FROM beans VERSION AS OF 4

-- 특정 버전으로 복원
RESTORE TABLE beans TO VERSION AS OF 5

📌 질문했던 것: CREATE OR REPLACE TEMP VIEW는 왜 트랜잭션 로그에 안 남을까?

VIEW는 실제 데이터 파일을 건드리지 않는다. "어떻게 볼지"에 대한 정의만 저장하기 때문에 Delta 트랜잭션 로그에 기록되지 않는다. TEMP 여부와 무관하게, 일반 VIEW도 마찬가지.

3. OPTIMIZE + Z-Order — 파일 압축 및 데이터 재배치

OPTIMIZE beans ZORDER BY (name)

이 명령 하나가 두 가지를 동시에 한다:

① Compaction (파일 압축) 작은 파일 수백 개 → 큰 파일 몇 개로 병합

② Z-Ordering (데이터 재배치) name 기준으로 비슷한 값을 같은 파일에 모아둠

Before: 파일1 [Alice, Zoe, Bob]  파일2 [Carol, Alice, Zoe]
After:  파일1 [Alice, Alice]     파일2 [Bob, Bob]     파일3 [Carol, Carol]

→ WHERE name = 'Alice' 쿼리 시 파일 1개만 읽으면 됨 (Data Skipping)

📌 질문했던 것: 파일 개수는 어떻게 확인해?

DESCRIBE DETAIL beans
-- numFiles, sizeInBytes 컬럼 확인

OPTIMIZE 결과 출력에서 numFilesAdded, numFilesRemoved로 변화를 확인할 수 있다. 현재 실습의 beans 테이블은 numFiles: 1, sizeInBytes: 1685 (~1.6KB)로 매우 작아서 Z-Order 효과가 거의 없다. 데이터가 클수록 효과가 극적으로 나타남.

4. VACUUM — 오래된 파일 영구 삭제

-- 삭제될 파일 미리 보기 (실제 삭제 안 함)
VACUUM beans RETAIN 0 HOURS DRY RUN

-- 실제 삭제
VACUUM beans RETAIN 0 HOURS

📌 질문했던 것: DRY RUN이 정확히 뭐야?

"실제로 실행하지 말고, 뭘 할지만 보여줘"라는 뜻. 삭제 대상 파일 목록만 출력한다. RETAIN 0 HOURS는 현재 버전 외 모든 파일을 삭제하는 위험한 옵션이라 DRY RUN으로 먼저 확인하는 것이 중요하다.

기본 보존 기간은 7일이며, 아래 설정으로 우회 가능 (프로덕션에서는 주의):

SET spark.databricks.delta.retentionDurationCheck.enabled = false;
SET spark.databricks.delta.vacuum.logging.enabled = true;

5. VACUUM 후 Time Travel이 불가능한 이유

SELECT * FROM beans@v1
-- FileReadException: 파일 없음 → 에러 발생

📌 질문했던 것: VACUUM 해도 DESCRIBE HISTORY는 보이는 이유?

VACUUM은 파일을 지우는 것이지, 트랜잭션 로그를 지우는 게 아니다.

beans/
├── _delta_log/          ← 로그 파일 (VACUUM이 건드리지 않음)
│   ├── 00000.json       ← "v0: 테이블 생성"
│   └── 00001.json       ← "v1: 데이터 추가"
│
└── part-0001.parquet    ← 실제 데이터 파일 (VACUUM이 삭제)

• DESCRIBE HISTORY → 로그를 읽는 것 → VACUUM 후에도 보임
• beans@v1 조회 → 실제 파일을 찾는 것 → VACUUM 후 FileReadException

즉, VACUUM 후 beans@v1에서 에러가 나는 것이 정상이다. 에러가 나야 VACUUM이 제대로 된 것.

정리: 각 명령어 역할 한눈에 보기

18가지 조합을 다 돌려봤는데, 어떤 조합이 얼마나 좋았는지 기억할 방법이 없으면 곤란하잖아요.

MLflow = 매 실험마다 파라미터, 결과, 모델을 자동으로 저장해주는 도구

코드에서 MLflow가 쓰이는 곳

크게 세 군데예요.

① 실험 공간 설정

mlflow.set_experiment(EXPERIMENT_NAME)
# "이 노트북의 실험들을 여기다 모아서 저장할게"

② 그리드 서치 중 매 조합마다 기록 (숫자 결과만)

with mlflow.start_run(run_name="ALS_GridSearch") as parent_run:      # 전체 실험
    for i, params in enumerate(param_grid):                           # 18번 반복
        with mlflow.start_run(run_name=run_name, nested=True):        # 각 조합
            mlflow.log_params({"rank": rank, "regParam": reg_param})  # 파라미터 저장
            mlflow.log_metric("rmse", rmse)                           # 결과 저장

폴더 구조로 보면 이렇게 쌓여요.

ALS_GridSearch (parent)
├── rank10_reg0.01_iter10 → rmse: 1.23
├── rank10_reg0.01_iter20 → rmse: 1.19
├── rank20_reg0.1_iter10  → rmse: 0.98  ← best
└── ... (18개)

③ 최종 모델 등록 (모델 파일)

with mlflow.start_run(run_name="Final_ALS_Model"):
    mlflow.log_params(best_params)       # 최적 파라미터
    mlflow.log_metric("test_rmse", best_rmse)  # 최종 성능
    mlflow.spark.log_model(             # 모델 파일 자체를 저장
        final_model,
        registered_model_name=f"{CATALOG}.{SCHEMA}.movie_recommender_als"
    )

MLflow가 없으면?

# 18번 돌릴 때마다 직접 print로 기록하고
print(f"rank=10, reg=0.01 → RMSE: 1.23")
print(f"rank=10, reg=0.1  → RMSE: 1.19")
# ... 나중에 어떤 게 최고였는지 찾아야 하고
# 모델 파일도 직접 저장해야 하고
# 다음 주에 "그때 어떤 파라미터 썼지?" 하면 답 없음

MLflow는 이걸 자동화해주고, Databricks UI에서 표로 한눈에 비교할 수 있게 해줘요.

Lakeflow SDP(Spark Declarative Pipeline)

명령적 vs 선언적

명령적 (Imperative) — 요리 과정을 직접 지시

"냉장고에서 닭 꺼내, 180도로 예열해, 30분 구워, 소스 발라, 접시에 담아"

선언적 (Declarative) — 결과만 말하기

"치킨 한 마리 주세요"

어떻게 만드는지는 관심 없고, 뭘 원하는지만 말하는 거예요.

from pyspark import pipelines as dp

1. 데이터셋 유형

Streaming Table vs Materialized View

@dp.table vs @dp.temporary_view

2. 데이터 품질 관리 (Expectations)

세 가지 유형

자주 쓰는 검증 조건

-- NULL 체크
"column IS NOT NULL"

-- 범위 검증
"amount > 0 AND amount < 1000000"

-- 값 목록
"status IN ('active', 'pending', 'completed')"

-- 패턴 매칭
"email LIKE '%@%'"

Python은 데코레이터로 선언하고, SQL은 키워드로 선언하는 차이만 있을 뿐 SDP가 내부적으로 처리하는 방식은 똑같아요.

협업 필터링 VS 콘텐츠 기반 필터링 비슷한 취향을 가진 사용자 집단 행동 패턴을 분석 아이템 자체의 특성과 메타데이터 분석

협업 필터링

A.유저 기반

01 유사 사용자 탐색 코사인 유사도, 피어슨 상관계수 등으로 비슷한 평점 패턴을 가진 유저 그룹 발견

02 이웃 선정 유사도 상위 K명의 이웃 사용자를 선택하여 추천 기반 구축

03 평점 예측 이웃들의 평점을 가중 평균하여 대상 아이템에 대한 예측 점수 계산

주의: 대규모 유저 대상 실시간 계산 비용이 급증하며, 희소 데이터 문제로 정확도가 떨어질 수 있습니다.

B.아이템 기반

좋아한 아이템과 유사한 것 추천. 유사도는 코사인 유사도로 측정

협업 필터링의 한계: 콜드 스타트

콜드스타드 문제 해결 전략

1. 하이브리드 추천 시스템 알고리즘을 여러개 사용

   Multi-armed bandit (밴딧 알고리즘) 여러 슬롯머신을 당겨보고 어떤걸 얼마나 당겨야 가장 높은 수익을 얻을 수 있는지 확인하는 것.

2. 프로필 완성 시작할 때, 스포티파이에서 3명의 아티스트를 선택하도록 하는 것

C. 고도화- 모델 기반

행렬 분해(Matrix Factorization)

DE에게 요구되는 능력치 의외로 RAG를 할 줄 알면 플러스

자격증 나열보다 '증거'를 만들어라 수치가 들어간 엔드투엔드 프로젝트 1-2개 > 자격증10개

자격증 공부할때 외우는건 맞지만 시간이 된다면 기초를 이해하고 가져가는게 자격증이 더 의미있다.

깃허브를 기록하기 시작해라.

computer vision

Segmentation vs 캡셔닝

segmentation이 "어디에 뭐가 있냐" 를 찾는 거라면, 캡셔닝은 "이 장면을 문장으로 설명" 하는 것

Semantic vs Instance Segmentation

핵심 차이는 "개별 객체를 구분하느냐"

Semantic Segmentation은 픽셀마다 "클래스"를 붙이는 것. 사람이 두 명 있어도 둘 다 person 레이블이 붙기 때문에 "어디에 사람이 있나"는 알지만 "몇 명인지, 어디서 어디까지가 한 명인지"는 모름.

Instance Segmentation은 거기서 한 발 더 나아가서, 같은 클래스 안에서도 개별 객체를 구분. person_1, person_2처럼 각각 다른 마스크를 가짐. 사람이 겹쳐 있어도 분리 가능.

어디에 쓰이냐

[Auto Loader]

클라우드 스토리지(Azure Blob Storage, S3 등)에 새로 업로드된 파일을 자동으로 감지해서 읽어오는 Databricks 기능이다.

어떻게 작동하냐

클라우드 스토리지에 새 파일 업로드
        ↓
Auto Loader가 자동 감지
        ↓
Spark로 증분 처리 (새 파일만)

일반 spark.read는 실행할 때 존재하는 파일만 읽고 끝나지만, Auto Loader는 새 파일이 생길 때마다 자동으로 처리한다.

[cloudFiles]

Auto Loader를 사용하기 위한 Databricks 전용 Spark 포맷 이름이다.

# 일반 포맷 (파일 형식)
spark.read.format("csv")
spark.read.format("parquet")

# Auto Loader 전용 포맷
spark.readStream.format("cloudFiles")  # ← 이게 Auto Loader 켜는 스위치

format() 안에 넣는다는 점은 같지만, cloudFiles는 파일 형식이 아니라 Auto Loader 기능을 활성화하는 키워드다. Databricks 외부 환경에서는 사용 불가.

기본 사용법

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")       # 실제 파일 형식은 여기서 지정
    .option("cloudFiles.schemaLocation", "/checkpoint/schema")
    .load("/mnt/storage/incoming/")
)

실전 활용

• Azure Blob Storage, AWS S3, GCS 등 어떤 클라우드 스토리지든 동일한 코드로 연결 가능
• cloudFiles.format만 바꿔서 CSV, JSON, Parquet, 이미지 등 다양한 형식 처리

[Checkpoint]

스트리밍 처리에서 "어디까지 처리했는지"를 기록해두는 위치다.

왜 필요하냐

파일 100개 처리 중 → 클러스터 갑자기 재시작
        ↓
Checkpoint 없으면: 처음부터 다시 (중복 처리)
Checkpoint 있으면: 87번까지 했으니 88번부터 이어서

사용법

query = (df.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoint/my_pipeline")  # ← 여기 지정
    .start("/mnt/output/")
)

실전 활용

• 체크포인트 경로는 DBFS나 클라우드 스토리지에 저장해야 클러스터 재시작 후에도 유지됨
• /tmp에 저장하면 클러스터 재시작 시 날아가므로 주의
• 파이프라인마다 고유한 체크포인트 경로 사용 (공유하면 충돌)

# 좋은 예
/checkpoint/user_logs_pipeline
/checkpoint/image_caption_pipeline

# 나쁜 예
/checkpoint/temp   ← 여러 파이프라인이 공유하면 충돌

[Schema Evolution]

데이터 구조가 바뀌어도 (새 컬럼 추가 등) 자동으로 스키마를 업데이트해주는 기능이다.

왜 필요하냐

1월 데이터: { "name": "철수", "age": 25 }
2월 데이터: { "name": "영희", "age": 23, "email": "y@test.com" }  ← 새 컬럼 추가

Schema Evolution 없으면 2월 데이터 읽을 때 에러 발생. 있으면 자동으로 email 컬럼 추가해서 처리.

Auto Loader에서 활성화

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoint/schema")  # 스키마 저장 위치
    .option("cloudFiles.inferColumnTypes", "true")              # 컬럼 타입 자동 추론
    .load("/mnt/storage/incoming/")
)

실전 활용

• 외부 API 응답 구조가 자주 바뀌는 경우
• 서비스 초기 단계에서 로그 포맷이 계속 변경될 때
• 여러 팀이 데이터를 올리는데 컬럼 구성이 조금씩 다를 때

주의: Schema Evolution은 컬럼 추가는 자동 처리하지만, 컬럼 삭제나 타입 변경은 수동으로 처리해야 할 수 있다.

전체 파이프라인 예시

실제 프로젝트에서 네 가지를 합쳐서 쓰면 이렇게 된다:

# 이미지 캡셔닝 파이프라인 예시
df = (spark.readStream
    .format("cloudFiles")                                         # Auto Loader ON
    .option("cloudFiles.format", "binaryFile")                    # 이미지 파일
    .option("cloudFiles.schemaLocation", "/checkpoint/schema")    # Schema Evolution
    .load("/mnt/storage/images/")                                 # 새 이미지 감지
)

query = (df.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoint/image_pipeline")   # Checkpoint
    .start("/mnt/output/captions/")
)

새 이미지 업로드
      ↓
cloudFiles 감지 (Auto Loader)
      ↓
Schema Evolution으로 구조 변화 자동 대응
      ↓
처리 결과 저장 + Checkpoint 기록
      ↓
클러스터 재시작해도 이어서 처리 가능

[Column Store (열 기반 저장)]

Auto Loader로 읽어온 데이터를 Delta Lake에 저장하는데, Delta Lake가 빠른 이유가 바로 Column Store 방식이기 때문이다.

데이터를 행(Row) 단위로 저장하느냐, 열(Column) 단위로 저장하느냐에 따라 성능이 크게 달라진다.

Row Store vs Column Store

예시 테이블:

Row Store (MySQL 등 일반 DB)

디스크: [1, 철수, 25, 3000] [2, 영희, 23, 4000] [3, 민수, 30, 5000]

Column Store (Parquet, Delta Lake)

디스크: [1, 2, 3] [철수, 영희, 민수] [25, 23, 30] [3000, 4000, 5000]

왜 분석용으로 빠르냐

SELECT AVG(salary) FROM employees

• Row Store → 모든 행 다 읽고 salary만 추출
• Column Store → salary 컬럼만 딱 읽으면 끝

필요한 컬럼만 골라 읽으니 I/O가 압도적으로 줄어든다.

왜 압축률이 좋냐

같은 컬럼끼리 모여 있으면 비슷한 값들이 연속으로 나열되어 압축이 잘 된다.

age 컬럼:    [25, 23, 30, 28, 25, 24...]  ← 비슷한 숫자 → 압축 잘 됨
salary 컬럼: [3000, 3000, 4000, 3000...]  ← 중복 많음  → 압축 잘 됨

Databricks에서는

Parquet, Delta Lake 모두 Column Store 방식이다. 대용량 데이터를 다뤄도 빠른 이유가 여기에 있다.

<6월 15일 9-11시 학습한거 다시 보기>

카탈로그 > 스키마 > 테이블에서 '계보(Lineage)' 탭에서 업스트림(원본), 다운스트림(활용처)

Diamond Data를 활용한 ETL 예제

꼭 알아야하는 개념

• Data lake: 온갖 다양한 _원시 데이터_를 저장하는 호수와 같은 공간 <-> Data warehouse: 정형 데이터 보관

• 정제된 데이터 저장할 때 (silver) 파티셔닝 전략 :

증분 처리 패턴

전체 덮어쓰기(overwrite)는 간단하지만 비효율적. 실무에서는 새로운 데이터만 처리하는 증분 방식이 일반적.

Append 모드

새 레코드를 기존 테이블에 추가. processing_date 같은 타임스탬프 컬럼으로 새 데이터를 식별.

df_new.write.format("delta") \
    .mode("append") \
    .saveAsTable(f"{CATALOG}.{SCHEMA}.bronze_diamonds")

Merge (Upsert) 패턴

존재하는 레코드는 업데이트, 새 레코드는 삽입. CDC(Change Data Capture) 시나리오에 이상적.

MERGE INTO target_table t
USING source_table s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

모드 비교

성능 최적화 팁 5가지

1. 적절한 파티셔닝

자주 필터링되는 컬럼을 파티션 키로 선택하여 불필요한 데이터 스캔을 줄임.

2. 캐싱 활용

여러 번 사용되는 DataFrame은 cache() 또는 persist()로 메모리에 저장.

df.cache()      # 메모리
df.persist()    # 메모리 + 디스크 (대용량 적합)
df.unpersist()  # 사용 후 해제

3. 브로드캐스트 조인

작은 테이블은 broadcast()를 사용하여 모든 노드에 복제하면 셔플을 피할 수 있음.

from pyspark.sql.functions import broadcast
df_result = df_large.join(broadcast(df_small), "key")

여기서 JOIN을 하려면 같은 key를 가진 데이터가 같은 노드에 있어야 합니다. 근데 지금은 흩어져 있으니까 네트워크로 데이터를 주고받아야 합니다. 이 과정이 셔플입니다.

노드 1 → 노드 2로 데이터 전송  ← 느림 😓
노드 3 → 노드 1로 데이터 전송  ← 느림 😓

브로드캐스트 조인은 어떻게 다르냐면 작은 테이블을 모든 노드에 통째로 복사해버립니다.

브로드캐스트 전:
  노드 1: 주문 데이터만 있음
  노드 2: 주문 데이터만 있음
  노드 3: 주문 데이터만 있음

브로드캐스트 후:
  노드 1: 주문 데이터 + 고객 테이블(복사본) ← 로컬에서 바로 JOIN
  노드 2: 주문 데이터 + 고객 테이블(복사본) ← 로컬에서 바로 JOIN
  노드 3: 주문 데이터 + 고객 테이블(복사본) ← 로컬에서 바로 JOIN

각 노드가 자기 데이터랑 로컬에 있는 복사본이랑 바로 JOIN하면 되니까 노드 간 데이터 전송(셔플)이 없어집니다.

4. 컬럼 프루닝

필요한 컬럼만 select하여 메모리 사용량과 I/O를 최소화.

df = spark.read.table("diamonds").select("cut", "price", "carat")

5. 필터 푸시다운

가능한 한 일찍 filter를 적용하여 처리할 데이터 양을 줄임.

df = spark.read.table("diamonds").filter("cut = 'Ideal'").select("price")

(면접 때 사용하면 좋은 용어) First party처럼 한 구독제에서 작동하는....

Databricks 기초

카탈로그

프로젝트(카탈로그)마다 스키마와 테이블을 생성하는 곳.
명시하지 않으면 기본은 delta 테이블이 생성.

Notebook

주피터 노트북 상위호환. 여러명 동시 편집 가능. sql, python, md 언어 동시에 한 문서에서 사용가능 (ex. %sql) html로 export 가능.

카탈로그.스키마.테이블 순으로 접근하거나

USE CATALOG `카탈로그 이름`;
USE SCHEMA 스키마이름; 
-> 실행하면 테이블 이름만으로 사용 가능


DESCRIBE HISTORY delta_students;
-> 버전 확인 후 특정 시점으로 되돌아가기 가능!
RESTORE TABLE delta_students VERSION AS OF 6;

%run

명령어를 사용하여 다른 노트북에서 노트북을 실행할 수 있습니다. 실행할 노트북은 상대 경로로 지정됩니다.

참조된 노트북은 현재 노트북의 일부인 것처럼 실행되므로 호출하는 노트북에서 임시 뷰 및 기타 로컬 선언을 사용할 수 있습니다.

SQL 문법

하나의 트렌젝션으로 처리하고 싶을때

MERGE INTO beans t
USING new_beans s
ON t.name = s.name and t.color = s.color
WHEN MATCHED THEN 
    update set grams=t.grams+s.grams
WHEN NOT MATCHED and s.delicious= true THEN 
  insert *

기본문법

CREATE TABLE beans (
  name STRING,
  ~~
) USING DELTA;
--
INSERT INTO beans VALUES
('pinto', 'brown', 1.5, true)
--
UPDATE beans
SET delicious = true
--
DELETE FROM beans
WHERE delicious = false;
--
DROP TABLE beans;

대시보드

powerBI처럼 시각화 가능. 게시해서 공유 가능.

Genie Space

도메인 특화 자연어 채팅 인터페이스. 사용자가 데이터에 질문을 하면 SQL 쿼리, 결과 테이블, 시각화를 돌려받습니다.

Compute

Photon 가속-> 더 빠르게 계산해주는 대신 비용 1.5배 auto scaling ->

Machine Learning

compute 만들때

AutoML

사람이 직접 모델을 고르고 튜닝하는 대신, Databricks가 알아서 여러 모델을 시도해보고 가장 좋은 걸 찾아줌!

databricks.automl.classify(...)  # 분류 문제 (예: 좋은 와인 vs 나쁜 와인)
databricks.automl.regress(...)   # 회귀 문제 (예: 와인 점수가 몇 점?)
databricks.automl.forecast(...)  # 시계열 예측 (예: 내일 판매량?)

# 💡 timeout_minutes와 max_trials로 비용을 제어합니다

summary = databricks.automl.classify(

    dataset=f"{CATALOG}.{SCHEMA}.wine_quality_lab",

    target_col="is_good_quality",

    primary_metric="f1", # 어떤 기준으로 모델을 평가할지

    timeout_minutes=20,  # 최대 20분 안에 끝내라. 시간 다 되면 그때까지 결과 중 최고를 반환

    exclude_cols=["quality"],  # 원본 quality 컬럼 제외 (정보 누출 방지!)
)

정말 대박인 자동화

1. 데이터 전처리: 결측치, 인코딩, 스케일링 자동으로 해줌
2. 알고리즘 선택: 여러 모델 자동 학습 후 비교
3. 하이퍼 튜닝: 최적 조합을 자동 탐색

• 하이퍼 파라미터 탐색 전략: 베이지안 최적화 기반-> 이전 결과로 다음 후보를 확률적으로 추정하기에 적은 시도로 최적값에 근접

4. 평가 & 재현: 소스 노트북 자동 생성...!

Experiments

AutoML의 결과를 Experiments에서 전부 확인가능!

단일 서비스 리소스로 독립 생성이 가능. 최근에는 Azure AI Foundry (Foundry Tools) 산하로 편입되는 방향으로 가고 있음.

OCR / 문서처리

• OCR/Read 손글씨도 인식하는데 단어 순서를 반대로 추출하기로 함... 가장 기본적인 text 추출

• Layout 테이블, figure(그래프), 체크박스도 추출해줌.

• General documents 신청서 같은 서류에서 key-value pair와 그냥 문장 덩어리를 추출함.

  prebuilt model

• Invoices Query fields로 field 추가하면 더 잘 인식함.

  +a

Azure AI Language Studio

https://language.cognitive.azure.com/home 지역별로 제공 여부가 다르고 East US의 경우 모든 서비스를 제공

• 정보 추출: 비구조화된 텍스트에서 범주 그룹에 속하는 개체를 추출해줌. NER(Named Entity Recognition)

• 텍스트 분류: 텍스트의 언어를 감지하거나 감정 분류

  • 고객 감정 분석
  • 자동으로 텍스트 분류: 사용자 지정 텍스트 분류, 언어 감지

• 질문과 대화형 언어 이해

  • 고객 쿼리에 응답 QnA: 사용자 지정 질문 답변
    • 대화형 환경 구축: 대화 언어 이해

• 텍스트 요약

• 번역

Speech Studio

음성 > 텍스트

음성 텍스트 변환 빠른 시작 - Foundry Tools | Microsoft Learn

• 실시간 음성 텍스트 변환

• Whisper 모델 이용한 실시간 음성 텍스트 변환

• 일괄 처리 음성 텍스트 변환 대용량 파일을 비동기적으로 처리

• custom speech 음성 모델 고르고 말하기 스타일 적용

• 발음 평가 👍

• 음성 번역 다른 언어로 번역

텍스트 > 음성

텍스트 음성 변환 빠른 시작 - 음성 서비스 - Foundry Tools | Microsoft Learn

• 전문적인 음성 미세 조정 고유한 음성 생성

• 개인 음성 -> 사용 불가.... 인간의 음성 샘플에서 개인 음성 tts

• 오디오 콘텐츠 만들기 음성 말하기 스타일, 속도, 발음 조정

• 텍스트 음성 변환 아바타 음성에 매치하는 아바타 영상 생성

Azure AI Vision

Foundry에서 사용.

• Image analysis 이미지 감지, 분류, 주석, 이해
• video analysis 공간 분석(영상의 움직임 분석), 비디오 검색(자연어로 검색가능한 비디오 인덱스 생성)
• OCR
• Facial recognition 얼굴 인식. 신원확인이나 얼굴 흐림 처리 등 다양한 활용.

사용위치 접속법

Foundry와 Azure AI를 생성하고 (East US 지역 가능 아마도) 아래 이미지와 같이 프로젝트를 생성. 아래와 같이 접속하면 사용 가능.

활용처

Common object detection: 물체의 이름 + 좌표 Image captioning: 사진 전체를 문장으로 설명 Dense captioning: 각 영역을 문장으로 설명 Image search: 비슷한 이미지 검색 기능 Common tag extraction: 관련 키워드를 신뢰도 점수와 함께 추출 Optical character recognition Other image analysis capabilities

Custom Vision

개별 리소스로 생성해서 별도의 포털(Custom Vision portal)에서 사용가능.

두개가 한번에 생성되는데 코드로 실습할 때는 train과 predict로 둘 다 필요함. 코드 실습 참고 문서

가위와 포크를 구분하는 모델 학습 실습을 진행함. 코드로 진행한 것도 게시하면 포털에서 확인가능.

4. Stream Analytics에서 Power BI를 출력으로 추가할 때, '그룹 작업 영역'에 아까 저장해둔 작업 영역 ID를 입력.
5. '권한 부여'를 누르고, 로그인 화면이 뜨면 작업영역이랑 동일한 powerBI 계정으로 로그인
6. powerBI로 돌아와서 '데이터 가져오기' > 'powerBI 의미 체계 모델' > '연결' > 내가 사용하려는 데이터셋 선택하면 완료!

피드백 정리

✅ 잘 된 부분

발표 앞부분의 논리 흐름이 좋았어. 문제 제기와 Why Now까지의 흐름이 설득력 있었고 첫인상이 좋았다는 평가를 받았어. 특히 규제가 존재해서 우리 서비스가 필요하다는 부분이 가장 좋게 평가됐어.

**문제의 원인**
1. 점역이 비싸고 수작업이라 오래 걸린다
2. 접근성 표준은 있지만 그걸 준수할 기술이 없다
3. 결과적으로 7.6%의 자료만 접근성이 보장된다

**지금 해결해야 하는 이유**
1. **규제가 존재**해서 우리 서비스가 법적 필수재가 된다
2. 예산은 이미 배정되어 있다

**기존 솔루션의 문제**
1. 충분한 자동화가 없다
2. 서비스들이 분산되어 있다

**우리 솔루션**
1. 충분한 자동화
2. 분산된 서비스를 통합한 플랫폼

다른 팀 적용: 발표 초반 문제 정의와 감정적 훅은 심사위원의 첫인상을 결정해. 앞부분에 가장 많은 공을 들여.

❌ 부족했던 부분

1. 솔루션 설명이 불명확했어

심사위원이 "뭘 하려는 건지", "결과물이 뭔지"를 계속 물었어. 흩어진 기술들을 모아서 빈 공간을 채운다는 핵심 개념이 발표에서 전달되지 않았어. 특히 Pedagogical Masking 같은 독자적 기술이 충분히 설명되지 않고 슬라이드에만 적혀서 넘어갔어.

다른 팀 적용: 솔루션의 핵심을 한 문장으로 말할 수 없다면 청중도 이해하지 못해. "우리는 X를 Y방식으로 해결한다"를 한 문장으로 압축해서 반드시 발표에 포함시켜.

2. 교과서인지 학습지인지 명확하지 않았어

타깃 제품이 무엇인지 흐지부지 넘어갔어. 심사위원 입장에서는 범위가 불명확하면 신뢰도가 떨어져.

다른 팀 적용: 타깃을 좁게 정의할수록 설득력이 높아져. "우리는 모든 것을 한다"보다 "우리는 이것부터 한다"가 더 강해.

3. 비즈니스 모델 설명이 작동 방식 없이 숫자만 있었어

돈 버는 구조는 설명됐지만 실제로 어떻게 운영되는지가 빠졌어. 심사위원은 비즈니스 모델에서 두 가지를 봐. 어떻게 작동하는지, 그리고 어떻게 돈을 버는지. 둘 다 한눈에 보여야 해.

다른 팀 적용: 비즈니스 모델 슬라이드는 화려하게 만들 필요 없어. "A가 B를 하면 C가 발생하고 우리는 D를 가져간다"는 흐름이 한눈에 보이면 충분해.

X	?

4. Callback 답변이 너무 길었어

질문의 의도를 파악하고 요점만 말하는 게 필요했어. 심사위원이 원한 건 "있는 걸 모아서 빈 공간을 채운다" 한 문장이었는데 너무 길게 답했어.

다른 팀 적용: Q&A에서 답변은 두 문장 이내로 시작해. 핵심을 먼저 말하고, 필요하면 한 문장을 덧붙여. 길게 말할수록 핵심이 묻혀.

💡 핵심 인사이트 — 심사위원이 가장 강조한 것

"결과물은 똑같아. How to가 다른 거야. 오퍼레이션이 다른 거야."

기존 점자 교재와 Aequalis의 결과물은 같아. 차이는 만드는 방식이야. 이걸 한 문장으로 압축해서 발표에 넣었어야 했어.

심사위원이 제안한 한 줄:

"흩어져 있는 점역 기술들을 모아서 큐레이션하고, 빈 공간은 우리만의 기술로 채운다. 결과물은 같지만 훨씬 빠르고 정확하게."

다른 팀 적용: 발표에서 "우리만의 한 줄"이 없으면 심사위원은 발표가 끝난 후 아무것도 기억하지 못해. 발표 준비의 마지막 단계는 항상 "이 발표를 한 문장으로 요약하면?"에 답하는 거야.

🔍 추가 피드백 인사이트

플랫폼 비즈니스는 차별화가 핵심이야

플랫폼 비즈니스는 비슷한 게 너무 많아. "찾으면 나온다"는 말이 나올 수 있어. 그래서 왜 우리가, 왜 지금, 왜 이 시장에서 이 생각을 하게 됐는지를 보여줘야 해. 차별점은 기능이 아니라 문제 정의에서 나와. 문제 정의가 정확할수록 "이 팀이 들어가면 해결된다"는 확신을 줄 수 있어.

다른 팀 적용: 제일 좋은 비즈니스는 벤치마킹이라고 했어. 이미 검증된 모델을 가져오되, 우리가 이 문제를 발견한 맥락과 이유를 보여주는 게 차별점이야. "왜 우리가 이걸 만들게 됐는가"가 설득력의 핵심이야.

4단계 구조를 지켜

심사위원이 직접 제시한 발표 구조야.

1. 현상 — 이런 일이 일어나고 있다
2. 문제 정의 — 이게 왜 문제인가
3. 솔루션 — 우리는 이렇게 해결한다
4. 임팩트 — 그 결과 이렇게 달라진다

다른 팀 적용: 4분 발표든 10분 발표든 이 구조를 벗어나지 마. 솔루션을 먼저 말하고 싶은 충동을 참아. 문제가 먼저 공감돼야 솔루션이 설득돼.