Parquet과 Data Lake

name: bigdata-week06

x-spark: &spark
  image: spark:4.1.3-scala2.13-java21-python3-ubuntu
  environment:
    PYSPARK_PYTHON: python3
  init: true
  stop_grace_period: 30s

services:
  spark-master:
    <<: *spark
    hostname: spark-master
    entrypoint: ["/opt/spark/bin/spark-class"]
    command:
      - org.apache.spark.deploy.master.Master
      - --host
      - spark-master
      - --port
      - "7077"
      - --webui-port
      - "8080"
    ports:
      - "127.0.0.1:18080:8080"
    healthcheck:
      test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080', timeout=3)"]
      interval: 5s
      timeout: 5s
      retries: 20
      start_period: 20s

  spark-worker:
    <<: *spark
    entrypoint: ["/bin/bash", "-lc"]
    command:
      - >-
        exec /opt/spark/bin/spark-class
        org.apache.spark.deploy.worker.Worker
        --host "$$HOSTNAME"
        --cores 1
        --memory 1g
        --webui-port 8081
        --work-dir /tmp/spark-work
        spark://spark-master:7077
    volumes:
      - ./data:/opt/spark/data:ro
      - ./output:/opt/spark/output
    depends_on:
      spark-master:
        condition: service_healthy

  spark-client:
    <<: *spark
    hostname: spark-client
    entrypoint: ["/bin/bash", "-lc"]
    command: ["exec sleep infinity"]
    ports:
      - "127.0.0.1:24040:4040"
    volumes:
      - ./apps:/opt/spark/apps:ro
      - ./data:/opt/spark/data:ro
      - ./output:/opt/spark/output
    depends_on:
      spark-master:
        condition: service_healthy

import argparse
import hashlib
import time
from datetime import date
from decimal import Decimal
from pathlib import Path

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (
    DecimalType, IntegerType, LongType, StringType,
    StructField, StructType, TimestampType,
)

