주문 데이터로 배우는
DataFrame 프로그래밍

출처: 2026-bigdata-04.pdf, 4–12쪽 · 2026-bigdata-week04-dataframe-notes.md.

주문 한 건을 한 행으로 읽고, 잘못된 수량과 가격을 걸러 낸 뒤 사용자별 구매금액을 계산합니다. 3장의 숫자 한 열 합계와 달리, 여러 열을 가진 표에서 조건에 맞는 행을 골라 집계합니다.

CSV(Comma-Separated Values)는 쉼표로 열을 구분한 텍스트 파일입니다. 첫 줄에는 열 이름을, 이후 줄에는 주문 값을 둡니다. 아래는 강의노트가 제시한 4행 예시이며 전체 실습 데이터가 아닙니다.

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

첫 행은 U001이 P001을 2개 주문한 기록입니다. 주문금액은 2 × 12000.00 = 24000.00입니다. 네 번째 행은 수량이 0이므로 이번 정제 조건을 통과하지 못합니다.

1. 읽기: 열 이름·자료형을 정한 Schema로 CSV를 해석합니다.
2. 정제: 수량과 가격의 NULL·범위를 검사합니다.
3. 변환: 수량과 단가를 곱해 주문금액을 만듭니다.
4. 집계·정렬: 사용자별 주문 수와 금액을 계산하고 순서를 정합니다.
5. 확인: 결과를 요구하는 Action을 호출하고 작은 입력의 예상값과 비교합니다.

출처: 2026-bigdata-04.pdf, 13–20쪽 · 2026-bigdata-week04-dataframe-notes.md.

DataFrame은 이름과 자료형이 있는 열(Column)과 행(Row)으로 구성된 데이터 구조입니다. 큰 표를 나누어 처리할 때는 데이터 조각과 계산 역할을 구분합니다.

• Partition: 나누어 처리하는 데이터 조각입니다.
• Executor: 각 조각을 실제로 계산하는 프로세스입니다.
• Driver: 연산을 계획하고 실행을 요청하는 프로세스입니다.

SparkSession에서 시작하기

SparkSession은 DataFrame과 Spark SQL 기능을 사용하는 시작점입니다. 세션 이름은 실행 화면에서 애플리케이션을 찾는 데 사용합니다.

# SparkSession 클래스를 PySpark SQL 모듈에서 가져옴
from pyspark.sql import SparkSession

# SparkSession을 생성하거나 현재 Process의 기존 SparkSession을 재사용
spark = (
    # SparkSession 설정을 시작하는 Builder
    SparkSession.builder
    # Spark UI에 표시할 Application 이름 지정
    .appName("Week04-DataFrame")
    # 기존 Session을 반환하거나 새 Session 생성
    .getOrCreate()
)

• builder: 세션 설정을 구성합니다.
• appName: 실행 화면에 표시할 이름을 지정합니다.
• getOrCreate: 사용할 수 있는 기존 세션을 얻거나 새 세션을 만듭니다.

강의에서는 Master 주소를 제출 명령의 --master spark://spark-master:7077로 전달합니다. Python 코드와 제출 명령에 서로 다른 Master 주소를 중복 지정하지 않습니다. 이 주소는 2·3장 Compose 네트워크 내부 주소이며 Windows의 파일 경로가 아닙니다.

출처: 2026-bigdata-04.pdf, 21–22, 39–43쪽 · 2026-bigdata-week04-dataframe-notes.md.

CSV에는 값이 문자열로 기록되어 있습니다. Schema(스키마)는 각 열의 이름·자료형·NULL 허용 여부를 정해 같은 파일을 일관되게 해석하도록 합니다. NULL은 값이 없거나 알 수 없다는 뜻입니다.

# 주문 Schema 정의에 필요한 Spark SQL 자료형을 가져옴
from pyspark.sql.types import (
    DecimalType,    # 고정 소수점 숫자
    IntegerType,    # 32비트 정수
    LongType,       # 64비트 정수
    StringType,     # 문자열
    StructField,    # Column 하나의 이름·자료형·NULL 규칙
    StructType,     # 여러 StructField로 구성한 Schema
    TimestampType,  # 날짜와 시각
)

