Spark SQL과 DataFrame 결과 비교

출처: 2026-bigdata-week05-lecture-notes.pdf, 1–4쪽.

4장은 DataFrame API로 주문을 읽고 정제·집계하는 방법을 다룹니다. 5장은 같은 처리를 SQL로 표현하고 값·행 수·정렬 순서가 같은지 확인합니다.

DataFrame의 열 선택·조건식·집계에 대응하는 SQL을 배우고, 두 방식의 결과를 비교합니다. Join·Window Function·CTE는 이 장의 범위에 포함하지 않습니다. 선행 개념과의 연결은 2026-bigdata-04.pdf 88–90쪽과 2026-bigdata-week04-dataframe-notes.md 23절을 참고합니다.

보조 비유: 같은 주문 장부를 함수 호출 목록으로 설명하거나 조회 문장으로 설명하는 차이입니다. 두 표현 모두 Driver에서 실행계획이 되고, Action이 결과를 요구하면 Executor의 Task가 행을 처리합니다. 표현 언어만 바뀌었다고 계산이 Python 반복문으로 옮겨지는 것은 아닙니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 2–10쪽.

4장의 4행 표는 개념을 설명하는 작은 예시입니다. 이 장은 원본 3쪽과 제공 실습 orders.csv의 주문 12행을 사용합니다. 수량·가격 조건에 따라 주문 4·6·9를 제외하면 정상 주문은 9건입니다. 아래 표와 수치는 이 입력을 정제·집계한 예상 결과입니다.

order_id,user_id,product_id,quantity,price,event_time
1,U001,P001,2,12000.00,2026-09-01 09:00:00
2,U002,P002,1,35000.00,2026-09-01 09:05:00
3,U001,P003,3,8000.00,2026-09-01 09:10:00
4,U003,P001,0,12000.00,2026-09-01 09:15:00
5,U002,P004,2,15000.00,2026-09-01 09:20:00
6,U004,P002,1,-100.00,2026-09-01 09:25:00
7,U003,P005,4,5000.00,2026-09-01 09:30:00
8,U001,P002,1,35000.00,2026-09-01 09:35:00
9,U004,P003,,8000.00,2026-09-01 09:40:00
10,U002,P001,2,12000.00,2026-09-01 09:45:00
11,U003,P004,1,15000.00,2026-09-01 09:50:00
12,U001,P005,5,5000.00,2026-09-01 09:55:00