ROOT = Path("/opt/spark/output/week06-lab")
CLEAN = ROOT / "orders_clean"
SCHEMA = StructType([
    StructField("order_id", LongType(), True),
    StructField("user_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("price", DecimalType(12, 2), True),
    StructField("event_time", TimestampType(), True),
])


def read_csv(spark):
    return (spark.read.option("header", True)
            .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
            .schema(SCHEMA).csv("/opt/spark/data/orders.csv"))


def normal_orders(raw):
    return raw.filter(
        F.col("quantity").isNotNull() & F.col("price").isNotNull()
        & (F.col("quantity") > 0) & (F.col("price") >= 0)
    ).withColumn("amount", F.col("quantity") * F.col("price"))


def saved_orders(spark):
    if not CLEAN.is_dir():
        raise RuntimeError("Run --step clean before this step")
    return spark.read.parquet(str(CLEAN))


def clean(spark):
    raw = read_csv(spark)
    valid = normal_orders(raw)
    # CSV input is unchanged; replace only this lab's output directory.
    valid.write.mode("overwrite").option("compression", "snappy").parquet(str(CLEAN))
    saved = spark.read.parquet(str(CLEAN))
    assert raw.count() == 12 and saved.count() == 9
    assert valid.orderBy("order_id").collect() == saved.orderBy("order_id").collect()
    print("RAW_ROWS=12\nVALID_ROWS=9\nSAVED_ROWS=9", flush=True)
    print("OUTPUT=" + str(CLEAN), flush=True)


def inspect(spark):
    saved = saved_orders(spark)
    saved.printSchema()
    saved.select("order_id", "user_id", "amount").orderBy("order_id").show(truncate=False)
    print("DATA_FILES=" + str(len(list(CLEAN.glob("part-*.parquet")))), flush=True)
    assert saved.schema["amount"].dataType.simpleString() == "decimal(23,2)"
    assert [r.order_id for r in saved.orderBy("order_id").collect()] == [1, 2, 3, 5, 7, 8, 10, 11, 12]
    print("SCHEMA_AND_VALUES=PASS", flush=True)


def projection(spark):
    selected = saved_orders(spark).select("user_id", "amount")
    # explain prints a plan; show triggers result calculation.
    selected.explain(mode="formatted")
    selected.orderBy("user_id", "amount").show(truncate=False)
    assert selected.count() == 9
    print("SELECTED_COLUMNS=user_id,amount", flush=True)


def pushdown(spark):
    selected = saved_orders(spark).filter(F.col("amount") >= 30000).select("order_id", "user_id", "amount")
    selected.explain(mode="formatted")
    selected.orderBy("order_id").show(truncate=False)
    assert [r.order_id for r in selected.orderBy("order_id").collect()] == [2, 5, 8]
    print("HIGH_ORDER_IDS=2,5,8", flush=True)


def partition(spark):
    # Independent four-row example from lecture chapter 10.
    original = spark.createDataFrame([
        (101, "U001", Decimal("24000.00"), date(2026, 9, 1)),
        (102, "U002", Decimal("35000.00"), date(2026, 9, 1)),
        (103, "U003", Decimal("20000.00"), date(2026, 9, 2)),
        (104, "U001", Decimal("35000.00"), date(2026, 9, 2)),
    ], "order_id long, user_id string, amount decimal(12,2), event_date date")
    path = ROOT / "partition_demo"
    original.write.mode("overwrite").partitionBy("event_date").parquet(str(path))
    for file in sorted(path.rglob("*.parquet")):
        print("FILE=" + str(file.relative_to(path)), flush=True)
    one_file = sorted((path / "event_date=2026-09-01").glob("*.parquet"))[0]
    inside_file = spark.read.parquet(str(one_file))
    saved = spark.read.parquet(str(path))
    print("SINGLE_FILE_COLUMNS=" + ",".join(inside_file.columns), flush=True)
    print("DATASET_COLUMNS=" + ",".join(saved.columns), flush=True)
    saved.orderBy("order_id").show(truncate=False)
    selected = saved.filter(
        (F.col("event_date") == "2026-09-01") & (F.col("amount") >= 30000)
    ).select("order_id", "user_id", "amount")
    selected.explain(mode="formatted")
    selected.show(truncate=False)
    assert "event_date" not in inside_file.columns and "event_date" in saved.columns
    assert saved.select(*original.columns).orderBy("order_id").collect() == original.orderBy("order_id").collect()
    assert [r.order_id for r in selected.collect()] == [102]
    assert len(list(path.glob("event_date=*"))) == 2
    print("DATE_DIRECTORIES=2\nPARTITION_ROWS=4\nDAILY_HIGH_ORDER_IDS=102", flush=True)


def append(spark):
    valid = saved_orders(spark)
    path = ROOT / "append_demo"
    # Reset on each invocation, then append one duplicate order deliberately.
    valid.write.mode("overwrite").parquet(str(path))
    before = {f.name: hashlib.sha256(f.read_bytes()).hexdigest() for f in path.glob("part-*.parquet")}
    print("BEFORE_FILES=" + str(len(before)) + "\nBEFORE_ROWS=9", flush=True)
    valid.filter(F.col("order_id") == 1).write.mode("append").parquet(str(path))
    after = {f.name: hashlib.sha256(f.read_bytes()).hexdigest() for f in path.glob("part-*.parquet")}
    saved = spark.read.parquet(str(path))
    assert saved.count() == 10 and saved.filter(F.col("order_id") == 1).count() == 2
    assert len(after) > len(before) and all(after.get(name) == value for name, value in before.items())
    print("AFTER_FILES=" + str(len(after)) + "\nAFTER_ROWS=10\nORDER_1_COUNT=2", flush=True)
    print("ORIGINAL_FILES_UNCHANGED=True", flush=True)


def options(spark):
    valid = saved_orders(spark)
    path = ROOT / "file_limit_demo"
    valid.write.mode("overwrite").option("compression", "gzip").option("maxRecordsPerFile", 4).parquet(str(path))
    files = sorted(path.glob("part-*.parquet"))
    counts = [spark.read.parquet(str(file)).count() for file in files]
    for file in files:
        print("FILE=" + file.name, flush=True)
    assert sum(counts) == 9 and all(0 < n <= 4 for n in counts)
    print("ROWS_PER_FILE=" + ",".join(str(n) for n in counts), flush=True)
    print("TOTAL_ROWS=9", flush=True)


def small_files(spark):
    path = ROOT / "tiny_files_demo"
    saved_orders(spark).write.mode("overwrite").partitionBy("order_id").parquet(str(path))
    directory_count = len(list(path.glob("order_id=*")))
    file_count = len(list(path.rglob("*.parquet")))
    assert directory_count == 9 and file_count == 9
    print(f"ORDER_DIRECTORIES={directory_count}\nPARQUET_FILES={file_count}", flush=True)


STEPS = {"clean": clean, "inspect": inspect, "projection": projection, "pushdown": pushdown,
         "partition": partition, "append": append, "options": options, "small-files": small_files}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--step", choices=[*STEPS, "all"], default="all")
    parser.add_argument("--wait-seconds", type=int, default=0)
    args = parser.parse_args()
    if args.wait_seconds < 0:
        parser.error("--wait-seconds must be zero or positive")
    spark = SparkSession.builder.appName("Week06-Lab-" + args.step).getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    spark.conf.set("spark.sql.session.timeZone", "Asia/Seoul")
    try:
        for name in STEPS if args.step == "all" else [args.step]:
            print("STEP=" + name, flush=True)
            STEPS[name](spark)
            print("STEP_CHECK=" + name + ",PASS", flush=True)
        print("CHECK=PASS", flush=True)
        print("UI observation: " + str(args.wait_seconds) + " seconds", flush=True)
        time.sleep(args.wait_seconds)
    finally:
        spark.stop()
        print("APP_STOPPED", flush=True)


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
set -euo pipefail
exec /opt/spark/bin/spark-submit \
  --master spark://spark-master:7077 \
  --deploy-mode client \
  --driver-memory 512m \
  --executor-memory 512m \
  --executor-cores 1 \
  --total-executor-cores 2 \
  --conf spark.dynamicAllocation.enabled=false \
  --conf spark.driver.host=spark-client \
  --conf spark.driver.bindAddress=0.0.0.0 \
  --conf spark.driver.port=39000 \
  --conf spark.blockManager.port=39001 \
  /opt/spark/apps/week06_lab.py "$@"

# DataFrame과 SQL을 사용할 SparkSession 클래스
from pyspark.sql import SparkSession
# 컬럼 계산과 집계에 사용할 내장 함수
from pyspark.sql import functions as F
# CSV 컬럼의 자료형을 지정할 클래스
from pyspark.sql.types import (
    DecimalType, IntegerType, LongType, StringType,
    StructField, StructType, TimestampType,
)

# Master 주소 등 실행 옵션은 기존처럼 spark-submit에서 지정
spark = SparkSession.builder.appName("Week06-Parquet").getOrCreate()
# 출력에서 INFO 로그를 줄이고 경고·오류는 유지
spark.sparkContext.setLogLevel("WARN")
# 날짜 변환의 기준 시간대를 명시
spark.conf.set("spark.sql.session.timeZone", "Asia/Seoul")

# CSV의 컬럼 이름·자료형 지정; NULL과 잘못된 값은 아래 품질 조건에서 검사
order_schema = StructType([
    StructField("order_id", LongType(), True),
    StructField("user_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("price", DecimalType(12, 2), True),
    StructField("event_time", TimestampType(), True),
])
# 헤더와 시각 형식을 지정하고 CSV를 읽을 DataFrame 정의
orders = (
    spark.read
    .option("header", True)
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
    .schema(order_schema)
    .csv("/opt/spark/data/orders.csv")
)


# 수량·가격이 있고, 수량이 양수이며 가격이 음수가 아닌 주문만 선택
valid_orders = orders.filter(
    F.col("quantity").isNotNull()
    & F.col("price").isNotNull()
    & (F.col("quantity") > 0)
    & (F.col("price") >= 0)
)
# 주문 금액을 계산한 컬럼 추가; 원본 orders는 변경하지 않음
clean_orders = valid_orders.withColumn(
    "amount", F.col("quantity") * F.col("price")
)
# 정제 결과의 주문 번호·수량·가격·금액을 실제로 계산해 출력
clean_orders.select(
    "order_id", "user_id", "quantity", "price", "amount"
).orderBy("order_id").show(truncate=False)


# 이번 주 예제 전용 출력 디렉터리
clean_path = "/opt/spark/output/week06/orders_clean"
# 정상 주문과 계산한 amount를 Snappy 압축 Parquet으로 저장
(
    clean_orders.write
    # 해당 출력 경로에 기존 결과가 있으면 교체; 원본 CSV는 대상이 아님
    .mode("overwrite")
    # 사용할 압축 방식 지정
    .option("compression", "snappy")
    # Action: 실제 입력 읽기·정제·금액 계산·파일 기록 수행
    .parquet(clean_path)
)


# 예상한 저장 경로 충돌 오류만 처리할 PySpark 예외 클래스
from pyspark.errors import AnalysisException

# 저장 모드 실험 전용 경로
mode_path = "/opt/spark/output/week06/mode_demo"
# 주문 1 한 건만 먼저 저장
clean_orders.filter(F.col("order_id") == 1).write.mode("overwrite").parquet(mode_path)
print("INITIAL_ROWS=", spark.read.parquet(mode_path).count(), sep="")
# 9건을 저장하려 해도 기존 경로가 있으므로 저장 생략; 기존 1건 유지
clean_orders.write.mode("ignore").parquet(mode_path)
print("AFTER_IGNORE=", spark.read.parquet(mode_path).count(), sep="")
try:
    # 기존 경로가 있으므로 오류 발생; 이 오류를 아래 except에서 확인
    clean_orders.write.mode("error").parquet(mode_path)
except AnalysisException as error:
    # Spark가 제공하는 오류 이름 출력
    print("ERROR_MODE=", error.getCondition(), sep="")
# 기존 1건에 정상 주문 9건을 추가하므로 10건; 주문 1은 중복
clean_orders.write.mode("append").parquet(mode_path)
print("AFTER_APPEND=", spark.read.parquet(mode_path).count(), sep="")
# 기존 10건을 정상 주문 9건으로 교체
clean_orders.write.mode("overwrite").parquet(mode_path)
print("AFTER_OVERWRITE=", spark.read.parquet(mode_path).count(), sep="")


# 파일 이름을 확인하기 위한 Python의 경로 클래스
from pathlib import Path

# 기본 저장 결과와 구분한 option 실험 경로
option_path = "/opt/spark/output/week06/option_demo"
(
    clean_orders.write.mode("overwrite")
    # Gzip 압축 사용
    .option("compression", "gzip")
    # 파일 하나의 행 수를 최대 4건으로 제한
    .option("maxRecordsPerFile", 4)
    .parquet(option_path)
)
# glob은 이름 규칙에 맞는 파일들을 찾는 Python 메서드
option_files = sorted(Path(option_path).glob("part-*.parquet"))
# 생성한 파일 이름 출력; UUID 등은 실행마다 달라짐
for file in option_files:
    print(file.name)
# 파일마다 실제 행 수를 읽어 출력
print("ROWS_PER_FILE=", [spark.read.parquet(str(file)).count() for file in option_files], sep="")


# iterdir는 해당 폴더 안의 항목 목록을 반환; 앞에서 Path를 가져옴
for entry in sorted(Path(clean_path).iterdir()):
    # 점으로 시작하는 파일시스템 보조 파일은 출력에서 생략
    if not entry.name.startswith("."):
        print(entry.name)


# 파일에 저장된 Schema 정보를 이용해 DataFrame 생성
parquet_orders = spark.read.parquet(clean_path)
# 컬럼 이름과 자료형 확인; 입력 CSV의 Schema 추론은 필요하지 않음
parquet_orders.printSchema()
# 값 비교를 위해 주문 식별자로 정렬해 출력
parquet_orders.orderBy("order_id").show(truncate=False)


# 작은 예제 9행을 비교하기 위해 컬럼과 행 순서를 고정
compare_columns = clean_orders.columns
# 저장 전의 정제 결과 계산
before_rows = clean_orders.select(*compare_columns).orderBy("order_id").collect()
# 저장한 Parquet에서 같은 컬럼과 순서로 조회
after_rows = parquet_orders.select(*compare_columns).orderBy("order_id").collect()
# 저장 전후 값과 정상 주문 수 확인
# assert: 두 List가 다르면 AssertionError; 같으면 다음 줄로 진행
assert before_rows == after_rows
# len은 수집한 Python List의 원소 수; 9가 아니면 오류
assert len(after_rows) == 9
# 실제 정상 주문 식별자도 확인
assert [row["order_id"] for row in after_rows] == [1, 2, 3, 5, 7, 8, 10, 11, 12]
# assert가 자체 출력하지 않으므로 통과 여부를 명시적으로 출력
print("ROUND_TRIP=PASS; ROWS=", len(after_rows), sep="")


# 일곱 컬럼 중 사용자와 주문 금액만 필요한 조회 정의
projected = parquet_orders.select("user_id", "amount")
# 실행계획 출력; 이 호출 자체는 결과 행을 수집하지 않음
projected.explain(mode="formatted")
# Action 호출: 계획한 두 컬럼을 실제로 읽어 주문 금액 결과 출력
projected.orderBy("user_id", "amount").show(truncate=False)


# 필터에 amount를 사용하고 필요한 세 컬럼을 선택
high_orders = (
    parquet_orders.filter(F.col("amount") >= 30000)
    .select("order_id", "user_id", "amount")
)
# 정렬을 추가하기 전 파일 Scan의 조건 전달 여부 확인
high_orders.explain(mode="formatted")
# 같은 조건을 SQL WHERE로 작성
# 먼저 이 DataFrame을 SQL에서 사용할 이름으로 등록
parquet_orders.createOrReplaceTempView("orders_parquet")
sql_high_orders = spark.sql("""
    SELECT order_id, user_id, amount
    FROM orders_parquet
    WHERE amount >= 30000
""")
# 비교할 때만 순서를 명시; 각 조회 결과는 세 행
assert (
    high_orders.orderBy("order_id").collect()
    == sql_high_orders.orderBy("order_id").collect()
)
# 조건을 만족하는 주문을 실제로 출력
high_orders.orderBy("order_id").show(truncate=False)


# Python에서 날짜 값과 정확한 십진수 금액을 만들 클래스
from datetime import date
from decimal import Decimal

# 이 장의 저장 구조를 관찰할 독립적인 4행 DataFrame 생성
partition_input = spark.createDataFrame(
    [
        (101, "U001", Decimal("24000.00"), date(2026, 9, 1)),
        (102, "U002", Decimal("35000.00"), date(2026, 9, 1)),
        (103, "U003", Decimal("20000.00"), date(2026, 9, 2)),
        (104, "U001", Decimal("35000.00"), date(2026, 9, 2)),
    ],
    # 각 컬럼의 이름과 자료형을 지정하는 Schema 문자열
    "order_id long, user_id string, amount decimal(12,2), event_date date",
)
# 저장 전 입력 컬럼과 네 행 출력
print("INPUT_COLUMNS=", partition_input.columns, sep="")
partition_input.orderBy("order_id").show(truncate=False)


# 6장의 기본 저장 결과와 구분한 새 출력 경로
partitioned_path = "/opt/spark/output/week06/partition_demo"
(
    partition_input.write.mode("overwrite")
    .option("compression", "snappy")
    # event_date 값이 같은 행들을 같은 이름의 날짜 디렉터리에 저장
    .partitionBy("event_date")
    .parquet(partitioned_path)
)
# rglob은 하위 폴더도 탐색; relative_to는 앞의 partition_demo 경로를 빼고 표시
for file in sorted(Path(partitioned_path).rglob("*.parquet")):
    print(file.relative_to(partitioned_path))


# 9월 1일 디렉터리의 데이터 파일 하나 선택; file은 Python Path 객체
one_file = sorted(
    (Path(partitioned_path) / "event_date=2026-09-01").glob("*.parquet")
)[0]
# 파일 경로만 지정: 그 파일 내부의 세 컬럼 읽기
single_file_orders = spark.read.parquet(str(one_file))
print("SINGLE_FILE_COLUMNS=", single_file_orders.columns, sep="")
# 데이터셋 최상위 경로 지정: 하위 파일들과 날짜 디렉터리 이름 함께 해석
partitioned_orders = spark.read.parquet(partitioned_path)
print("DATASET_COLUMNS=", partitioned_orders.columns, sep="")
# 경로에서 얻은 날짜를 포함해 네 행 출력
partitioned_orders.orderBy("order_id").show(truncate=False)
# 입력 DataFrame의 컬럼도 그대로인지 확인
print("INPUT_STILL_HAS_DATE=", "event_date" in partition_input.columns, sep="")


# F.col은 기존 컬럼 참조, F.lit은 코드에 적은 값을 각 행에서 그대로 사용
literal_demo = partitioned_orders.select(
    "order_id",
    F.col("event_date"),
    # 문자열을 전달했으므로 결과 자료형도 string
    F.lit("2026-09-01").alias("fixed_text"),
    # 정수를 전달했으므로 결과는 integer 값 1
    F.lit(1).alias("fixed_number"),
)
# 실제 날짜는 행마다 다르지만 fixed_text와 fixed_number는 모든 행에서 같음
literal_demo.orderBy("order_id").show(truncate=False)
print("LITERAL_TYPES=", literal_demo.dtypes, sep="")


# 날짜 컬럼 조건과 금액 컬럼 조건을 함께 적용
daily_high_orders = (
    partitioned_orders.filter(
        # 문자열 값을 날짜로 변환해 event_date와 비교
        (F.col("event_date") == F.to_date(F.lit("2026-09-01")))
        & (F.col("amount") >= 30000)
    )
    .select("order_id", "user_id", "amount")
)
# 파일 읽기 연산과 그 연산에 적용된 조건을 출력
daily_high_orders.explain(mode="formatted")
daily_high_orders.orderBy("order_id").show(truncate=False)
# 5주차와 같은 방식으로 SQL 조회 이름 등록
partitioned_orders.createOrReplaceTempView("partition_orders")
# 같은 날짜·금액 조건을 SQL로 작성
sql_daily_high_orders = spark.sql("""
    SELECT order_id, user_id, amount
    FROM partition_orders
    WHERE event_date = DATE '2026-09-01'
      AND amount >= 30000
""")
# 두 조회의 정렬된 결과가 다르면 AssertionError
assert (
    daily_high_orders.orderBy("order_id").collect()
    == sql_daily_high_orders.orderBy("order_id").collect()
)


# 비교 연산 오른쪽의 문자열은 비교할 값; 컬럼 이름으로 찾지 않음
without_lit = (
    partitioned_orders.filter(
        (F.col("event_date") == "2026-09-01")
        & (F.col("amount") >= 30000)
    )
    .select("order_id", "user_id", "amount")
)
# lit·to_date를 명시한 코드와 같은 주문 102인지 확인
assert without_lit.collect() == daily_high_orders.collect()
without_lit.show(truncate=False)
print("WITHOUT_LIT=PASS", flush=True)


# append의 파일 생성 동작을 확인할 전용 출력 경로
append_path = "/opt/spark/output/week06/append_files_demo"
# 초기 정상 주문 9건 저장
clean_orders.write.mode("overwrite").parquet(append_path)
# 추가 저장 전의 데이터 파일 목록 보관
files_before_append = set(Path(append_path).glob("part-*.parquet"))
print("BEFORE_FILES=", len(files_before_append), sep="")
print("BEFORE_ROWS=", spark.read.parquet(append_path).count(), sep="")
# 주문 1 한 건을 담은 DataFrame을 추가 저장
clean_orders.filter(F.col("order_id") == 1).write.mode("append").parquet(append_path)
files_after_append = set(Path(append_path).glob("part-*.parquet"))
print("AFTER_FILES=", len(files_after_append), sep="")
print("AFTER_ROWS=", spark.read.parquet(append_path).count(), sep="")
# issubset은 원래 파일 이름들이 추가 저장 후 목록에도 모두 있는지 검사
print("ORIGINAL_FILES_PRESENT=", files_before_append.issubset(files_after_append), sep="")


# 주문 식별자마다 디렉터리를 만드는 반례용 경로
# 이 대입은 경로 문자열만 지정; 디렉터리는 아래 parquet 호출에서 생성
tiny_path = "/opt/spark/output/week06/tiny_files_demo"
(
    clean_orders.write.mode("overwrite")
    # 주문 식별자는 9개 모두 달라 디렉터리도 9개 생성
    .partitionBy("order_id")
    .parquet(tiny_path)
)
# 주문 번호별 디렉터리 수와 그 아래 Parquet 파일 수 출력
print("ORDER_DIRECTORIES=", len(list(Path(tiny_path).glob("order_id=*"))), sep="")
print("PARQUET_FILES=", len(list(Path(tiny_path).rglob("*.parquet"))), sep="")


# 예상 금액을 십진수로 작성할 Python 클래스
from decimal import Decimal

# 원본과 각 저장 결과의 전체 행 수 확인
assert orders.count() == 12
assert parquet_orders.count() == 9
assert partitioned_orders.count() == 4
# 4·5주차의 집계 연산으로 저장된 정상 주문의 사용자별 결과를 검사
df_summary_rows = (
    parquet_orders.groupBy("user_id")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("amount").alias("total_amount"),
    )
    .orderBy(F.desc("total_amount"), F.asc("user_id"))
    .collect()
)
# 저장 결과를 예상한 사용자별 집계값과 비교
assert [tuple(row) for row in df_summary_rows] == [
    ("U001", 4, Decimal("108000.00")),
    ("U002", 3, Decimal("89000.00")),
    ("U003", 2, Decimal("35000.00")),
]
# 10장의 독립적인 4행 예제가 날짜별 저장 후에도 같은지 확인
assert (
    partitioned_orders.select(*partition_input.columns).orderBy("order_id").collect()
    == partition_input.orderBy("order_id").collect()
)
# 금액 조건에 맞는 주문 식별자 확인
assert [
    row["order_id"] for row in high_orders.orderBy("order_id").collect()
] == [2, 5, 8]
# 10장의 날짜·금액 조건을 모두 만족하는 주문은 102
assert [row["order_id"] for row in daily_high_orders.collect()] == [102]
# 날짜 컬럼의 중복을 제거하고 정렬해 두 날짜 확인
assert [
    str(row["event_date"])
    for row in partitioned_orders.select("event_date").distinct().orderBy("event_date").collect()
] == ["2026-09-01", "2026-09-02"]
# 모든 검사가 끝났다는 표시
print("CHECK=PASS", flush=True)
# Spark 자원 정리와 Application 종료
spark.stop()

import hashlib
from pathlib import Path

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (
    DecimalType, IntegerType, LongType, StringType,
    StructField, StructType, TimestampType,
)

DATA = Path("/opt/spark/data")
OUTPUT = Path("/opt/spark/output/exercises")
SALES_SCHEMA = StructType([
    StructField("sale_id", LongType(), True),
    StructField("branch", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(12, 2), True),
    StructField("sold_at", TimestampType(), True),
])
INVENTORY_SCHEMA = StructType([
    StructField("event_id", LongType(), True),
    StructField("warehouse", StringType(), True),
    StructField("delta_quantity", IntegerType(), True),
    StructField("event_time", TimestampType(), True),
])


def session(name):
    spark = SparkSession.builder.appName(name).getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    spark.conf.set("spark.sql.session.timeZone", "Asia/Seoul")
    return spark


def read_csv(spark, name, schema):
    return (spark.read.option("header", True)
            .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
            .schema(schema).csv(str(DATA / name)))


def snapshot(path):
    return {str(p.relative_to(path)): hashlib.sha256(p.read_bytes()).hexdigest()
            for p in path.rglob("*.parquet")}


def run_sales(prepare, write_clean, query_saved):
    spark = session("Week06-Exercise01")
    try:
        raw = read_csv(spark, "store_sales.csv", SALES_SCHEMA)
        clean = prepare(raw)
        expected_ids = [301, 302, 303, 305, 307, 308, 310, 311, 312]
        assert raw.count() == 12
        assert clean.columns == [*raw.columns, "amount", "sale_date"], "Keep the five input columns, then amount and sale_date"
        assert [r.sale_id for r in clean.orderBy("sale_id").collect()] == expected_ids
        assert clean.schema["amount"].dataType.simpleString() == "decimal(23,2)"
        assert clean.schema["sale_date"].dataType.simpleString() == "date"
        path = OUTPUT / "ex01"
        # Repeat the write to check that rerunning replaces, not duplicates, data.
        for _ in range(2):
            write_clean(clean, str(path))
        saved = spark.read.parquet(str(path))
        assert saved.select(*clean.columns).orderBy("sale_id").collect() == clean.orderBy("sale_id").collect()
        directories = sorted(p.name for p in path.glob("sale_date=*") if p.is_dir())
        assert directories == ["sale_date=2026-10-01", "sale_date=2026-10-02"]
        files = sorted(path.rglob("*.parquet"))
        assert files and all(".snappy.parquet" in p.name for p in files), "Use Snappy compression"
        file_columns = spark.read.parquet(str(files[0])).columns
        assert len(file_columns) == 6 and "sale_date" not in file_columns
        result = query_saved(saved)
        assert result.columns == ["sale_id", "branch", "amount"]
        rows = result.collect()
        assert [(r.sale_id, r.branch, str(r.amount)) for r in rows] == [
            (307, "Seoul", "12000.00"), (310, "Incheon", "12000.00"), (311, "Seoul", "12000.00")]
        assert int(saved.agg(F.sum("amount")).first()[0]) == 86000
        saved.orderBy("sale_id").show(truncate=False)
        result.show(truncate=False)
        print("RAW_ROWS=12\nSAVED_ROWS=9\nDATE_DIRECTORIES=2", flush=True)
        print("SINGLE_FILE_COLUMNS=" + ",".join(file_columns), flush=True)
        print("DATASET_COLUMNS=" + ",".join(saved.columns), flush=True)
        print("RESULT_IDS=307,310,311\nTOTAL_AMOUNT=86000.00", flush=True)
        print("CHECK=PASS", flush=True)
    finally:
        spark.stop()
        print("APP_STOPPED", flush=True)


def run_inventory(prepare, write_initial, append_new, summarize):
    spark = session("Week06-Exercise02")
    try:
        initial = prepare(read_csv(spark, "inventory_initial.csv", INVENTORY_SCHEMA))
        incoming = prepare(read_csv(spark, "inventory_incoming.csv", INVENTORY_SCHEMA))
        assert initial.count() == 4 and incoming.count() == 3
        assert initial.columns == incoming.columns == INVENTORY_SCHEMA.fieldNames()
        assert [r.event_id for r in initial.orderBy("event_id").collect()] == [401, 402, 403, 406]
        assert [r.event_id for r in incoming.orderBy("event_id").collect()] == [402, 407, 408]
        path = OUTPUT / "ex02"
        # Each run must reset the initial data, then append exactly one batch.
        for attempt in (1, 2):
            write_initial(initial, str(path))
            before = snapshot(path)
            assert spark.read.parquet(str(path)).count() == 4
            append_new(incoming, str(path))
            after = snapshot(path)
            saved = spark.read.parquet(str(path))
            assert saved.count() == 7
            assert saved.filter(F.col("event_id") == 402).count() == 2
            assert len(after) > len(before), "Append must add data files"
            assert all(after.get(name) == digest for name, digest in before.items()), "Existing files must remain unchanged"
            print(f"RUN={attempt}\nBEFORE_FILES={len(before)}\nAFTER_FILES={len(after)}", flush=True)
            print("BEFORE_ROWS=4\nAFTER_ROWS=7\nEVENT_402_COUNT=2", flush=True)
        result = summarize(saved)
        assert result.columns == ["warehouse", "event_count", "net_quantity"]
        assert [tuple(r) for r in result.collect()] == [("A", 3, 13), ("B", 4, 17)]
        result.show(truncate=False)
        print("ORIGINAL_FILES_UNCHANGED=True\nRERUN_ROWS=7\nCHECK=PASS", flush=True)
    finally:
        spark.stop()
        print("APP_STOPPED", flush=True)

#!/usr/bin/env bash
set -euo pipefail
case "${1:-}" in
  exercise01_store_sales.py|exercise02_inventory_append.py) app="$1"; shift ;;
  *) printf '%s\n' 'Specify exercise01_store_sales.py or exercise02_inventory_append.py' >&2; exit 2 ;;