# 주문 CSV의 전체 Schema 정의
order_schema = StructType([
    # 주문 식별자: long, NULL 불가
    StructField("order_id", LongType(), False),
    # 사용자 식별자: string, NULL 불가
    StructField("user_id", StringType(), False),
    # 상품 식별자: string, NULL 불가
    StructField("product_id", StringType(), False),
    # 주문 수량: integer, NULL 허용
    StructField("quantity", IntegerType(), True),
    # 상품 단가: decimal(12,2), NULL 허용
    StructField("price", DecimalType(12, 2), True),
    # 주문 시각: timestamp, NULL 허용
    StructField("event_time", TimestampType(), True),
])

• StructType: 여러 필드 정의를 묶어 전체 표의 구조를 만듭니다.
• StructField: 이름, 자료형, nullable 순으로 한 필드를 정의합니다. 마지막 True는 NULL을 허용한다는 메타데이터입니다.
• LongType·IntegerType: 각각 64비트·32비트 정수입니다. 주문 식별자와 수량의 자료형을 구분합니다.
• TimestampType: 날짜와 시각을 표현합니다. 입력 문자열을 해석할 형식도 확인해야 합니다.

금액을 Decimal로 읽는 이유

DecimalType(12, 2)는 전체 유효 숫자 최대 12자리, 소수점 아래 2자리인 고정 소수점 수입니다. Double의 이진 부동소수점 표현에서는 0.1 + 0.2 같은 계산이 십진수와 정확히 일치하지 않을 수 있습니다. 이 예제는 단가를 소수 둘째 자리까지 표현하기 위해 Decimal을 씁니다.

추론과 명시적 정의

inferSchema=True를 쓰면 Spark가 값을 읽어 자료형을 추론합니다. 처음 받은 파일을 살펴볼 때 편리하지만 데이터를 추가로 읽어야 합니다. 가격에 N/A가 섞이면 문자열로 판단할 수 있습니다. 반복해서 처리할 파일의 형식이 정해져 있다면 Schema를 직접 정의해 열 이름과 자료형을 명시합니다.

보충 근거: Apache Spark CSV 옵션 문서의 inferSchema 추가 읽기 설명.

출처: 2026-bigdata-04.pdf, 28–38쪽 · 2026-bigdata-week04-dataframe-notes.md.

한 주문 안에 고객 정보나 여러 상품이 들어 있다면 평평한 열만으로 구조를 설명하기 어렵습니다. 중첩 Schema는 한 열 안의 하위 필드와 배열·키/값 관계를 표현합니다.

이번 주문 CSV는 평면 구조이고, 아래는 강의노트의 별도 JSON 구조 예시입니다.

root
 |-- order_id: long
 |-- customer: struct
 |    |-- user_id: string
 |    |-- region: string
 |-- items: array
 |    |-- element: struct
 |    |    |-- product_id: string
 |    |    |-- quantity: integer
 |    |    |-- price: decimal(12,2)
 |-- attributes: map
 |    |-- key: string
 |    |-- value: string

중첩 Schema 정의 읽기

Struct는 고객 정보처럼 이름이 있는 필드를 묶고, Array는 여러 상품을 담으며, Map은 추가 메모의 항목과 값을 연결합니다. 아래 전체 정의에서 세 구조를 조합하는 방법을 확인할 수 있습니다.

고객과 상품의 하위 구조를 먼저 정의한 뒤 전체 주문 Schema에 넣습니다. 배열 자체의 NULL 허용 여부와 배열 안 원소의 NULL 허용 여부는 별개입니다.

# 중첩 Schema 정의에 필요한 Spark SQL 자료형을 가져옴
from pyspark.sql.types import (
    ArrayType,    # 같은 자료형의 원소 여러 개를 저장하는 배열
    DecimalType,  # 고정 소수점 숫자
    IntegerType,  # 32비트 정수
    LongType,     # 64비트 정수
    MapType,      # Key와 Value 쌍
    StringType,   # 문자열
    StructField,  # Column 하나의 이름·자료형·NULL 규칙
    StructType,   # 여러 StructField로 구성한 Schema
)