아래 코드는 강의 해설용 발췌입니다. Spark 4.1.3 환경에서 Python 프로그램 안에 순서대로 작성하며, 파일 경로는 컨테이너 내부 경로입니다. 입력 CSV가 Client와 모든 Worker의 같은 경로에 있어야 합니다.

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (
    StructType, StructField, LongType, StringType,
    IntegerType, DecimalType, TimestampType,
)
spark = SparkSession.builder.appName("Orders-SQL-Comparison").getOrCreate()
order_schema = StructType([
    StructField("order_id", LongType(), False),
    StructField("user_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), True),
    StructField("price", DecimalType(12, 2), True),
    StructField("event_time", TimestampType(), True),
])
orders = (spark.read.option("header", True)
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
    .schema(order_schema).csv("/opt/spark/data/orders.csv"))
orders.printSchema()

nullable=False는 데이터베이스의 NOT NULL 저장 제약과 적용 시점이 다릅니다. Python List로 createDataFrame을 호출하면 기본 verifySchema=True에서 NULL 불가 필드의 None을 거부합니다. 반면 원본의 CSV 비교 예에서는 읽은 Schema의 nullable이 True가 되고 누락 수량이 남습니다. 선언만 믿지 말고 실제 Schema와 NULL 행을 검사합니다.

required_id = StructType([StructField("order_id", LongType(), False)])
try:
    spark.createDataFrame([(None,)], required_id)
except ValueError:
    print("NULL_REJECTED")
orders.filter(F.col("quantity").isNull()).select("order_id").show()

출처: 2026-bigdata-week05-lecture-notes.pdf, 11–14쪽.

SQL의 FROM은 Python 변수 이름을 직접 찾지 않습니다. Temporary View는 DataFrame을 SQL에서 조회할 이름으로 등록한 임시 뷰입니다. 보조 비유로는 장부에 붙인 조회용 이름표이며, 장부 복사나 메모리 적재 명령이 아닙니다. 같은 SparkSession에서 사용하고 프로그램을 다시 시작하면 다시 등록합니다.

orders.createOrReplaceTempView("orders_view")
first_result = spark.sql("""
    SELECT order_id, user_id, quantity
    FROM orders_view
    ORDER BY order_id
    LIMIT 4
""")
first_result.show()
orders.createOrReplaceTempView("orders")

orders는 Python 변수, orders_view는 등록한 SQL 이름, orders.csv는 입력 파일입니다. 등록하지 않은 이름을 조회하면 이 예제에서는 TABLE_OR_VIEW_NOT_FOUND 오류가 납니다. 이후 코드는 orders라는 View를 사용합니다. spark.sql의 SELECT 결과도 DataFrame이므로 show·count·collect를 호출할 수 있습니다.

ORDER BY order_id LIMIT 4의 예상 결과는 주문 1·2·3·4입니다. LIMIT만 쓰면 어느 네 행이 선택되는지 보장하지 않습니다. LIMIT 4가 CSV 네 행만 읽는다는 뜻도 아닙니다. 정렬된 앞부분을 찾으려면 더 많은 입력을 읽어야 할 수 있습니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 15–18쪽.

SELECT는 출력할 열과 계산식을, AS는 결과 열 이름을 정합니다. 수량×가격을 계산해도 잘못된 주문은 자동으로 사라지지 않습니다. WHERE에서 수량·가격이 있고 수량은 양수, 가격은 0 이상이라는 조건을 적용합니다.

valid_enriched = spark.sql("""
    SELECT order_id, user_id, product_id, quantity, price, event_time,
           quantity * price AS amount
    FROM orders
    WHERE quantity IS NOT NULL AND price IS NOT NULL
      AND quantity > 0 AND price >= 0
""")
valid_enriched.createOrReplaceTempView("valid_enriched")
valid_enriched.orderBy("order_id").show()

WHERE는 TRUE인 행만 남깁니다. NULL은 0도 빈 문자열도 아니며 quantity = NULL의 비교 결과는 NULL입니다. 수량 누락 주문을 찾으려면 IS NULL을 사용합니다. 12행 입력에서 누락 주문은 9번이고, 정상 주문 식별자는 1·2·3·5·7·8·10·11·12입니다.

missing = spark.sql("SELECT order_id FROM orders WHERE quantity IS NULL")
missing.show()

출처: 2026-bigdata-week05-lecture-notes.pdf, 19–22쪽.

GROUP BY는 같은 사용자 주문을 한 그룹으로 묶습니다. COUNT(*)는 주문 행 수, SUM은 금액 합계, AVG는 평균을 구합니다. 사용자별 한 행에는 그룹 키와 집계값을 둡니다. 서로 다른 주문 번호 네 개를 가진 U001 그룹에서 order_id 하나를 그대로 출력할 수는 없습니다.

sql_summary = spark.sql("""
    SELECT user_id, COUNT(*) AS order_count,
           SUM(amount) AS total_amount, AVG(amount) AS avg_amount
    FROM valid_enriched
    GROUP BY user_id
    ORDER BY total_amount DESC, user_id ASC
""")
sql_summary.show()

표는 12행 입력을 정제한 예상값이며 평균은 소수점 6자리 기준입니다. 예를 들어 U001의 총액은 24000 + 24000 + 35000 + 25000 = 108000입니다. 주문 4건의 평균은 27000입니다. COUNT(*)는 상품 수량의 합계가 아닙니다.

spark.sql("""
    SELECT COUNT(*) AS all_rows, COUNT(quantity) AS quantity_values
    FROM orders
""").show()

필터 전 원본에서 COUNT(*)는 12, COUNT(quantity)는 NULL 수량 하나를 제외한 11입니다. SUM·AVG도 NULL 값을 제외합니다. 값이 모두 NULL이거나 입력이 없으면 SUM·AVG는 NULL이고 COUNT(column)은 0입니다. 모든 열이 NULL인 행도 존재하는 행이므로 COUNT(*)에는 포함됩니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 23–25쪽.

WHERE는 집계 전 주문 행을 고릅니다. HAVING은 집계한 사용자 그룹을 고릅니다. 보조 비유로는 잘못된 영수증을 먼저 버리는 일이 WHERE, 고객별 합계를 낸 후 기준 금액 이상인 고객을 고르는 일이 HAVING입니다.

high_total = spark.sql("""
    SELECT user_id, SUM(amount) AS total_amount
    FROM valid_enriched
    GROUP BY user_id
    HAVING SUM(amount) >= 50000
    ORDER BY total_amount DESC, user_id ASC
""")
df_high_total = (valid_enriched.groupBy("user_id")
    .agg(F.sum("amount").alias("total_amount"))
    .filter(F.col("total_amount") >= 50000)
    .orderBy(F.desc("total_amount"), F.asc("user_id")))

이미 정상 주문만 담긴 valid_enriched를 사용하므로 이 SQL에는 WHERE가 없습니다. 9개 주문을 세 사용자로 집계한 다음 총액 50000 이상인 U001·U002 두 행이 남습니다. 이 설명은 조회문의 의미이며 Spark의 물리 실행 순서를 고정하지 않습니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 26–30쪽.

같은 입력에서 필터·계산식·집계·정렬 기준을 모두 맞춥니다. 두 구현이 같은 조건을 빠뜨리면 서로의 결과는 같아도 틀릴 수 있습니다. 작은 결과에 한해 두 방식의 일치와 독립적으로 계산한 정답을 함께 검사합니다.

df_valid = orders.filter(
    F.col("quantity").isNotNull() & F.col("price").isNotNull()
    & (F.col("quantity") > 0) & (F.col("price") >= 0)
)
df_enriched = df_valid.withColumn("amount", F.col("quantity") * F.col("price"))
df_summary = (df_enriched.groupBy("user_id")
    .agg(F.count("*").alias("order_count"),
         F.sum("amount").alias("total_amount"),
         F.avg("amount").alias("avg_amount"))
    .orderBy(F.desc("total_amount"), F.asc("user_id")))
sql_full_summary = spark.sql("""
    SELECT user_id, COUNT(*) AS order_count,
           SUM(quantity * price) AS total_amount,
           AVG(quantity * price) AS avg_amount
    FROM orders
    WHERE quantity IS NOT NULL AND price IS NOT NULL
      AND quantity > 0 AND price >= 0
    GROUP BY user_id
    ORDER BY total_amount DESC, user_id ASC
""")

from decimal import Decimal
sql_rows = sql_full_summary.collect()
df_rows = df_summary.collect()
assert sql_full_summary.columns == df_summary.columns
assert sql_full_summary.dtypes == df_summary.dtypes
assert sql_rows == df_rows
expected = [
    ("U001", 4, Decimal("108000.00"), Decimal("27000.000000")),
    ("U002", 3, Decimal("89000.00"), Decimal("29666.666667")),
    ("U003", 2, Decimal("35000.00"), Decimal("17500.000000")),
]
actual = [(r["user_id"], r["order_count"], r["total_amount"],
           r["avg_amount"].quantize(Decimal("0.000001"))) for r in sql_rows]
assert actual == expected

금액 정답은 부동소수점 float 대신 Decimal로 표현하고, 평균 비교 자릿수를 맞춥니다. 컬럼 이름·자료형 비교는 원본의 값·순서 비교에 덧붙인 보충 검사입니다. 총액 내림차순과 동률일 때 user_id 오름차순을 양쪽에 동일하게 지정해야 행 순서 비교가 의미를 가집니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 31–35쪽.

COALESCE는 각 행에서 인수를 왼쪽부터 확인해 처음 NULL이 아닌 값을 선택합니다. 여러 행을 더하는 SUM과 다릅니다. 문자열 값은 작은따옴표로, 컬럼 이름은 이 예제처럼 그대로 씁니다. 모든 인수가 NULL이면 결과도 NULL입니다.

spark.sql("""
    SELECT COALESCE('Seoul', 'UNKNOWN') AS with_value,
           COALESCE(NULL, 'UNKNOWN') AS missing_value,
           COALESCE(NULL, NULL) AS all_missing
""").show()
filled = spark.sql("""
    SELECT order_id, COALESCE(quantity, 0) AS quantity
    FROM orders ORDER BY order_id
""")
df_filled = (orders.fillna({"quantity": 0})
    .select("order_id", "quantity").orderBy("order_id"))

첫 조회는 FROM 없이 상수를 계산하며 예상 결과는 한 행의 Seoul·UNKNOWN·NULL입니다. 두 번째 조회는 12행을 유지하고 주문 9의 수량만 NULL에서 0으로 바꿉니다. 원본 파일과 orders는 그대로입니다. 수량 0은 quantity > 0을 만족하지 않으므로 대체했다고 정상 주문이 되지는 않습니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 36–38쪽.

CASE는 조건을 위에서부터 검사해 처음 만족하는 값을 선택합니다. DataFrame의 when·otherwise와 같은 역할입니다. 정상 주문의 금액을 HIGH·MEDIUM·LOW로 분류할 때 큰 금액 조건부터 검사합니다.

classified = spark.sql("""
    SELECT order_id, amount,
           CASE WHEN amount >= 30000 THEN 'HIGH'
                WHEN amount >= 10000 THEN 'MEDIUM'
                ELSE 'LOW' END AS amount_level
    FROM valid_enriched ORDER BY order_id
""")
df_classified = (df_enriched.withColumn("amount_level",
    F.when(F.col("amount") >= 30000, "HIGH")
     .when(F.col("amount") >= 10000, "MEDIUM").otherwise("LOW"))
    .select("order_id", "amount", "amount_level").orderBy("order_id"))

예상 결과는 정상 주문 9행 그대로이며 HIGH는 주문 2·5·8의 3건, MEDIUM은 6건, LOW는 0건입니다. 10000 이상 조건을 먼저 쓰면 35000도 먼저 일치해 MEDIUM이 됩니다. NULL 처리와 달리 이 예제는 이미 정상 주문만 남긴 입력을 사용합니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 39–42쪽.

SELECT 결과도 DataFrame입니다. show는 표를 콘솔에 출력하고, count는 그 DataFrame의 행 수를 Python 정수로 반환하며, collect는 모든 결과 행을 Row 객체의 Python List로 가져옵니다. 사용자별 집계 결과는 3행이지만 전체 주문 수를 구하는 집계 쿼리 결과는 1행입니다.

sql_order_count = spark.sql("SELECT COUNT(*) AS order_count FROM orders")
assert sql_order_count.first()["order_count"] == 12
assert sql_order_count.count() == 1
assert sql_full_summary.count() == 3
null_row = spark.createDataFrame([(None, None)], "a string, b string")
null_row.createOrReplaceTempView("null_row")
spark.sql("""
    SELECT COUNT(*) AS all_rows, COUNT(a) AS a_values, COUNT(b) AS b_values
    FROM null_row
""").show()

모든 열이 NULL인 한 행의 예상 집계값은 1·0·0입니다. 행이 없는 입력과 혼동하지 않습니다. View 등록만으로 결과를 계산하지 않으며 SELECT 결과에 Action을 호출해 확인합니다. collect는 결과 전체를 Driver 메모리로 모으므로 이 장에서는 크기를 아는 3행 집계 결과에만 사용합니다.

출처: 2026-bigdata-week05-lecture-notes.pdf, 43–46쪽.

오류 없이 실행되는 SQL도 요구사항을 놓칠 수 있습니다. WHERE 없이 원본을 집계하면 U001·U002·U003 총액은 우연히 같아도 U004의 -100이 추가되고, 수량 0인 주문 4 때문에 U003 주문 수가 달라집니다. 총액 하나만 확인해서는 이 오류를 잡을 수 없습니다.

-- Incorrect example: missing quality conditions
SELECT user_id, COUNT(*) AS order_count,
       SUM(quantity * price) AS total_amount
FROM orders
GROUP BY user_id
ORDER BY total_amount DESC, user_id ASC

확인 문제: 총액 50000 이상 사용자를 HAVING으로 선택하고, NULL 수량은 IS NULL로 찾아 COALESCE로 0을 지정해 봅니다. CASE의 조건 순서를 뒤집었을 때 어떤 주문이 달라지는지도 설명합니다. 코드가 종료될 때는 spark.stop()으로 세션을 정리합니다. 실제 실행 시에는 자신의 결과와 위 예상 기준을 비교합니다.