보충 실습 파일과 실행 안내

# Independently authored public lab. Run commands from this directory.
name: knut-public-spark

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

services:
  spark-master:
    <<: *runtime
    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:28080: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: 15s

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

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

"""Small, synthetic CSV pipeline. Python standard library only."""
import argparse
import csv
from datetime import date
from pathlib import Path
import tempfile

ROOT = Path(__file__).resolve().parents[1]
FIELDS = ["order_id", "order_date", "amount"]


def clean_orders(source):
    unique = {}
    input_rows = 0
    with source.open(encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream)
        if reader.fieldnames != FIELDS:
            raise ValueError("CSV header must be order_id,order_date,amount")
        for line, row in enumerate(reader, start=2):
            input_rows += 1
            if None in row or any(row[key] is None for key in FIELDS):
                raise ValueError(f"line {line}: expected three columns")
            order_id = row["order_id"].strip()
            if not order_id:
                raise ValueError(f"line {line}: empty order_id")
            order_date = date.fromisoformat(row["order_date"].strip()).isoformat()
            amount = int(row["amount"])
            if amount < 0:
                raise ValueError(f"line {line}: amount must be nonnegative")
            record = (order_date, amount)
            if order_id in unique and unique[order_id] != record:
                raise ValueError(f"line {line}: conflicting order_id {order_id}")
            unique[order_id] = record
    return unique, input_rows


def run(source, output):
    if source.resolve() == output.resolve():
        raise ValueError("input and output must be different files")
    orders, input_rows = clean_orders(source)
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="", dir=output.parent,
            prefix="orders-", suffix=".tmp", delete=False,
        ) as stream:
            temporary = Path(stream.name)
            writer = csv.writer(stream, lineterminator="\n")
            writer.writerow(FIELDS)
            for order_id, (order_date, amount) in sorted(orders.items()):
                writer.writerow([order_id, order_date, amount])
        temporary.replace(output)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    print(f"INPUT_ROWS={input_rows}")
    print(f"UNIQUE_ORDERS={len(orders)}")
    print(f"DUPLICATE_ROWS={input_rows - len(orders)}")
    print(f"TOTAL_AMOUNT={sum(amount for _, amount in orders.values())}")
    print(f"OUTPUT={output}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", type=Path, default=ROOT / "data/orders.csv")
    parser.add_argument("--output", type=Path, default=ROOT / "output/orders-clean.csv")
    args = parser.parse_args()
    try:
        run(args.input, args.output)
    except (ValueError, OSError) as error:
        parser.exit(1, f"ERROR: {error}\n")


if __name__ == "__main__":
    main()

"""Wait for the requested number of live one-core workers; no PySpark needed."""
import argparse
import json
import time
from urllib.error import URLError
from urllib.request import urlopen


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--workers", type=int, required=True)
    parser.add_argument("--timeout", type=int, default=90)
    args = parser.parse_args()
    if args.workers < 1 or args.timeout < 1:
        parser.error("workers and timeout must be positive")
    deadline = time.monotonic() + args.timeout
    observed = "master not reached"
    while time.monotonic() < deadline:
        try:
            with urlopen("http://spark-master:8080/json/", timeout=3) as response:
                status = json.load(response)
            live = [worker for worker in status["workers"] if worker["state"] == "ALIVE"]
            cores = sum(int(worker["cores"]) for worker in live)
            observed = f"ALIVE_WORKERS={len(live)} TOTAL_CORES={cores}"
            if len(live) == args.workers and cores == args.workers:
                print(observed)
                print("REGISTRATION_CHECK=PASS")
                return
        except (URLError, TimeoutError, OSError, ValueError, KeyError) as error:
            observed = str(error)
        time.sleep(1)
    parser.exit(1, f"REGISTRATION_CHECK=FAIL: {observed}\n")


if __name__ == "__main__":
    main()