# customer Column 내부의 하위 필드 정의
customer_schema = StructType([
    # user_id는 문자열이고 NULL을 허용하지 않음
    StructField("user_id", StringType(), False),

    # region은 문자열이고 NULL을 허용함
    StructField("region", StringType(), True),
])

# items 배열의 원소 하나가 가질 필드 정의
item_schema = StructType([
    # 상품 식별자, NULL 불가
    StructField("product_id", StringType(), False),

    # 주문 수량, NULL 불가
    StructField("quantity", IntegerType(), False),

    # 상품 가격, NULL 불가
    StructField("price", DecimalType(12, 2), False),
])

# 주문 한 건의 전체 Schema 정의
nested_order_schema = StructType([
    # 주문 식별자, NULL 불가
    StructField("order_id", LongType(), False),

    # customer Column은 customer_schema 구조를 가지며 NULL 불가
    StructField("customer", customer_schema, False),

    # items Column은 item_schema 원소의 배열
    StructField(
        "items",
        # 배열 내부에 NULL 원소를 허용하지 않음
        ArrayType(item_schema, containsNull=False),
        # items 배열 자체도 NULL을 허용하지 않음
        False,
    ),

    # attributes Column은 문자열 Key와 문자열 Value의 Map
    StructField(
        "attributes",
        MapType(
            # Key 자료형
            StringType(),
            # Value 자료형
            StringType(),
            # Map Value에는 NULL을 허용함
            valueContainsNull=True,
        ),
        # attributes Map 자체도 NULL을 허용함
        True,
    ),
])

• containsNull=False: 배열 안에 NULL 원소를 허용하지 않는다는 뜻입니다. 배열 열 자체의 NULL 규칙은 바깥 StructField에서 지정합니다.
• valueContainsNull=True: Map의 값이 NULL일 수 있다는 뜻입니다.

배열을 행으로 펼치기

explode()는 배열 원소 하나를 결과 행 하나로 펼칩니다. 상품 두 개가 들어 있는 주문은 두 행이 됩니다.

빈 배열이나 NULL 배열은 결과 행을 만들지 않으므로 그 주문 행도 남겨야 할 때는 explode_outer()를 검토합니다.

# items 배열을 상품별 행으로 펼친 DataFrame 정의
order_items = (
    # 중첩 주문 DataFrame
    orders
    # 주문 식별자와 items 배열의 각 원소를 item Struct로 선택
    .select(
        "order_id",
        # items 원소 하나마다 결과 행 하나를 생성
        F.explode("items").alias("item"),
    )
    # item Struct의 하위 필드를 평면 Column으로 선택
    .select(
        "order_id",
        # item의 상품 식별자를 product_id로 지정
        F.col("item.product_id").alias("product_id"),
        # item의 주문 수량을 quantity로 지정
        F.col("item.quantity").alias("quantity"),
        # item의 상품 단가를 price로 지정
        F.col("item.price").alias("price"),
    )
)

Map은 F.col("attributes")["channel"].alias("channel")처럼 키를 지정해 값을 고릅니다. 이 예시에서 channel 값은 mobile입니다.

중첩 예제의 orders와 평면 CSV의 orders는 Schema가 다르므로 각각의 입력 구조에 맞는 코드를 사용합니다.

출처: 2026-bigdata-04.pdf, 44–48쪽 · 2026-bigdata-week04-dataframe-notes.md.

정의한 Schema를 CSV 읽기 설정에 지정합니다. 다음 경로는 강의노트의 컨테이너 내부 예시 경로입니다. 제공 실습의 12행 orders.csv를 사용할 때는 컨테이너에서 접근할 수 있는 실제 입력 경로에 맞춥니다.

# 명시적 Schema를 적용하여 주문 CSV DataFrame 정의
orders = (
    # SparkSession에서 DataFrameReader를 얻음
    spark.read
    # CSV 첫 행을 Column 이름으로 사용
    .option("header", True)
    # event_time 문자열을 해석할 시각 형식 지정
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
    # 앞에서 정의한 주문 Schema 적용
    .schema(order_schema)
    # 지정한 CSV 경로를 입력으로 사용
    .csv("/opt/spark/data/orders.csv")
)