esac
exec /opt/spark/bin/spark-submit \
  --master spark://spark-master:7077 \
  --deploy-mode client \
  --driver-memory 512m \
  --executor-memory 512m \
  --executor-cores 1 \
  --total-executor-cores 2 \
  --conf spark.dynamicAllocation.enabled=false \
  --conf spark.driver.host=spark-client \
  --conf spark.driver.bindAddress=0.0.0.0 \
  --conf spark.driver.port=39000 \
  --conf spark.blockManager.port=39001 \
  "/opt/spark/apps/$app" "$@"

from pyspark.sql import functions as F
from exercise_support import run_sales


def prepare(raw):
    # 수량·단가가 NULL인 행, 수량이 0 이하인 행, 단가가 음수인 행 제외
    # 원본 다섯 컬럼 뒤에 amount와 sale_date 컬럼을 차례로 추가
    raise NotImplementedError("prepare 함수를 작성하세요")


def write_clean(clean, path):
    # sale_date별 디렉터리로 구분하여 Snappy Parquet 저장, 재실행 시 덮어쓰기
    raise NotImplementedError("write_clean 함수를 작성하세요")


def query_saved(saved):
    # 2026-10-02 중 amount >= 10000인 행: sale_id, branch, amount 순서로 반환
    # sale_id 오름차순 정렬. DataFrame API 또는 SQL 사용
    raise NotImplementedError("query_saved 함수를 작성하세요")


if __name__ == "__main__":
    run_sales(prepare, write_clean, query_saved)

from pyspark.sql import functions as F
from exercise_support import run_inventory


def prepare(raw):
    # delta_quantity가 NULL이거나 0인 행 제외. 음수는 출고이므로 유지
    raise NotImplementedError("prepare 함수를 작성하세요")


def write_initial(clean, path):
    # 최초 데이터로 결과 디렉터리를 덮어써서 매 실행을 같은 상태에서 시작
    raise NotImplementedError("write_initial 함수를 작성하세요")


def append_new(clean, path):
    # 기존 파일을 유지하면서 신규 데이터를 Parquet으로 추가 저장
    raise NotImplementedError("append_new 함수를 작성하세요")


def summarize(saved):
    # warehouse별 event_count(행 수), net_quantity(수량 합계) 계산
    # warehouse, event_count, net_quantity 컬럼을 warehouse 오름차순으로 반환
    raise NotImplementedError("summarize 함수를 작성하세요")


if __name__ == "__main__":
    run_inventory(prepare, write_initial, append_new, summarize)