1단계: 노트북 기본 조작 및 유틸리티
하나의 노트북 안에서 파이썬과 SQL을 자유롭게 넘나들고, 파일 시스템을 탐색하는 기본 코드입니다.
언어 변경 및 매직 커맨드
기본적으로 노트북의 언어가 지정되어 있지만, 매직 커맨드(%)를 통해 셀 단위로 언어를 변경할 수 있습니다.
# 파이썬 기본 셀 실행
print("I'm running Python!")# %sql 매직 커맨드로 SQL 실행
%sql
SELECT "I'm running SQL!"파일 시스템 탐색 및 데이터 출력
dbutils.fs.ls()를 사용해 Databricks 파일 시스템(DBFS) 경로 내의 파일 목록을 가져오고, display()로 표 형태로 깔끔하게 렌더링합니다.
import DApath = f"{DA.paths.datasets}" # (실습 원본 코드의 변수 할당)
path = f"{DA.paths.datasets}"
# 파일 목록 가져오기 및 표 형태로 출력하기
files = dbutils.fs.ls(path)
display(files)2단계: 외부 파일에서 데이터 추출 및 연결
데이터를 델타 테이블로 만들기 전, 원시 파일(JSON, CSV)이나 데이터베이스를 직접 읽거나 임시로 연결하는 방법입니다.
1. 자체 설명적 형식(JSON, Parquet) 직접 쿼리
구조가 명확한 파일은 백틱(`)으로 경로를 감싸 바로 SELECT 할 수 있습니다.
-- 단일 파일 직접 쿼리
SELECT * FROM json.`${DA.paths.kafka_events}/001.json`
-- 폴더 내의 모든 파일 직접 쿼리
SELECT * FROM json.`${DA.paths.kafka_events}`2. CSV 파일 등 외부 소스를 테이블로 연결
헤더나 구분자 설정이 필요한 경우 USING 구문과 OPTIONS를 사용합니다.
-- CSV 파일을 외부 테이블로 연결
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;
SELECT COUNT(*) FROM sales_csv;3. 캐시 새로고침 (외부 소스의 한계 극복)
원본 데이터가 바뀌어도 스파크는 캐시(과거 기억)를 봅니다. 이를 수동으로 최신화하는 명령어입니다.
REFRESH TABLE sales_csv;4. 외부 SQL 데이터베이스(JDBC) 연결
DROP TABLE IF EXISTS users_jdbc;
CREATE TABLE users_jdbc
USING JDBC
OPTIONS (
url = "jdbc:sqlite:${DA.paths.ecommerce_db}",
dbtable = "users"
);3단계: 델타 테이블 생성 및 스키마 관리
데이터 분석의 핵심 저장소인 델타 테이블을 생성하고, 품질을 관리하는 코드입니다.
1. CTAS (Create Table As Select)
원본 데이터의 스키마를 유추하여 가장 빠르게 델타 테이블을 만듭니다.
CREATE OR REPLACE TABLE sales AS
SELECT * FROM parquet.`${da.paths.datasets}/ecommerce/raw/sales-historical`;
-- 원하는 컬럼만 선택하고 이름(AS)을 바꾸어 새 테이블/뷰 생성
CREATE OR REPLACE TABLE purchases AS
SELECT
order_id AS id,
transaction_timestamp,
purchase_revenue_in_usd AS price
FROM sales;2. 생성된 열(Generated Columns) 및 코멘트 추가
기존 데이터를 바탕으로 자동 계산되는 열을 추가합니다.
CREATE OR REPLACE TABLE purchase_dates (
id STRING,
transaction_timestamp LONG,
price STRING,
-- 타임스탬프를 DATE 타입으로 자동 변환하는 열 생성
date DATE GENERATED ALWAYS AS (
cast(cast(transaction_timestamp/1e6 AS TIMESTAMP) AS DATE))
COMMENT "generated based on `transactions_timestamp` column"
);3. 데이터 품질 제약 조건 (CHECK)
조건에 맞지 않는 쓰레기 데이터의 삽입을 원천 차단합니다.
ALTER TABLE purchase_dates
ADD CONSTRAINT valid_date
CHECK (date > '2020-01-01');4. 메타데이터 보강 및 파티셔닝
현재 시간(current_timestamp())과 소스 파일명(input_file_name())을 메타데이터로 기록합니다.
CREATE OR REPLACE TABLE users_pii
COMMENT "Contains PII"
LOCATION "${da.paths.working_dir}/tmp/users_pii"
PARTITIONED BY (first_touch_date)
AS
SELECT *,
cast(cast(user_first_touch_timestamp/1e6 AS TIMESTAMP) AS DATE) first_touch_date,
current_timestamp() updated,
input_file_name() source_file
FROM parquet.`${da.paths.datasets}/ecommerce/raw/users-historical/`;5. 델타 테이블 복제 (CLONE)
- DEEP CLONE: 데이터 전체를 물리적으로 복사 (증분 동기화 가능)
- SHALLOW CLONE: 로그(메타데이터)만 복사하여 빠르게 테스트용 테이블 생성
-- 물리적 복사
CREATE OR REPLACE TABLE purchases_clone
DEEP CLONE purchases;
-- 논리적(메타데이터) 복사
CREATE OR REPLACE TABLE purchases_shallow_clone
SHALLOW CLONE purchases;4단계: 테이블 데이터 조작 (DML) 및 쓰기
테이블이 생성된 후 데이터를 삽입, 수정, 삭제, 병합하는 핵심 로직들입니다.
1. 기본 DML (삽입, 수정, 삭제)
-- 데이터 삽입 (단건 및 다건)
INSERT INTO beans VALUES
("black", "black", 500, true),
("lentils", "brown", 1000, true);
-- 데이터 수정
UPDATE beans SET grams = 1500 WHERE name = 'pinto';
-- 데이터 삭제
DELETE FROM beans WHERE delicious = false;2. 덮어쓰기 (Overwrite)
완전히 새로운 구조로 덮어쓸 때는 CREATE OR REPLACE, 기존 스키마를 유지하며 안전하게 덮어쓸 때는 INSERT OVERWRITE를 사용합니다.
INSERT OVERWRITE sales
SELECT * FROM parquet.`${da.paths.datasets}/ecommerce/raw/sales-historical/`;3. 삽입 (추가)
기존 데이터에 단순히 이어 붙입니다 (중복 방지 안 됨).
INSERT INTO sales
SELECT * FROM parquet.`${da.paths.datasets}/ecommerce/raw/sales-30m`;4. 병합 (MERGE INTO - Upsert)
가장 강력한 기능 중 하나로, 조건에 따라 삽입, 수정, 삭제를 한 번의 트랜잭션으로 처리합니다.
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 *;
-- (응용) 중복 방지용 삽입 전용 병합
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 *;5. 증분 로드 (COPY INTO)
새로 추가된 원본 파일만 멱등성 있게(중복 없이) 가져옵니다.
COPY INTO sales
FROM "${da.paths.datasets}/ecommerce/raw/sales-30m"
FILEFORMAT = PARQUET;5단계: 델타 레이크 고급 기능 (최적화 및 복구)
델타 레이크의 트랜잭션 로그를 활용한 버저닝, 성능 최적화, 그리고 데이터 정리 기능입니다.
1. 테이블 정보 및 로그 확인
DESCRIBE EXTENDED students;
DESCRIBE DETAIL students;
-- 테이블의 모든 변경(트랜잭션) 이력 확인
DESCRIBE HISTORY students; 2. 성능 최적화 (OPTIMIZE & ZORDER)
작은 파일들을 합치고, 자주 조회하는 컬럼(예: id)을 기준으로 데이터를 정렬해 검색 속도를 높입니다.
OPTIMIZE students ZORDER BY id;3. 시간 여행 (Time Travel) 및 복구 (RESTORE)
과거의 특정 시점(버전) 데이터를 조회하거나 테이블 전체를 롤백합니다.
-- 버전 3 시점의 데이터 조회
SELECT * FROM students VERSION AS OF 3;
-- 실수로 전체 삭제(DELETE FROM students)를 한 경우, 버전 8로 롤백
RESTORE TABLE students TO VERSION AS OF 8;4. 과거 데이터 파일 정리 (VACUUM)
시간 여행에 필요 없는 오래된 물리적 파일들을 삭제하여 저장 공간을 절약합니다. (실습용으로 0시간을 주었으나, 실제로는 기본 7일을 유지합니다).
-- 지워질 파일 목록 미리보기 (DRY RUN)
VACUUM students RETAIN 0 HOURS DRY RUN;
-- 실제 정리 실행
VACUUM students RETAIN 0 HOURS;6단계: 결측치와 중복 처리
분석을 시작하기 전, 비어있는 값(Null)이나 중복된 데이터를 찾아 알맞게 가공하는 필수 과정입니다.
1. 빈 값(Null) 개수 세기
count(*)는 Null을 포함한 전체 줄 수를 세지만, count(컬럼명)은 Null을 무시합니다. 따라서 Null 자체가 몇 개인지 정확히 세려면 count_if를 써야 합니다.
SELECT
count_if(user_id IS NULL) AS missing_user_ids,
count_if(email IS NULL) AS missing_emails
FROM users_dirty;2. 중복 데이터 제거하기
DISTINCT는 완전히 똑같은 행을 하나로 묶어주지만, 모든 칸이 Null인 행들도 1개로 쳐버리는 함정이 있습니다. 그래서 핵심 데이터(ID 등)가 같은 것끼리 GROUP BY로 묶고, 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 -- 아예 ID가 없는 쓰레기 데이터는 버림
GROUP BY user_id, user_first_touch_timestamp;3. 컴퓨터 시간(밀리초)과 문자열 예쁘게 가공하기
SELECT *,
-- 밀리초를 초 단위로 바꾸어 시간 형식으로 캐스팅
CAST(user_first_touch_timestamp / 1e6 AS timestamp)
AS first_touch,
-- 보기 편한 날짜 형식(MMM d, yyyy)으로 변환
date_format(first_touch, "MMM d, yyyy")
AS first_touch_date,
-- 이메일에서 @ 뒤의 도메인만 정규식(regexp)으로 쏙 뽑아내기
regexp_extract(email, "(?<=@).+", 0)
AS email_domain
FROM deduped_users;7단계: 복잡한 데이터 구조(JSON/리스트) 분해하기
하나의 칸 안에 딕셔너리(JSON)나 장바구니처럼 여러 데이터가 묶여있는 배열(Array)을 다루는 방법입니다.
1. JSON 데이터에서 값만 쏙 빼오기 (:)
JSON 텍스트 안에서 device 값과 geo 안의 city 값을 콜론(:)으로 바로 뽑아냅니다.
SELECT value:device, value:geo:city FROM events_strings;2. JSON을 엑셀 표처럼 쫙 펼치기 (schema_of_json & from_json)
데이터 구조를 알아서 파악하게 한 뒤, .*을 써서 모든 속성을 각각의 컬럼(열)으로 평면화시킵니다.
-- 1) schema_of_json으로 샘플의 뼈대를 파악하고, from_json으로 묶인 데이터를 풉니다.
CREATE OR REPLACE TEMP VIEW parsed_events AS
SELECT from_json(value, schema_of_json('{"device":"Linux", ...}')) AS json
FROM events_strings;
-- 2) 별표(*)를 써서 JSON 안의 속성들을 각각의 개별 컬럼으로 쫙 펼칩니다.
CREATE OR REPLACE TEMP VIEW new_events_final AS
SELECT json.*
FROM parsed_events;3. 장바구니(배열) 쪼개고 묶기 (explode & collect_set)
- explode (쪼개기): 한 칸에 들어있는 3개의 상품을 폭죽처럼 터뜨려, 3개의 행(줄)으로 나누어 줍니다.
- collect_set (묶기): 반대로 여러 줄로 흩어진 유저의 이력을 다시 하나의 배열(리스트)로 묶어주며, 중복은 알아서 제거합니다.
-- 배열 쪼개기
SELECT user_id, explode(items) AS item FROM events;
-- 배열로 다시 묶기 (중복 제거 포함)
SELECT user_id, collect_set(event_name) AS event_history
FROM events GROUP BY user_id;8단계: 데이터 합치고 회전시키기 (집합과 PIVOT)
여러 테이블을 더하고 빼거나, 요약 표를 만드는 과정입니다.
1. 집합 연산 (Set Operators)
- UNION: 두 테이블을 위아래로 모두 합칩니다.
- INTERSECT: 두 테이블 양쪽에 모두 있는 데이터(교집합)만 남깁니다.
- MINUS: 앞 테이블에서 뒤 테이블과 겹치는 부분을 뺍니다(차집합).
2. PIVOT (데이터 90도 회전하기)
세로로 길게 늘어선 데이터를 가로로 눕혀서, 아이템별로 한눈에 보기 좋은 '크로스탭(요약 표)'을 만듭니다.
CREATE OR REPLACE TABLE transactions AS
SELECT * FROM (
SELECT email, order_id,
item.item_id AS item_id,
item.quantity AS quantity
FROM sales_enriched
) PIVOT (
-- 수량(quantity)을 다 더해서, 아이템 ID별로 각각의 새로운 열(가로 칸)을 만듭니다.
sum(quantity) FOR item_id
in ('P_FOAM_K', 'M_STAN_Q', 'P_FOAM_S')
);9단계: 고차 함수 & UDF
배열 안의 데이터를 직접 가공하거나, 매번 쓰기 귀찮은 조건문을 '나만의 함수'로 만드는 방법입니다.
1. 고차 함수 (Higher-Order Functions)
- FILTER: 배열 안에서 조건에 맞는 것만 걸러냅니다.
- TRANSFORM: 배열 안의 모든 요소에 일괄적으로 계산을 적용합니다.
-- K로 끝나는 킹사이즈(King) 제품만 남기기
SELECT FILTER (items, i -> i.item_id LIKE "%K") AS king_items
FROM sales;
-- 남은 킹사이즈 제품들의 매출에 100을 곱해서 정수로 변환하기
SELECT TRANSFORM (king_items, k -> CAST(k.item_revenue_in_usd * 100 AS INT))
FROM king_size_sales;2. SQL UDF (사용자 정의 함수 만들기)
길고 복잡한 CASE WHEN 조건문을 한 번만 함수로 등록해 두면, 언제 어디서든 짧은 이름으로 불러 쓸 수 있어 코드가 아주 깔끔해집니다. 데이터베이스에 등록되므로 권한이 있는 다른 사람과도 공유할 수 있습니다.
-- 1. 나만의 함수 정의하기
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"
ELSE concat("I don't eat ", food)
END;
-- 2. 정의한 함수 깔끔하게 사용하기
SELECT foods_i_like(food) FROM foods;