• header: 첫 줄을 헤더로 처리합니다. 위처럼 Schema를 지정한 코드는 입력 열을 이름으로 자동 재배열하지 않으므로 CSV와 Schema의 열 순서도 맞춰 확인합니다.
• timestampFormat: 시각 문자열의 해석 형식을 지정합니다.
• schema: 앞에서 만든 order_schema를 읽기에 적용합니다.
• csv: 입력 경로를 지정하고 DataFrame을 정의합니다. 데이터 계산은 Action이 결과를 요청할 때 수행되지만, 파일 확인이나 Schema 추론 같은 준비 작업은 그 전에 이루어질 수 있습니다.

출처: 2026-bigdata-04.pdf, 23–28, 49–53쪽 · 2026-bigdata-week04-dataframe-notes.md.

Python 숫자끼리 곱하면 즉시 값이 나옵니다. Spark의 Column 표현식은 각 행의 어떤 열을 어떻게 계산할지 정의한 규칙입니다. Driver에서 이 표현식을 만들고, Executor가 각 행에 적용해 계산합니다.

# pyspark.sql.functions 모듈을 F라는 짧은 이름으로 사용
from pyspark.sql import functions as F

# quantity Column 참조와 price Column 참조를 곱하는 표현식 생성
amount_expression = F.col("quantity") * F.col("price")

F는 pyspark.sql.functions를 짧게 부르는 이름입니다. F.col("quantity")는 수량 값 하나가 아니라 열 참조를 반환합니다. amount_expression에도 24000이라는 결과가 들어 있는 것이 아닙니다.

필요한 열과 계산식 선택

select()는 필요한 열을 고르고 계산식으로 새 열을 정의합니다. alias()는 결과 열 이름을 지정합니다.

# Column 참조와 계산식을 이용하여 selected DataFrame 정의
selected = orders.select(
    # order_id Column 참조
    F.col("order_id"),
    # user_id Column 참조
    F.col("user_id"),
    # quantity와 price를 곱하고 결과 Column 이름을 amount로 지정
    (F.col("quantity") * F.col("price")).alias("amount"),
)

한 행에 적용되는 계산

수량이 2이고 단가가 12000.00인 행에는 amount=24000.00이 만들어집니다. 코드에는 특정 주문의 숫자를 직접 넣지 않았으므로 같은 규칙을 다른 행에도 적용할 수 있습니다.

Python에서 호출해도 행 계산은 Executor에서

내장 연산의 실행 위치: Python으로 처리 방법을 정의하고, 실제 주문 행의 계산은 Executor가 수행합니다.

내장 DataFrame 연산과 달리, 사용자가 작성한 Python UDF(사용자 정의 함수)나 RDD의 Python 함수는 Executor의 Python Worker에서 실행될 수 있습니다.

출처: 2026-bigdata-04.pdf, 23–24, 54–58쪽 · 2026-bigdata-week04-dataframe-notes.md.

집계할 주문의 조건을 먼저 정합니다. 이 예제에서는 수량·가격이 NULL이 아니고, 수량이 0보다 크며, 가격이 0 이상인 주문을 남깁니다. filter()와 where()는 같은 조건이면 같은 행을 선택합니다.

# 수량과 가격이 데이터 품질 조건을 만족하는 행만 선택
valid_orders = orders.filter(
    # quantity가 NULL이 아닌지 검사
    (F.col("quantity").isNotNull()) &
    # price가 NULL이 아닌지 검사
    (F.col("price").isNotNull()) &
    # quantity가 0보다 큰지 검사
    (F.col("quantity") > 0) &
    # price가 0 이상인지 검사
    (F.col("price") >= 0)
)

• &, |, ~: 각각 Column 조건의 AND, OR, NOT입니다. Python의 and, or, not으로 대체하지 않습니다.
• 괄호: 각 비교식을 괄호로 묶어 연산 순서를 분명히 합니다.
• isNull·isNotNull: NULL인지 직접 확인합니다. == None으로 일반 값처럼 비교하지 않습니다.

