Apache Spark 기초 개념

# DataFrame 생성 예시
df = spark.createDataFrame([
    ("Alice", 28, "Seoul",  4500000),
    ("Bob",   34, "Busan",  5200000),
    ("Carol", 25, "Daegu",  3800000),
], ["name", "age", "city", "salary"])

# DataFrame 내용 확인
df.show()
# +-----+---+-----+-------+
# | name|age| city| salary|
# +-----+---+-----+-------+
# |Alice| 28|Seoul|4500000|
# |  Bob| 34|Busan|5200000|
# |Carol| 25|Daegu|3800000|
# +-----+---+-----+-------+

# 이 코드들은 아직 실행되지 않습니다 (계획만 세움)
filtered  = df.filter(df.age >= 30)                    # 30세 이상 필터링
selected  = filtered.select("name", "salary")          # 필요한 컬럼만 선택
sorted_df = selected.orderBy("salary", ascending=False)  # 연봉 내림차순 정렬

# Action: 이 시점에 위의 모든 변환이 실행됩니다
sorted_df.show()        # 결과를 화면에 출력 (Action!)
sorted_df.count()       # 건수를 반환 (Action!)
sorted_df.collect()     # 전체 데이터를 Driver로 가져옴 (Action!)
sorted_df.write.save()  # 파일로 저장 (Action!)

# DataFrame을 임시 뷰로 등록
df.createOrReplaceTempView("employees")

# SQL로 분석
result = spark.sql("""
    SELECT
        city,
        COUNT(*)    AS headcount,
        AVG(salary) AS avg_salary
    FROM employees
    GROUP BY city
    ORDER BY avg_salary DESC
""")

result.show()

%sql
SELECT city, COUNT(*) AS headcount, AVG(salary) AS avg_salary
FROM employees
GROUP BY city
ORDER BY avg_salary DESC

from pyspark.sql.functions import col, sum, avg, count, when, upper

# 1. 데이터 읽기 (Delta 테이블에서)
orders = spark.read.table("catalog.schema.orders")

# 2. 필터링: 완료된 주문만
completed = orders.filter(col("status") == "COMPLETED")

# 3. 변환: 새 컬럼 추가 (주문 금액 기준으로 사이즈 라벨링)
enriched = completed.withColumn(
    "order_size",
    when(col("amount") >= 100000, "large")
    .when(col("amount") >=  50000, "medium")
    .otherwise("small")
)

# 4. 집계: 주문 크기별 통계
summary = enriched.groupBy("order_size").agg(
    count("*").alias("order_count"),
    sum("amount").alias("total_revenue"),
    avg("amount").alias("avg_order_value"),
)

# 5. 결과 출력
display(summary)

# 6. 결과를 Delta 테이블로 저장
(summary.write
    .format("delta")
    .mode("overwrite")               # 기존 데이터 덮어쓰기 (append/ignore/errorifexists 와 택1)
    .saveAsTable("catalog.schema.order_summary"))