"""Verify a known answer and leave time to inspect the Spark UI."""
import argparse
import time


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--wait-seconds", type=int, default=0)
    args = parser.parse_args()
    if args.wait_seconds < 0:
        parser.error("wait-seconds must be nonnegative")
    from pyspark.sql import SparkSession, functions as F

    spark = SparkSession.builder.appName("Public-First-Sum").getOrCreate()
    try:
        spark.sparkContext.setLogLevel("WARN")
        print(f"MASTER={spark.sparkContext.master}", flush=True)
        print(f"APP_ID={spark.sparkContext.applicationId}", flush=True)
        numbers = spark.range(1, 101, 1, 4)
        numbers.printSchema()
        result = numbers.agg(F.sum("id").alias("total"))
        result.explain()
        total = result.first()["total"]
        if total != 5050:
            raise ValueError(f"expected 5050, received {total}")
        print(f"SUM_1_TO_100={total}", flush=True)
        print("CHECK=PASS", flush=True)
        print(f"UI_WAIT_SECONDS={args.wait_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

app="${1:-sum_app.py}"
if (( $# > 0 )); then shift; fi
case "$app" in
  sum_app.py|benchmark_app.py) ;;
  *) echo "Choose sum_app.py or benchmark_app.py" >&2; exit 2 ;;
esac
cores="${TOTAL_EXECUTOR_CORES:-2}"
if [[ ! "$cores" =~ ^[1-9][0-9]*$ ]]; then
  echo "TOTAL_EXECUTOR_CORES must be a positive integer" >&2
  exit 2
fi

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 "$cores" \
  --conf spark.dynamicAllocation.enabled=false \
  --conf spark.sql.adaptive.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" "$@"

"""Count the supplied small text fixture, save Parquet, then read it back."""
import argparse
import csv
from pathlib import Path


def small_counts(frame):
    rows = frame.limit(101).collect()
    if len(rows) > 100:
        raise ValueError("this teaching check supports at most 100 distinct words")
    return {row["word"]: row["count"] for row in rows}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", default="/opt/spark/lab-data/words.txt")
    parser.add_argument("--expected", default="/opt/spark/lab-expected/wordcounts.csv")
    parser.add_argument("--output", default="/opt/spark/lab-output/wordcounts")
    args = parser.parse_args()
    with Path(args.expected).open(encoding="utf-8-sig", newline="") as stream:
        expected = {row["word"]: int(row["count"]) for row in csv.DictReader(stream)}
    from pyspark.sql import SparkSession, functions as F

    spark = SparkSession.builder.appName("Public-Wordcount").getOrCreate()
    try:
        spark.sparkContext.setLogLevel("WARN")
        if not spark.sparkContext.master.startswith("local["):
            raise ValueError("use --master local[2]: files are mounted only in the client")
        lines = spark.read.text(args.input)
        words = lines.select(
            F.explode(F.split(F.col("value"), r"\s+")).alias("word")
        ).where(F.col("word") != "")
        counts = words.groupBy("word").count()
        ordered = counts.orderBy(F.desc("count"), F.asc("word"))
        print("INPUT_LINES")
        lines.show(truncate=False)
        print("WORD_ROWS")
        words.show(truncate=False)
        print("WORD_COUNTS")
        ordered.show(truncate=False)
        ordered.explain()
        if small_counts(ordered) != expected:
            raise ValueError("word counts differ from the expected CSV")
        ordered.write.mode("errorifexists").parquet(args.output)
        restored = spark.read.parquet(args.output)
        if small_counts(restored) != expected:
            raise ValueError("Parquet round-trip check failed")
        print("WORDCOUNT_CHECK=PASS", flush=True)
        print(f"PARQUET_OUTPUT={args.output}", flush=True)
    finally:
        spark.stop()
    print("APP_STOPPED", flush=True)


if __name__ == "__main__":
    main()

"""Measure repeated aggregation actions; timings are produced only by real runs."""
import argparse
import statistics
import time


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--end", type=int, default=1_000_001)
    parser.add_argument("--partitions", type=int, default=128)
    parser.add_argument("--repetitions", type=int, default=3)
    parser.add_argument("--wait-seconds", type=int, default=0)
    args = parser.parse_args()
    if not 2 <= args.end <= 4_294_967_296:
        parser.error("end must be 2..4294967296 (keeps the sum within signed 64-bit)")
    if args.partitions < 1 or args.repetitions < 1 or args.wait_seconds < 0:
        parser.error("partitions/repetitions must be positive; wait must be nonnegative")
    from pyspark.sql import SparkSession, functions as F

    spark = SparkSession.builder.appName("Public-Worker-Benchmark").getOrCreate()
    try:
        spark.sparkContext.setLogLevel("WARN")
        context = spark.sparkContext
        print(f"MASTER={context.master}", flush=True)
        print(f"APP_ID={context.applicationId}", flush=True)
        print(f"REQUESTED_CORES={context.getConf().get('spark.cores.max', 'unset')}", flush=True)
        print(f"END_EXCLUSIVE={args.end}", flush=True)
        print(f"PARTITIONS={args.partitions}", flush=True)
        print(f"REPETITIONS={args.repetitions}", flush=True)
        numbers = spark.range(1, args.end, 1, args.partitions)
        expected = (args.end - 1) * args.end // 2
        durations = []
        for repeat in range(1, args.repetitions + 1):
            started = time.perf_counter()
            total = numbers.agg(F.sum("id").alias("total")).first()["total"]
            elapsed = time.perf_counter() - started
            if total != expected:
                raise ValueError(f"expected {expected}, received {total}")
            durations.append(elapsed)
            print(f"RUN_{repeat}_SECONDS={elapsed:.6f}", flush=True)
        print(f"SUM_RESULT={total}", flush=True)
        print(f"MEDIAN_SECONDS={statistics.median(durations):.6f}", flush=True)
        print("CHECK=PASS", flush=True)
        print(f"UI_WAIT_SECONDS={args.wait_seconds}", flush=True)
        time.sleep(args.wait_seconds)
    finally:
        spark.stop()
    print("APP_STOPPED", flush=True)


if __name__ == "__main__":
    main()

order_id,order_date,amount
A001,2026-09-01,10000
A002,2026-09-01,20000
A002,2026-09-01,20000
A003,2026-09-02,5000
A003,2026-09-02,5000

spark spark
data spark

data pipeline

order_id,order_date,amount
A001,2026-09-01,10000
A002,2026-09-01,20000
A003,2026-09-02,5000

word,count
spark,3
data,2
pipeline,1

condition,workers,requested_cores,end_exclusive,partitions,repetitions,run1_seconds,run2_seconds,run3_seconds,median_seconds,observed_executors,check,notes
A1,1,1,1000000001,128,3,,,,,,,not_measured
A2,2,2,1000000001,128,3,,,,,,,not_measured
A3,4,4,1000000001,128,3,,,,,,,not_measured
A4,8,8,1000000001,128,3,,,,,,,not_measured
B,8,2,1000000001,128,3,,,,,,,not_measured