보충 근거: Spark 4.1.3 NULL 의미 규칙.

filter()는 조건이 True인 행만 남깁니다. NULL은 0이나 빈 문자열과 같지 않으며, NULL과의 일반 비교는 알 수 없는 결과가 될 수 있습니다.

fillna({"quantity": 0})는 NULL을 0으로 대체하므로 단순 검사와 의미가 다릅니다. 대체해도 되는지 업무 규칙을 먼저 정합니다.

Schema의 NULL 허용과 실제 행의 NULL

NULL 여부는 isNull()과 isNotNull()으로 검사합니다. 아래는 Spark가 NULL 여부를 저장하는 방식에 대한 보충 설명입니다.

nullable=True는 열이 NULL을 가질 수 있다는 선언입니다. 모든 행이 NULL이라는 뜻은 아닙니다.

Spark SQL의 대표적 내부 행 형식인 UnsafeRow는 NULL 여부를 별도 비트 영역으로 관리합니다. 필드의 비트가 1이면 그 행의 필드가 NULL입니다. 애플리케이션에서는 이 내부 비트를 직접 고치지 않고 위 공개 API를 사용합니다.

출처: 2026-bigdata-04.pdf, 59–61쪽 · 2026-bigdata-week04-dataframe-notes.md.

집계에 필요한 주문금액을 만들기 위해 정상 주문의 수량과 단가를 곱합니다. withColumn()은 열을 추가하거나 같은 이름의 열을 바꾸는 연산이 포함된 새 DataFrame을 반환합니다.

# 정상 주문에 amount Column을 추가한 새 DataFrame 정의
enriched_orders = valid_orders.withColumn(
    # 추가할 Column 이름
    "amount",
    # 각 행의 주문 수량과 상품 단가를 곱하는 표현식
    F.col("quantity") * F.col("price"),
)

DataFrame은 불변 구조입니다. orders = orders.withColumn(...)처럼 같은 Python 변수에 다시 대입해도 기존 객체가 제자리에서 바뀌지는 않습니다. 변수가 새 DataFrame을 가리키는 것입니다.

조건에 따른 값 만들기

# amount 값에 따라 amount_level을 분류한 새 DataFrame 정의
classified_orders = enriched_orders.withColumn(
    # 추가할 분류 결과 Column 이름
    "amount_level",
    # amount가 100000 이상이면 HIGH 반환
    F.when(F.col("amount") >= 100000, "HIGH")
     # 그 밖의 행은 NORMAL 반환
     .otherwise("NORMAL"),
)

when()은 조건을 만족할 때의 값을, otherwise()는 나머지 경우의 값을 지정합니다. 이 예제의 분류 기준 금액은 100000입니다.

출처: 2026-bigdata-04.pdf, 62–65쪽 · 2026-bigdata-week04-dataframe-notes.md.

개별 주문금액을 사용자별로 묶으면 주문 수·총액·평균을 알 수 있습니다. Group Key(그룹 기준)는 값이 같은 행끼리 묶는 기준 열이며, 여기서는 user_id입니다.

# 사용자별 주문 수, 총 구매금액, 평균 구매금액을 계산
summary = (
    # amount가 추가된 정상 주문 DataFrame
    enriched_orders
    # user_id가 같은 행을 집계 그룹으로 지정
    .groupBy("user_id")
    # 각 사용자 그룹에 적용할 집계식 지정
    .agg(
        # 사용자별 주문 행 수를 order_count로 지정
        F.count("*").alias("order_count"),
        # 사용자별 amount 합계를 total_amount로 지정
        F.sum("amount").alias("total_amount"),
        # 사용자별 amount 평균을 avg_amount로 지정
        F.avg("amount").alias("avg_amount"),
    )
)

• groupBy: 그룹 기준을 지정하고 집계식을 받을 GroupedData를 반환합니다.
• agg: 그룹별 집계식을 지정한 DataFrame을 반환합니다.
• F.count("*"): 각 그룹의 주문 행 수를 계산하는 표현식입니다. orders.count()처럼 전체 행 수를 Python 숫자로 반환하는 Action과 다릅니다.
• F.sum·F.avg: 주문금액의 합계와 평균입니다. 여기서는 앞서 정제한 주문만 대상입니다.

같은 키를 다시 모으는 이유

이처럼 연산에 필요한 기준으로 Partition 사이 데이터를 다시 배치하는 과정을 Shuffle(셔플)이라고 합니다.

표는 데이터 이동을 설명하는 학습용 도식이며, 실제 실행에서는 부분 집계 등을 거쳐 전송할 데이터가 줄어들 수 있습니다.

groupBy().agg()를 쓸 때는 계획을 정의합니다. 실제 계산과 필요한 데이터 이동은 Action이 결과를 요구할 때 이루어집니다.

출처: 2026-bigdata-04.pdf, 9–10, 66–67쪽 · 2026-bigdata-week04-dataframe-notes.md.

집계 결과를 비교하려면 출력 순서를 정해야 합니다. 총 구매금액 내림차순으로 정렬하되, 금액이 같으면 사용자 식별자 오름차순으로 정렬하면 동점의 순서도 명확해집니다.

# 총 구매금액 내림차순, 사용자 식별자 오름차순으로 정렬
ranked = summary.orderBy(
    # total_amount가 큰 행을 먼저 배치
    F.desc("total_amount"),
    # total_amount가 같으면 user_id가 작은 행을 먼저 배치
    F.asc("user_id"),
)

• F.desc(): 큰 값부터 배치하는 내림차순입니다.
• F.asc(): 작은 값부터 배치하는 오름차순입니다.
• orderBy(): 전체 결과의 정렬 기준을 정의하는 Transformation입니다.

분산된 결과를 전체 순서에 맞게 배치하려면 일반적으로 Shuffle이 필요합니다. 정확한 실행 구조는 입력과 실행계획에 따라 달라질 수 있습니다.

강의의 4행 예시를 손으로 확인하기

첫 절의 4행만 입력으로 삼아 앞의 수량·가격 조건을 적용하면 정상 주문 3행이 남습니다. 아래는 이 4행에서 손으로 계산한 예상값입니다.

U001은 24000원짜리 주문 두 건을 합쳐 48000원입니다. U003의 수량 0 주문은 제외되므로 결과에 나타나지 않습니다.

Decimal 연산의 출력 소수 자릿수는 위 표와 다를 수 있으므로 표시 폭보다 숫자 값·행 수·열 이름·정렬 기준을 비교합니다.

실제로 실행할 입력이 준비되면 ranked.show(truncate=False)로 결과를 확인합니다. 예상과 다르면 정제 전후 행 수, 빠진 사용자, 금액 계산, 그룹 키, 정렬 기준을 차례로 확인합니다.

출처: 2026-bigdata-04.pdf, 68–78쪽 · 2026-bigdata-week04-dataframe-notes.md.

연산을 적을 때마다 전체 데이터를 읽으면 중간 결과를 반복해서 만들 수 있습니다. Spark는 Transformation으로 무엇을 할지 모아 두고 Action이 결과를 요구할 때 실행합니다. 이를 Lazy Evaluation(지연 평가)이라고 합니다.

DataFrame이 기억하는 것

Logical Plan(논리 계획)은 어떤 입력에 어떤 연산을 적용할지 나타냅니다. 실제 계산된 전체 행이나 실행 중인 Task와는 다릅니다. 연산을 연결하면서 계획에 처리 단계가 추가됩니다.

• orders: CSV 입력을 나타냅니다.
• valid_orders: 입력과 정상 주문을 고르는 필터를 포함합니다.
• enriched_orders: 앞의 계획에 금액 계산을 더합니다.
• summary: 집계까지 포함한 계획을 나타냅니다.

Action 요청에 따라 Spark는 계획을 분석·최적화하고 Physical Plan(물리 계획)이라는 구체적 실행 방법을 정합니다. 실행은 Stage(단계)와 Task로 나뉩니다. Action 하나가 실행하는 Job 수는 처리 과정에 따라 달라질 수 있습니다.

출력 사이의 실행 시점 읽기

다음 코드는 간단한 수량 필터로 지연 평가의 실행 시점을 보여 줍니다. 앞 절의 전체 정제 조건을 적용한 코드는 아닙니다.

# Python 출력만 수행하며 Spark Job은 없음
print("STEP_A")

# Transformation 연결: Logical Plan만 구성
summary = (
    orders
    .filter(F.col("quantity") > 0)
    .withColumn("amount", F.col("quantity") * F.col("price"))
    .groupBy("user_id")
    .agg(F.sum("amount").alias("total_amount"))
)

# 아직 Spark Job은 없음
print("STEP_B")

# Action: 위 Logical Plan을 실행하는 Spark Job 시작
summary.show()

# show()의 Job이 끝난 후 출력
print("STEP_C")

• STEP_A → STEP_B: 제시된 Transformation으로 계획을 구성합니다. 이 연산의 결과 계산을 아직 요청하지 않습니다.
• STEP_B → STEP_C: show()가 결과를 요구하고 계산·출력이 끝난 뒤 STEP_C로 넘어갑니다.
• 반복 Action: count, show, collect를 각각 호출하면 별도의 계산 요청입니다. 캐시를 사용하지 않으면 앞선 연산을 다시 계산할 수 있습니다.

출처: 2026-bigdata-04.pdf, 79–81쪽 · 2026-bigdata-week04-dataframe-notes.md.

결과를 Python에서 다루려면 어디로 얼마나 가져오는지 먼저 판단해야 합니다. collect()는 모든 결과 행을 Driver 메모리로 가져옵니다. Worker가 많아도 Driver 한곳에 모을 수 있는 크기에는 한계가 있습니다.

rows = orders.collect()  # 모든 원본 행을 Driver로 이동: 대용량에서는 피할 패턴
for row in rows:
    amount = row["quantity"] * row["price"]

위 코드는 위험한 패턴을 설명하는 예입니다. 이미 정의한 withColumn(), groupBy(), agg()로 Executor에서 처리할 수 있는 일을 Driver의 Python 반복문으로 옮기지 않습니다.

보충 근거: Apache Spark collect API는 모든 결과가 Driver 메모리에 적재되므로 작은 결과에만 사용하도록 설명합니다.

출처: 2026-bigdata-04.pdf, 82–90쪽 · 2026-bigdata-week04-dataframe-notes.md.

코드가 실행되어도 잘못된 행을 포함하거나 집계 기준을 다르게 적용할 수 있습니다. AI가 작성한 코드도 입력·정제 조건·집계 기준·실행 위치·예상 결과를 차례로 확인합니다.

실행 중 문제가 생겼을 때

• 열을 찾지 못함: orders.columns와 printSchema()에서 실제 이름·오타를 확인합니다.
• CSV 경로를 찾지 못함: 컨테이너 내부 경로와 실제 마운트 위치를 확인합니다.
• 예상보다 결과가 적음: NULL과 0·음수의 정제 조건을 확인합니다. 배열을 펼쳤다면 빈 배열·NULL 배열도 살펴봅니다.
• Worker가 모두 바쁘지 않음: 입력 Partition과 Task 수를 먼저 확인합니다. Worker 수만으로 병렬성이 결정되지는 않습니다.
• 작은 CSV가 느림: 세션·작업 시작 시간과 Shuffle 시간이 처리 시간에 포함될 수 있습니다.

Spark SQL로 같은 조건 표현하기

5장에서는 같은 조건을 Spark SQL로 표현하고 결과를 비교합니다.

아래 두 표현은 모두 수량이 양수인 행을 선택합니다. SQL 예시는 orders라는 임시 뷰가 있어야 실행할 수 있습니다.

orders.filter(F.col("quantity") > 0)

SELECT *
FROM orders
WHERE quantity > 0

Schema는 표의 구조를 정하고, Column 표현식은 처리할 연산을 정의합니다. Action이 결과를 요청하면 Spark가 이 연산들을 분산 실행합니다.