[Spark Ch-03] DataFrame API: union은 위치로, unionByName은 이름으로 맞춘다

결론부터

1. withColumn은 같은 이름이 있으면 바꾸고 없으면 추가합니다. drop과 withColumnRenamed는 없는 컬럼 이름을 주면 오류 없이 아무 일도 하지 않습니다(no-op). 오타가 오류로 드러나지 않으니 시험 보기에서 함정이 되기 좋습니다.
2. na.drop()의 기본값은 how='any'라서 null이 하나라도 있는 행을 지웁니다. thresh를 주면 how를 무시합니다. F.count(col)은 null을 빼고 세고 count(*)와 DataFrame.count()는 모든 행을 셉니다.
3. join의 how 기본값은 inner입니다. on에 컬럼 이름(문자열이나 리스트)을 주면 키 컬럼이 결과에 하나만 남고 조건식을 주면 양쪽 컬럼이 다 남습니다. left_semi와 left_anti는 왼쪽 컬럼만 돌려줍니다.
4. union과 unionAll은 같은 메서드이고 컬럼을 위치로 맞추며 중복을 없애지 않습니다. 이름으로 맞추려면 unionByName을 씁니다. 한쪽에 없는 컬럼을 null로 채우려면 allowMissingColumns=True(3.1.0~)를 줍니다.
5. Spark 4.0부터 ANSI 모드가 기본으로 켜져서 to_date, to_timestamp, unix_timestamp가 해석할 수 없는 문자열에 null 대신 오류를 냅니다. null로 받으려면 try_to_date(4.0.0~) 같은 try_ 함수를 씁니다. udf의 반환 타입 기본값은 StringType이고 4.2.0부터 Python UDF가 Arrow를 기본으로 씁니다.

시험 안내서(2026년 10월 8일판, 2025년 10월판과 같음) 3영역의 항목은 열 개입니다. 비슷한 것끼리 묶어 목차로 씁니다. 11절과 12절은 목록에 없지만 중복 제거, 집계, 배열 처리 문항의 보기에 섞이기 쉬워 덧붙였습니다.

예제 데이터입니다. 주문에 상품 배열과 null을 일부러 넣었습니다.

1
from pyspark.sql import SparkSession, Row
2
from pyspark.sql import functions as F
3
from pyspark.sql.types import (StructType, StructField, IntegerType, LongType,
4
                               StringType, ArrayType)
5

6
spark = SparkSession.builder.master("local[4]").appName("ch03-df").getOrCreate()
7
spark.conf.set("spark.sql.session.timeZone", "UTC")   # 날짜 예제의 결과를 고정하려고
8

9
orders = spark.createDataFrame(
10
    [
11
        (1, 101, "PAID",     32000, 1790812800, ["book", "pen"]),
12
        (2, 102, "PAID",     15000, 1790812800, ["mug"]),
13
        (3, 101, "CANCELED",  8000, 1790947800, []),
14
        (4, 103, "PAID",     54000, 1790947800, None),
15
        (5, 102, None,       15000, 1791018900, ["book"]),
16
        (5, 102, None,       15000, 1791018900, ["book"]),   # 완전히 같은 중복 행
17
    ],
18
    "order_id INT, user_id INT, status STRING, amount INT, ordered_at LONG, items ARRAY<STRING>",
19
)
20
users = spark.createDataFrame(
21
    [(101, "Kim Minji", "Seoul"), (102, "Lee Junho", "Busan"), (104, "Park Sora", None)],
22
    "user_id INT, full_name STRING, city STRING",
23
)
24
payments = spark.createDataFrame(
25
    [(9001, 1, 101, "CARD"), (9002, 2, 102, "BANK"), (9003, 4, 103, "CARD")],
26
    "payment_id INT, order_id INT, user_id INT, method STRING",
27
)

ordered_at의 세 값은 UTC 기준 2026-10-01 00:00:00, 2026-10-02 13:30:00, 2026-10-03 09:15:00의 epoch 초입니다(Python datetime으로 계산한 값).

1. 컬럼과 행 다루기

1-1. 시그니처

withColumnRenamed의 인자 순서는 (기존 이름, 새 이름)입니다. 시험 안내서 샘플 문항 3번의 오답 보기 B가 순서를 뒤집은 코드입니다.

1
o = (orders
2
     .withColumn("amount_vat", F.col("amount") * 1.1)       # 추가
3
     .withColumn("amount", F.col("amount").cast("long"))    # 같은 이름이라 교체
4
     .withColumnRenamed("ordered_at", "ordered_epoch")      # (기존, 새)
5
     .withColumnsRenamed({"user_id": "uid"})                # 3.4.0~
6
     .drop("amount_vat", "no_such_column"))                 # 없는 이름은 무시

1-2. split의 정규식과 limit

split의 pattern은 정규식이라 .이나 |를 그대로 쓰면 의도와 다르게 나뉩니다. limit이 양수면 결과 배열 길이가 limit을 넘지 않고 마지막 원소에 나머지가 다 들어갑니다. 0 이하면 가능한 만큼 나눕니다. 기본값 -1은 3.0에서 추가됐습니다.

1
names = users.select(
2
    F.split("full_name", " ").getItem(0).alias("last_name"),   # 'Kim'
3
    F.split("full_name", " ").getItem(1).alias("first_name"),  # 'Minji'
4
)
5

6
skus = spark.createDataFrame([("A.B.C",)], "sku STRING")
7
skus.select(F.split("sku", r"\.").alias("parts"))        # 문서 기준 예상 결과: ['A', 'B', 'C']
8
skus.select(F.split("sku", r"\.", 2).alias("parts"))     # 문서 기준 예상 결과: ['A', 'B.C']

4.0.0부터 pattern에 Column도 받습니다. 다만 문자열은 계속 정규식으로 해석하므로 컬럼 이름 문자열을 pattern으로 줄 수는 없습니다.

1-3. filter

1
orders.filter(F.col("amount") >= 15000)
2
orders.where("amount >= 15000 AND status = 'PAID'")              # SQL 문자열
3
orders.filter((F.col("status") == "PAID") & (F.col("amount") > 20000))   # 괄호 필수
4
orders.filter(F.col("status").isin("PAID", "REFUNDED"))
5
orders.filter(~F.col("status").isin("CANCELED"))                 # ~ 는 NOT

Python의 and, or, not은 Column에 쓸 수 없고 &, |, ~를 씁니다. &가 비교 연산자보다 먼저 묶이므로 각 조건을 괄호로 감쌉니다. status가 null인 행은 status == "PAID"도 status != "PAID"도 통과하지 못합니다. NULL Semantics 문서는 WHERE와 JOIN 조건에서 결과가 True인 행만 남고 null(unknown)은 걸러진다고 적습니다.

1-4. explode와 explode_outer

1
orders.select("order_id", F.explode("items").alias("item"))
2
# 문서 기준 예상 결과: 주문 1은 2행, 2는 1행, 5는 2행(중복 행 2개 각각 1행), 3(빈 배열)과 4(null)는 없음
3

4
orders.select("order_id", F.explode_outer("items").alias("item"))
5
# 주문 3과 4가 item = null 인 행으로 남는다

explode 문서는 SELECT 하나에 explode를 하나만 쓸 수 있다고 적습니다. 두 배열을 펼치려면 select를 두 번 이어 씁니다.

헷갈리는 지점

• withColumn 문서는 반복문에서 여러 번 부르면 내부 projection이 쌓여 계획이 커지고 성능 문제나 StackOverflowException이 날 수 있다고 경고합니다. 여러 컬럼은 select나 withColumns로 한 번에 만듭니다.
• withColumn의 두 번째 인자는 Column 식이어야 합니다. 상수는 F.lit(0)으로 감쌉니다. 다른 DataFrame의 컬럼을 넣으면 오류가 납니다.
• drop("a")와 drop(F.col("a"))는 의미가 다를 수 있다고 문서가 적습니다. 이름은 글자 그대로 찾고 Column은 같은 식을 찾습니다. 조인 뒤 같은 이름 컬럼이 둘일 때 차이가 납니다.

2. 중복 제거와 검증

2-1. distinct와 dropDuplicates

1
DataFrame.distinct()
2
DataFrame.dropDuplicates(subset=None)      # drop_duplicates는 별칭

1
orders.distinct().count()                       # 문서 기준 예상 결과: 5 (주문 5 중복 제거)
2
orders.dropDuplicates(["user_id"]).count()      # 회원 101, 102, 103 하나씩 → 3
3
orders.select("user_id").distinct().count()     # 3, 다만 결과 컬럼은 user_id 하나

dropDuplicates(subset)이 같은 키의 여러 행 중 어느 행을 남기는지는 API 문서에 적혀 있지 않습니다. 특정 행(예: 가장 최근 주문)을 남겨야 하면 윈도 함수 row_number()로 직접 고릅니다(11절). 스트리밍 DataFrame의 dropDuplicates는 트리거를 넘어 모든 데이터를 상태로 들고 있고 withWatermark()로 범위를 줄인다고 문서가 적습니다(5영역에서 다룹니다).

2-2. null 처리

1
DataFrame.dropna(how='any', thresh=None, subset=None)     # = df.na.drop(...)
2
DataFrame.fillna(value, subset=None)                      # = df.na.fill(...)
3
DataFrameNaFunctions.replace(to_replace, value=<no value>, subset=None)

시험 안내서 샘플 문항 6번이 “컬럼 중 하나라도 값이 빠진 행을 모두 지우는 코드”로 na.drop()을 정답으로 듭니다.

fillna에는 함정이 둘 있습니다. 값이 dict면 subset을 무시합니다. 값의 타입과 맞지 않는 컬럼은 subset에 있어도 조용히 건너뜁니다.

1
orders.na.fill("UNKNOWN", subset=["status", "amount"])   # amount는 정수라 무시되고 status만 채워진다
2
orders.na.fill({"status": "UNKNOWN", "amount": 0})       # 컬럼마다 다른 값
3
orders.na.replace("REFUNDED", "RETURNED", subset=["status"])

2-3. 검증 조건

1
orders.filter(F.col("status").isNull())          # Column.isNull()
2
orders.filter(F.col("status").isNotNull())
3
orders.filter(F.isnull("status"))                # 함수형도 있다
4
orders.select(F.coalesce("status", F.lit("UNKNOWN")).alias("status"))   # 첫 번째 null 아닌 값
5

6
# 검증 규칙을 컬럼으로 만들고 통과하지 못한 행만 따로 본다
7
checked = orders.withColumn(
8
    "is_valid",
9
    F.col("status").isNotNull() & (F.col("amount") > 0) & (F.size("items") > 0),
10
)
11
checked.filter(~F.col("is_valid"))

마지막 예에서 items가 null이면 size의 결과가 무엇인지가 판정을 바꾸고 이 값은 버전마다 다릅니다. SQL 함수 문서 기준으로 size는 spark.sql.ansi.enabled가 false이고 spark.sql.legacy.sizeOfNull이 true일 때만 null 입력에 -1을 돌려주고 그 밖에는 null을 돌려줍니다.

4.x에서는 is_valid가 null이라 ~is_valid도 null이고 그 행은 두 필터 어디에도 잡히지 않습니다. 검증 컬럼은 F.coalesce(조건, F.lit(False))로 null을 거짓으로 바꿔 두면 두 버전에서 결과가 같습니다.

헷갈리는 지점

• col == None으로 null을 찾으면 결과가 null이라 아무 행도 통과하지 못합니다. isNull()을 씁니다.
• distinct()와 GROUP BY는 null을 같은 값으로 묶습니다. 비교 연산(=)에서는 null과 null이 같지 않습니다. NULL Semantics 문서에 둘 다 나옵니다.
• na.drop()은 NaN도 null처럼 지웁니다(문서 문장 “null or NaN values”). thresh도 null과 NaN이 아닌 값의 개수로 셉니다(문서 예제 3).

3. 집계

3-1. 세는 방법 다섯 가지

1
orders.select(
2
    F.count("*").alias("rows"),                     # 6
3
    F.count("status").alias("status_not_null"),     # 4
4
    F.count_distinct("user_id").alias("users"),     # 3
5
    F.approx_count_distinct("user_id").alias("users_approx"),
6
)
7
# 문서 기준 예상 결과(근사값 제외): 6, 4, 3

approx_count_distinct도 셔플을 합니다. 차이는 셔플하는 양입니다. Spark 4.2.0 소스(AggUtils.planAggregateWithOneDistinct)를 보면 count_distinct는 그룹 키에 DISTINCT 대상 컬럼을 더한 조합으로 먼저 묶습니다. 그래서 그룹마다 고유값이 모두 셔플됩니다. approx_count_distinct는 일반 집계처럼 파티션 안에서 부분 집계를 해 그룹마다 HyperLogLog++ 스케치 하나를 만들고 그 스케치만 셔플합니다. 고유값이 많을수록 셔플 데이터가 크게 줄어듭니다. 이 설명은 소스를 읽고 정리한 것이고 같은 내용의 문서 문장은 찾지 못했습니다.

approx_count_distinct 문서는 rsd를 0.01보다 작게 잡을 거면 count_distinct()를 쓰는 편이 효율적이라고 적습니다. countDistinct는 3.5.9 문서에 count_distinct의 별칭으로 나오고 count_distinct를 직접 쓰라고 권합니다. 4.2.0 API 문서에는 countDistinct 페이지가 없지만 4.2.0 소스(functions/builtin.py)에는 같은 설명의 별칭으로 남아 있어 호출은 됩니다.

3-2. 평균, 합계, groupBy

F.mean(col)은 F.avg(col)의 별칭입니다. 집계 함수는 null을 무시합니다(count(*) 제외). 모든 입력이 null이거나 입력이 비면 max, min, sum, avg는 null을 돌려준다고 NULL Semantics 문서가 적습니다.

1
DataFrame.groupBy(*cols)            # groupby는 별칭. 4.0.0부터 1부터 시작하는 컬럼 위치(int)도 받는다
2
GroupedData.agg(*exprs)             # dict 하나 또는 Column 여러 개
3
GroupedData.count()                 # 결과 컬럼 이름은 count

1
by_user = (orders
2
           .groupBy("user_id")
3
           .agg(F.count("*").alias("orders"),
4
                F.sum("amount").alias("total"),
5
                F.mean("amount").alias("avg_amount"),
6
                F.max("ordered_at").alias("last_at")))
7

8
orders.groupBy("user_id").agg({"amount": "avg", "*": "count"})   # dict 형식. 결과 이름은 avg(amount), count(1)
9
orders.groupBy("status").count()                                  # 컬럼 이름 status, count. null status도 한 그룹

agg 문서는 내장 집계 함수와 그룹 집계 pandas UDF를 한 번의 agg 호출에 섞을 수 없다고 적습니다.

3-3. describe와 summary

1
orders.describe("amount").show()
2
orders.summary().show()                    # 사분위수까지
3
orders.summary("count", "min", "90%").show()

두 메서드 모두 숫자와 문자열 컬럼에 대해 통계를 냅니다. 결과는 통계 이름 컬럼(summary)이 붙은 DataFrame이고 값은 문자열로 나옵니다. 문서는 결과 스키마의 하위 호환을 보장하지 않는 탐색용 기능이라고 적습니다. 사분위수는 근사값입니다.

헷갈리는 지점

• df.count()는 액션이라 int를 돌려주고 F.count()는 Column이라 select나 agg 안에 씁니다.
• summary()의 인자는 통계 이름입니다. 컬럼을 고르려면 df.select("amount").summary()처럼 먼저 고릅니다. 4.2.0 문서의 Parameters 설명에는 “Column names to calculate statistics by”라고 적혀 있지만 본문과 예제는 통계 이름을 받습니다.
• GroupedData.count()의 결과 컬럼 이름은 count이고 F.count("*")를 별칭 없이 쓰면 count(1)입니다.

4. 날짜와 시간

4-1. 시그니처와 반환 타입

from_unixtime 문서는 결과를 “현재 시스템 시간대” 기준 문자열로 낸다고 적습니다. 위 예제는 결과를 고정하려고 세션 시간대(spark.sql.session.timeZone, 기본 로컬 시간대)를 UTC로 맞췄습니다.

1
dated = orders.select(
2
    "order_id",
3
    F.from_unixtime("ordered_at").alias("ordered_str"),                  # '2026-10-01 00:00:00'
4
    F.from_unixtime("ordered_at", "yyyy/MM/dd").alias("ordered_ymd"),    # '2026/10/01'
5
    F.timestamp_seconds("ordered_at").alias("ordered_ts"),
6
    F.to_date(F.from_unixtime("ordered_at")).alias("ordered_date"),     # 2026-10-01 (DateType)
7
)
8
parts = dated.select(
9
    "order_id",
10
    F.year("ordered_ts").alias("y"),           # 2026
11
    F.month("ordered_ts").alias("m"),          # 10
12
    F.dayofmonth("ordered_ts").alias("d"),     # 1, 2, 3
13
    F.dayofweek("ordered_ts").alias("dow"),    # 10월 1일(목) 5, 10월 3일(토) 7
14
    F.date_format("ordered_ts", "EEE").alias("dow_name"),
15
)
16
# 문서 기준 예상 결과. 세션 시간대 UTC 기준이고 실행해서 확인하지 않았다

dayofweek는 일요일 1부터 토요일 7까지입니다. 월요일을 1로 생각하면 틀립니다. datediff는 인자 순서가 (end, start)라서 datediff("2026-10-03", "2026-10-01")이 2입니다.

4-2. 3.x와 4.x의 차이: ANSI 모드

4.2.0 SQL Migration Guide는 Spark 4.0부터 spark.sql.ansi.enabled가 기본으로 켜졌다고 적습니다. ANSI Compliance 문서는 ANSI 모드에서 to_date, to_timestamp, unix_timestamp, to_unix_timestamp가 입력을 해석할 수 없거나 패턴이 잘못되면 실패한다고 적습니다.

unix_timestamp의 API 문서는 3.5.9와 4.2.0 모두 “실패하면 null을 돌려준다”고 적습니다. ANSI 문서와 엇갈리는데 4.2.0 소스를 보면 ANSI 문서 쪽이 맞습니다. UnixTimestamp 식의 failOnError 기본값이 spark.sql.ansi.enabled이고 이 값이 true면 해석 실패에서 예외를 던지며 false일 때만 null을 돌려줍니다. 4.x 기본 설정에서는 오류가 납니다.

같은 가이드는 4.0에서 spark.sql.legacy.timeParserPolicy 기본값이 EXCEPTION에서 CORRECTED로 바뀌었다고 적습니다. ANSI 모드에서는 CANNOT_PARSE_TIMESTAMP 오류가 나고 ANSI가 꺼져 있으면 null이 나옵니다.

1
raw = spark.createDataFrame([("2026-10-01",), ("2026-13-45",), (None,)], "d STRING")
2
raw.select(F.try_to_date("d").alias("d"))     # 4.0.0~. 잘못된 값은 null
3
# raw.select(F.to_date("d"))                  # 4.x 기본 설정에서는 두 번째 행 때문에 액션에서 오류

헷갈리는 지점

• from_unixtime은 문자열을 돌려줍니다. 날짜 연산을 하려면 to_date나 to_timestamp로 다시 바꾸거나 timestamp_seconds를 씁니다.
• date_format 문서는 가능하면 year 같은 전용 함수를 쓰라고 권합니다. date_format(col, "yyyy")는 문자열 '2026'이고 year(col)은 정수 2026입니다.
• epoch 밀리초를 from_unixtime에 그대로 넣으면 초로 해석합니다. 1000으로 나누거나 timestamp_millis를 씁니다.

5. 조인

5-1. 시그니처와 조인 종류

1
DataFrame.join(other, on=None, how=None)     # how 기본값 'inner'
2
DataFrame.crossJoin(other)

how에 쓸 수 있는 문자열(4.2.0 문서): inner, cross, outer, full, fullouter, full_outer, left, leftouter, left_outer, right, rightouter, right_outer, semi, leftsemi, left_semi, anti, leftanti, left_anti.

5-2. on을 주는 방법에 따라 달라지는 컬럼

1
o = orders.dropDuplicates()
2

3
# 1) 이름으로: user_id 컬럼 하나
4
o.join(users, "user_id")                        # inner, 103과 104는 빠진다
5

6
# 2) 조건식으로: user_id가 둘이 되어 그냥 "user_id"를 고르면 모호하다
7
j = o.join(users, o.user_id == users.user_id, "left")
8
j.select(o.user_id, users.city)                 # DataFrame을 붙여 고른다
9

10
# 3) 다중 키
11
o.join(payments, ["order_id", "user_id"], "left")
12

13
# 4) 결제가 있는 주문만, 결제가 없는 주문만 (왼쪽 컬럼만 남는다)
14
o.join(payments, "order_id", "left_semi")       # 주문 1, 2, 4
15
o.join(payments, "order_id", "left_anti")       # 주문 3, 5
16

17
# 5) cross
18
o.crossJoin(users).count()                      # 문서 기준 예상 결과: 5 x 3 = 15

자기 자신과 조인하면 직접 컬럼 참조가 모호해진다고 문서가 경고합니다. 이때는 alias로 DataFrame에 이름을 붙이고 F.col("a.user_id")처럼 부릅니다.

1
a, b = o.alias("a"), o.alias("b")
2
same_user = a.join(b, (F.col("a.user_id") == F.col("b.user_id")) &
3
                      (F.col("a.order_id") < F.col("b.order_id")))

조인 키가 null이면 = 비교가 null이 되어 매칭되지 않습니다. null끼리 맞추려면 null-safe 비교(<=>, PySpark에서는 Column.eqNullSafe)를 씁니다. eqNullSafe 문서 예제에서 null 값과 'foo'를 비교하면 ==는 NULL, eqNullSafe는 false이고 eqNullSafe(None)은 true입니다.

5-3. broadcast 조인

broadcast 조인은 작은 쪽 DataFrame 전체를 모든 익스큐터에 보내고 큰 쪽은 셔플하지 않은 채 각 파티션에서 바로 조인합니다. 큰 쪽의 셔플을 없애는 것이 목적입니다.

Performance Tuning 문서는 broadcast 힌트를 주면 추정 크기가 임계값을 넘어도 그 쪽을 build 쪽으로 broadcast 조인을 우선한다고 적습니다. 다만 조인 종류가 지원하지 않으면 힌트를 따르지 않을 수 있습니다. 양쪽에 힌트가 다르면 우선순위는 BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL입니다. broadcast 대기 시간 상한은 spark.sql.broadcastTimeout(기본 300초)입니다.

1
big = o            # 실제로는 수억 건이라고 가정
2
small = users      # 수천 건
3

4
joined = big.join(F.broadcast(small), "user_id", "left")
5
joined.explain()   # 물리 계획에 BroadcastHashJoin이 나오는지 본다

헷갈리는 지점

• how를 생략하면 inner입니다. 샘플 문항류에서 “왼쪽 행이 모두 남는다”를 고르려면 how="left"가 있어야 합니다.
• left_semi는 오른쪽에 매칭이 여러 개여도 왼쪽 행을 한 번만 돌려줍니다. 4.2.0 소스의 해시 조인 구현(HashJoin.semiJoin)은 왼쪽 행마다 매칭이 있는지만 보고 걸러 내는 filter라서 행이 늘지 않습니다. inner 조인 뒤 왼쪽 컬럼만 고르면 매칭 수만큼 행이 늘어나니 중복 처리에서 다릅니다.
• broadcast 대상은 작은 쪽입니다. 큰 쪽을 broadcast()로 감싸면 드라이버와 익스큐터 메모리를 크게 씁니다.
• broadcast 조인(F.broadcast(df))과 broadcast 변수(sc.broadcast(value))는 다른 API입니다. 10절에서 다룹니다.

6. union, unionAll, unionByName

1
DataFrame.union(other)
2
DataFrame.unionAll(other)                                 # union의 별칭
3
DataFrame.unionByName(other, allowMissingColumns=False)   # allowMissingColumns는 3.1.0~

1
oct_orders = spark.createDataFrame([(10, 101, 5000)], "order_id INT, user_id INT, amount INT")
2
nov_orders = spark.createDataFrame([(201, 11, 7000)], "user_id INT, order_id INT, amount INT")
3

4
oct_orders.union(nov_orders)
5
# 위치로 붙어서 order_id = 201, user_id = 11 인 행이 생긴다. 타입이 같으니 오류도 없다
6

7
oct_orders.unionByName(nov_orders)
8
# 이름으로 맞춰 order_id = 11, user_id = 201
9

10
with_note = spark.createDataFrame([(12, 103, 9000, "gift")],
11
                                  "order_id INT, user_id INT, amount INT, note STRING")
12
oct_orders.unionByName(with_note, allowMissingColumns=True)
13
# oct_orders 쪽 행의 note는 null

같은 계열 집합 연산도 같이 정리해 둡니다. 4.2.0 문서 기준으로 모두 위치로 컬럼을 맞춥니다.

헷갈리는 지점

• union이 중복을 없앤다는 보기는 틀립니다. SQL의 UNION과 이름이 같아서 생기는 함정입니다. 예전 자료 중에도 union이 이름으로 맞춘다고 잘못 적은 글이 있습니다(Ch-00 5절).
• 위치가 바뀐 두 DataFrame의 컬럼 타입이 같으면 union은 오류 없이 값을 섞습니다. 4.2.0 union 문서 예제 2는 id LONG, money INT와 money INT, id LONG을 union해도 오류 없이 위치대로 붙는 모습을 보여 줍니다. 타입이 다르면 위치마다 공통 타입을 찾아 캐스트합니다(4.2.0 소스 WidenSetOperationTypes). 문자열과 정수가 같은 위치에 오면 ANSI가 꺼진 3.5에서는 STRING으로 맞추고 ANSI가 켜진 4.x에서는 BIGINT로 맞춥니다(ANSI Compliance 문서의 “least common type between Byte/Short/Int and String is Long”). 4.x에서는 숫자로 읽을 수 없는 문자열이 있으면 캐스트에서 오류가 납니다.
• 결과 컬럼 이름은 첫 번째 DataFrame을 따릅니다. 문서 예제 3에서 A, B와 C, D를 union한 결과 컬럼은 A, B입니다.
• 집합 연산(intersect, subtract)은 null끼리 같은 값으로 봅니다(NULL Semantics 문서).

7. 스키마를 지정한 읽기와 쓰기

7-1. StructType과 DDL 문자열

1
StructType(fields=None)
2
StructField(name, dataType, nullable=True, metadata=None)
3
StructType.add(field, data_type=None, nullable=True, metadata=None)

같은 스키마를 두 방식으로 씁니다.

1
order_schema = StructType([
2
    StructField("order_id", IntegerType(), False),
3
    StructField("user_id", IntegerType(), True),
4
    StructField("status", StringType(), True),
5
    StructField("amount", IntegerType(), True),
6
    StructField("ordered_at", LongType(), True),
7
    StructField("items", ArrayType(StringType()), True),
8
])
9

10
order_ddl = ("order_id INT, user_id INT, status STRING, amount INT, "
11
             "ordered_at BIGINT, items ARRAY<STRING>")

DDL 문자열로는 NOT NULL을 적지 않는 한 nullable이 true입니다. 위 두 스키마는 order_id의 nullable만 다릅니다. 4.0.0부터 StructType.fromDDL(ddl)로 DDL 문자열을 StructType 객체로 바꿀 수 있습니다.

7-2. 스키마를 지정하는 자리

DataFrameReader.schema 문서는 스키마를 주면 소스가 스키마 추론 단계를 건너뛰어 읽기가 빨라진다고 적습니다. CSV의 inferSchema=True는 데이터를 한 번 더 훑습니다.

1
orders_in = (spark.read
2
             .schema(order_ddl)
3
             .option("mode", "FAILFAST")
4
             .json("/data/orders_json"))
5

6
(orders_in.write
7
          .mode("overwrite")
8
          .parquet("/out/orders_typed"))      # Parquet는 스키마를 파일에 저장한다
9

10
spark.read.parquet("/out/orders_typed").printSchema()   # 다시 읽어도 같은 타입

createDataFrame에 컬럼 이름 리스트만 주면 타입은 데이터에서 추론합니다. 빈 DataFrame을 만들 때는 추론할 데이터가 없으므로 스키마가 필수입니다.

7-3. 스키마 확인

헷갈리는 지점

• df.schema()처럼 괄호를 붙이면 오류입니다. schema는 속성입니다. 샘플 문항 4번의 오답 보기 B에 이 코드가 있습니다.
• 파일에서 읽을 때 지정한 nullable=False는 지켜지지 않습니다. 4.2.0 소스의 DataSource.resolveRelation이 파일 소스의 데이터 스키마를 기본으로 asNullable로 바꿉니다. Parquet 문서도 읽을 때 모든 컬럼이 호환성 때문에 nullable로 바뀐다고 적습니다.
• CSV enforceSchema 기본값 true는 지정한 스키마를 파일에 강제로 적용하고 헤더 이름을 무시합니다. 문서는 잘못된 결과를 피하려면 끄라고 권합니다.

8. 정렬, 반복, 리스트 변환

8-1. 정렬

1
DataFrame.sort(*cols, **kwargs)        # ascending=True 기본
2
DataFrame.orderBy(*cols, **kwargs)     # sort와 같다
3
DataFrame.sortWithinPartitions(*cols, **kwargs)

1
orders.orderBy(F.desc("amount"), F.asc("order_id"))
2
orders.sort(F.col("amount").desc(), "order_id")
3
orders.sort(["amount", "order_id"], ascending=[False, True])   # 리스트 길이가 cols와 같아야 한다
4
orders.orderBy(F.col("status").desc_nulls_last())

sort/orderBy는 전체 정렬이라 셔플이 일어납니다. sortWithinPartitions는 파티션 안만 정렬합니다.

8-2. 드라이버로 가져오기

1
rows = orders.orderBy(F.desc("amount")).collect()
2
[r.asDict() for r in rows]            # dict 리스트
3
[r["order_id"] for r in rows]         # 한 컬럼 값 리스트
4
[r.order_id for r in orders.select("order_id").toLocalIterator()]
5

6
pdf = orders.limit(100).toPandas()    # 작게 줄인 뒤 변환

샘플 문항 4번의 정답 보기는 orderBy(col(...).desc(), ...)로 정렬하고 collect() 결과에 row.asDict()를 적용하는 코드입니다. sort(..., descending=True)는 존재하지 않는 인자이고 show()는 None을 돌려주며 toList()는 DataFrame 메서드가 아닙니다.

4.2.0부터 spark.sql.execution.arrow.pyspark.enabled가 기본 true라서 toPandas()와 pandas에서 만드는 createDataFrame이 Arrow를 씁니다(PySpark Migration Guide, 4.1 → 4.2). 3.x와 4.1까지는 기본 false였습니다.

8-3. 반복

안내서의 “iterating”은 행을 하나씩 도는 방법을 묻습니다. 어디서 도는지가 핵심이라 실행 위치로 나눕니다.

foreach와 foreachPartition은 드라이버의 리스트에 값을 모으는 용도로 쓸 수 없습니다. 함수가 익스큐터에서 돌고 각 익스큐터는 드라이버 변수의 복사본을 고칩니다. 외부 시스템에 쓰거나 accumulator를 더할 때 씁니다. 4.2.0 문서 기준으로 두 메서드는 4.0.0부터 Spark Connect에서도 쓸 수 있습니다.

DataFrame과 리스트 사이 변환은 다음 정도를 외워 둡니다.

1
orders.columns                                            # ['order_id', 'user_id', ...] 컬럼 이름 리스트
2
[r.asDict() for r in orders.collect()]                    # dict 리스트
3
[r["order_id"] for r in orders.select("order_id").collect()]   # 한 컬럼 값 리스트
4
[r[0] for r in orders.select("order_id").collect()]            # 위치로 꺼내도 같다
5
orders.select("order_id").rdd.flatMap(lambda r: r).collect()   # RDD 경유. Spark Connect에서는 .rdd를 쓸 수 없다
6

7
# 리스트 → DataFrame
8
spark.createDataFrame([(1, "PAID"), (2, "CANCELED")], ["order_id", "status"])
9
spark.createDataFrame([Row(order_id=1, status="PAID")])

헷갈리는 지점

• show()는 기본 20행, 20자 넘는 문자열 자르기(truncate=True)입니다. 반환값은 None입니다.
• take(5)와 limit(5)의 차이는 반환 타입입니다. take는 리스트(액션), limit은 DataFrame(트랜스포메이션)입니다.
• head()와 head(1)은 다릅니다. 앞은 Row, 뒤는 Row 하나가 든 리스트입니다.

9. UDF

9-1. 일반 Python UDF

1
F.udf(f=None, returnType=StringType(), *, useArrow=None)
2
spark.udf.register(name, f, returnType=None)

1
@F.udf(returnType="int")
2
def item_count(items):
3
    if items is None:          # null 처리는 함수가 직접 한다
4
        return 0
5
    return len(items)
6

7
def grade_of(amount):
8
    return "HIGH" if amount is not None and amount >= 30000 else "LOW"
9

10
grade_udf = F.udf(grade_of)    # returnType 생략 → StringType
11

12
orders.select("order_id", item_count("items").alias("n"), grade_udf("amount").alias("grade"))
13

14
# SQL에서 쓰려면 등록한다
15
spark.udf.register("grade_of", grade_of)                # 반환 타입 기본 string
16
spark.udf.register("item_count", item_count)            # 이미 udf면 그 반환 타입을 쓴다. returnType을 주면 안 된다
17
orders.createOrReplaceTempView("orders_v")
18
spark.sql("SELECT order_id, grade_of(amount) AS grade FROM orders_v")

반환 타입을 실제 값과 다르게 선언하면 결과가 틀어집니다. Arrow를 쓰지 않는 경로는 4.2.0 소스(EvaluatePython.makeFromJava)에서 선언 타입과 맞지 않는 값을 null로 바꾸고 선언이 문자열일 때만 값을 문자열로 바꿔 받습니다. 그래서 위 item_count를 returnType 없이 만들면 오류도 null도 없이 정수가 "2" 같은 문자열로 나옵니다. 4.2.0 기본인 Arrow 경로의 변환 규칙은 확인하지 못했습니다. 반환 타입은 항상 맞춰 적습니다.

9-2. pandas UDF

1
F.pandas_udf(f=None, returnType=None, functionType=None)

pandas UDF는 Arrow로 데이터를 넘기고 pandas로 계산합니다. 3.0부터는 Python 타입 힌트로 종류를 정하고 functionType은 호환용으로만 남아 있습니다. 7영역(pandas API on Spark)에서 자세히 다룹니다.

1
import pandas as pd
2

3
@F.pandas_udf("double")
4
def with_vat(amount: pd.Series) -> pd.Series:
5
    return amount * 1.1
6

7
orders.select("order_id", with_vat("amount").alias("amount_vat"))

9-3. stateful 연산자와 StateStore

안내서 문구는 “Create and invoke user-defined functions with or without stateful operators, including StateStores”입니다. 배치 DataFrame의 UDF는 행마다 독립적으로 계산하고 상태를 들고 있지 않습니다. 상태가 생기는 곳은 Structured Streaming입니다. 둘을 이렇게 나눕니다.

시험 보기에서 “UDF가 이전 배치의 값을 기억한다”는 설명은 상태 없는 UDF에는 맞지 않습니다. 이전 값을 기억해야 하면 위 표 세 번째 줄의 API를 씁니다. 각 연산자의 출력 모드와 워터마크는 Ch-05에서 다룹니다.

기본 구현은 상태를 익스큐터 JVM 메모리에 두기 때문에 상태 객체가 많으면 GC 일시 정지가 길어질 수 있다고 문서가 적습니다. RocksDB 제공자가 나온 이유입니다. 사용법은 Ch-05에서 다룹니다.

헷갈리는 지점

• UDF는 최적화기가 안을 들여다볼 수 없습니다. 같은 일을 하는 내장 함수(F.size, F.when)가 있으면 내장 함수가 낫습니다. 이 권고는 공식 문서 문장으로 확인하지 못했고 일반적인 설명입니다.
• udf가 결정적 함수로 취급되므로 최적화 과정에서 호출이 줄거나 늘 수 있다고 문서가 적습니다. 난수나 현재 시각을 쓰는 UDF는 asNondeterministic()을 붙입니다.
• register는 Python 함수에는 returnType을 줄 수 있고 이미 만든 UDF에는 줄 수 없습니다.

10. broadcast 변수와 accumulator

10-1. broadcast 변수

1
bc = spark.sparkContext.broadcast(value)   # SparkContext.broadcast(value) → Broadcast
2
bc.value
3
bc.unpersist(blocking=False)               # 익스큐터 복사본 삭제. 다시 쓰면 다시 broadcast
4
bc.destroy(blocking=False)                 # 영구 해제. 이후 사용 불가

RDD 가이드의 설명입니다. broadcast 변수는 읽기 전용 값을 Task마다 보내지 않고 머신마다 한 번 캐시해 둡니다. Spark는 Stage 안의 Task가 공통으로 쓰는 데이터를 이미 자동으로 broadcast하므로 직접 만드는 broadcast 변수는 여러 Stage가 같은 데이터를 쓰거나 역직렬화된 형태로 캐시하는 것이 중요할 때 쓸모가 있다고 적습니다. broadcast한 뒤에는 원래 객체를 바꾸지 말라고도 적습니다.

1
city_names = {"Seoul": "서울", "Busan": "부산"}
2
bc_city = spark.sparkContext.broadcast(city_names)
3

4
@F.udf("string")
5
def city_label(city):
6
    return bc_city.value.get(city, "기타")
7

8
users.select("user_id", city_label("city").alias("city_ko"))

10-2. accumulator

1
acc = spark.sparkContext.accumulator(0)    # SparkContext.accumulator(value, accum_param=None)
2
acc.add(1)                                 # 또는 acc += 1  (Task에서)
3
acc.value                                  # 드라이버에서만 읽는다

1
bad_rows = spark.sparkContext.accumulator(0)
2

3
def check(row):
4
    if row.status is None:
5
        bad_rows.add(1)
6

7
orders.foreach(check)      # 액션 안에서 더했으므로 Task마다 한 번만 반영
8
print(bad_rows.value)      # 문서 기준 예상 결과: 2 (status가 null인 중복 행 2개)

헷갈리는 지점

• broadcast 변수는 읽기 전용, accumulator는 Task 쪽에서 쓰기 전용(더하기)이라고 짝지어 외우면 됩니다.
• map 같은 트랜스포메이션 안에서 accumulator를 더하고 액션을 부르지 않으면 값은 0 그대로입니다. RDD 가이드 예제가 이 경우입니다.
• spark.sparkContext가 필요하므로 Spark Connect(Databricks 서버리스 포함)에서는 이 절의 코드를 쓸 수 없습니다(Ch-00 3절).

11. 윈도 함수

안내서 3영역 목록에 윈도 함수가 따로 적혀 있지는 않습니다. 다만 2-1절처럼 “같은 키에서 특정 행 하나 남기기”를 풀려면 필요하고 정렬, 집계 문항의 보기에 자주 섞입니다. groupBy는 그룹마다 한 행으로 줄이고 윈도 함수는 행 수를 그대로 둔 채 같은 그룹의 값을 옆에 붙입니다.

11-1. Window 스펙과 기본 프레임

1
from pyspark.sql import Window
2

3
Window.partitionBy(*cols)            # 그룹. 없으면 전체가 한 그룹
4
Window.orderBy(*cols)                # 그룹 안 순서
5
WindowSpec.rowsBetween(start, end)   # 행 개수 기준 프레임, 양 끝 포함
6
WindowSpec.rangeBetween(start, end)  # 정렬 키 값 기준 프레임, 양 끝 포함
7
Window.unboundedPreceding, Window.unboundedFollowing, Window.currentRow

Window 문서의 기본 프레임 규칙이 집계 결과를 가릅니다.

RANGE 프레임은 정렬 키가 같은 행을 한 덩어리로 봅니다. 같은 시각의 주문이 둘이면 두 행의 누적값이 같게 나옵니다. 행 단위 누적이 필요하면 rowsBetween(Window.unboundedPreceding, Window.currentRow)를 명시합니다.

partitionBy 없이 orderBy만 주면 모든 데이터가 파티션 하나로 모입니다. Spark 4.2.0 소스(WindowExecBase.scala)에 “No Partition Defined for Window operation! Moving all data to a single partition” 경고가 있습니다.

11-2. 순위와 앞뒤 행

순위 함수와 lag, lead는 orderBy가 있는 윈도에서 씁니다.

1
o = orders.dropDuplicates()
2
by_user_time = Window.partitionBy("user_id").orderBy(F.col("ordered_at").desc())
3

4
# 회원마다 가장 최근 주문 하나(2-1절의 dropDuplicates로는 고를 수 없던 것)
5
latest = (o.withColumn("rn", F.row_number().over(by_user_time))
6
           .filter(F.col("rn") == 1)
7
           .drop("rn"))
8
# 문서 기준 예상 결과: 101 → 주문 3, 102 → 주문 5, 103 → 주문 4
9

10
running = Window.partitionBy("user_id").orderBy("ordered_at")
11
o.select(
12
    "user_id", "order_id", "amount",
13
    F.sum("amount").over(running).alias("running_total"),      # 기본 프레임이라 누적
14
    F.sum("amount").over(Window.partitionBy("user_id")).alias("user_total"),   # orderBy 없음 → 그룹 합계
15
    F.lag("amount").over(running).alias("prev_amount"),       # 첫 주문은 null
16
)
17
# 회원 102: 주문 2(15000) → 누적 15000, 주문 5(15000) → 누적 30000. user_total은 둘 다 30000

row_number()는 동점일 때 어느 행이 1번이 될지 정해지지 않습니다. 결과를 고정하려면 orderBy에 order_id 같은 고유 컬럼을 하나 더 넣습니다.

12. 배열 고차 함수, pivot, 샘플링

12-1. 배열 고차 함수

배열 컬럼을 펼치지 않고 원소마다 식을 적용하는 함수들입니다. 모두 3.1.0에 Python API가 생겼고 Python lambda는 Column을 받아 Column을 돌려줘야 합니다. 문서는 lambda 안에서 Python UDF를 쓸 수 없다고 적습니다(SPARK-27052).

1
orders.select(
2
    "order_id",
3
    F.transform("items", lambda x: F.upper(x)).alias("items_upper"),
4
    F.filter("items", lambda x: x != "pen").alias("without_pen"),
5
    F.exists("items", lambda x: x == "book").alias("has_book"),
6
    F.forall("items", lambda x: F.length(x) <= 4).alias("short_names"),
7
    F.aggregate("items", F.lit(0), lambda acc, x: acc + F.length(x)).alias("chars"),
8
)
9
# 문서 정의로 예상한 주문 1(["book", "pen"])의 결과: ["BOOK", "PEN"], ["book"], true, true, 7

F.filter는 배열 함수이고 DataFrame.filter는 행을 거르는 메서드입니다. 이름이 같아서 보기에서 섞이기 쉽습니다.

12-2. pivot

1
GroupedData.pivot(pivot_col, values=None)

groupBy 뒤에 pivot을 붙이고 agg로 값을 채웁니다. pivot_col의 값마다 결과 컬럼이 하나씩 생깁니다.

1
o.groupBy("user_id").pivot("status", ["PAID", "CANCELED"]).agg(F.sum("amount"))
2
# 문서 기준 예상 결과
3
# user_id | PAID  | CANCELED
4
# 101     | 32000 | 8000
5
# 102     | 15000 | null
6
# 103     | 54000 | null

values를 생략하면 결과 스키마를 정하려고 pivot_col의 고유값을 먼저 계산한다고 문서가 적습니다(eager computation). 트랜스포메이션을 정의하는 시점에 Job이 하나 더 돕니다. 값 목록을 알고 있으면 values로 넘겨 그 계산을 없앱니다. 목록에 없는 값(위 예제에서 status가 null인 주문 5)은 결과 컬럼에 들어가지 않습니다.

pivot은 GroupedData의 메서드라서 groupBy 없이 DataFrame에서 바로 부를 수 없습니다. 반대로 넓은 표를 긴 표로 되돌리는 메서드는 DataFrame.unpivot(ids, values, variableColumnName, valueColumnName)(3.4.0~, 별칭 melt)입니다. 문서는 집계까지 되돌리지는 못한다고 적습니다.

12-3. 샘플링

1
orders.sample(fraction=0.5, seed=42)        # 약 절반. 시드를 빼면 실행마다 행 수가 달라진다
2
orders.sample(0.5, 42)                      # 첫 위치 인자가 float면 fraction으로 받는다(문서 예제)
3
orders.sampleBy("status", fractions={"PAID": 0.5, "CANCELED": 1.0}, seed=7)
4
train, test = orders.randomSplit([8.0, 2.0], seed=7)   # [0.8, 0.2]로 정규화

샘플 행 수를 정확히 n개로 맞춰야 하면 orderBy(F.rand(seed)).limit(n)처럼 무작위로 정렬한 뒤 자릅니다. 이때는 전체 정렬 비용이 듭니다.

13. 확인 문제

직접 만든 문제입니다. 시험 문항이나 문제 은행에서 가져오지 않았습니다.

문제 1. 다음 코드의 결과 DataFrame 컬럼으로 맞는 것은?

1
df = (orders.select("order_id", "user_id", "amount")
2
            .withColumnRenamed("userid", "uid")
3
            .withColumn("amount", F.col("amount") * 2)
4
            .drop("amt"))

A. 오류가 난다. B. order_id, uid, amount C. order_id, user_id, amount D. order_id, user_id, amount, amount

문제 2. status, amount, items 중 status가 null인 행만 지우는 코드는?

A. df.na.drop() B. df.na.drop("all") C. df.na.drop(subset=["status"]) D. df.na.fill("x", subset=["status"])

문제 3. 다음 두 DataFrame에 대해 a.union(b).collect()의 결과로 맞는 것은?

1
a = spark.createDataFrame([(1, 101)], "order_id INT, user_id INT")
2
b = spark.createDataFrame([(102, 2)], "user_id INT, order_id INT")

A. [Row(order_id=1, user_id=101), Row(order_id=2, user_id=102)]
B. [Row(order_id=1, user_id=101), Row(order_id=102, user_id=2)]
C. 컬럼 순서가 달라 예외가 난다.
D. 중복이 제거되어 한 행만 남는다.

문제 4. 결제가 한 번도 없는 회원만 회원 컬럼 그대로 얻는 코드는?

A. users.join(payments, "user_id", "left") B. users.join(payments, "user_id", "left_anti") C. users.join(payments, "user_id", "left_semi") D. users.crossJoin(payments)

문제 5. 세션 시간대가 UTC일 때 F.from_unixtime(F.lit(1790812800))의 결과 타입과 값은?

A. TimestampType, 2026-10-01 00:00:00 B. StringType, ‘2026-10-01 00:00:00’ C. LongType, 1790812800 D. DateType, 2026-10-01

문제 6. 다음 코드에서 acc.value로 맞는 것은? (orders는 6행)

1
acc = spark.sparkContext.accumulator(0)
2
def tag(r):
3
    acc.add(1)
4
    return r
5
rdd2 = orders.rdd.map(tag)
6
print(acc.value)

A. 0 B. 6 C. 12 D. 오류

문제 7. 5GB 주문 테이블과 2MB 회원 테이블을 user_id로 조인합니다. 기본 설정에서 맞는 설명은?

A. 기본 임계값 10MB보다 작으므로 회원 테이블이 자동으로 broadcast될 수 있다.
B. broadcast 조인은 양쪽을 모두 셔플한다.
C. F.broadcast(orders)로 큰 쪽을 감싸야 한다.
D. broadcast 힌트는 임계값을 넘는 테이블에는 적용되지 않는다.

문제 8. w = Window.partitionBy("user_id").orderBy("ordered_at")일 때 F.sum("amount").over(w)가 회원 102의 두 주문(15000, 15000, 시각이 다름)에 주는 값은?

A. 30000, 30000 B. 15000, 30000 C. 15000, 15000 D. rowsBetween이 없어서 오류

문제 9. 다음 중 status 값 목록을 미리 계산하는 Job 없이 pivot하는 코드는?

A. o.groupBy("user_id").pivot("status").agg(F.sum("amount"))
B. o.groupBy("user_id").pivot("status", ["PAID", "CANCELED"]).agg(F.sum("amount"))
C. o.pivot("status").groupBy("user_id").sum("amount")
D. o.groupBy("user_id").agg(F.pivot("status"))

범위와 한계

• 시그니처와 기본값은 2026년 10월 19일에 연 PySpark 4.2.0 API 문서와 Spark 4.2.0 문서 기준입니다. 3.x와 다른 부분은 문서에 버전이 적힌 것만 표시했습니다(ANSI 기본 켜짐 4.0, size(null)의 -1과 null, try_to_date 4.0.0, try_to_timestamp 3.5.0, Column.try_cast 4.0.0, StructType.fromDDL 4.0.0, 컬럼 위치 정렬 4.0.0, UDF 호출 쪽 키워드 인자 4.0.0, 타입 힌트 udf의 벡터화 4.1.0, Arrow 기본 4.2, Python UDF Arrow 기본 4.2, countDistinct는 3.5.9 문서와 비교). 2026년 10월 20일에 2-3절, 4-2절, 6절, 8-3절, 9절을 다시 확인해 고쳤고 11절과 12절을 더했습니다. 시험의 기준 Spark 버전은 공개되지 않았습니다.
• 코드는 아직 실행하지 않았습니다. 이 노트북에 Java가 없어 PySpark를 띄울 수 없었습니다. “문서 기준 예상 결과”는 API 문서의 정의와 예제로 예상한 값이고 epoch 값은 Python datetime으로 계산했습니다.
• 2026년 10월 21일에 Spark 4.2.0 소스로 union의 타입 맞추기, 0으로 나누기, unix_timestamp의 ANSI 동작, left_semi의 중복 처리, 비 Arrow UDF의 반환 타입 불일치, 파일 소스의 nullable 처리, countDistinct 별칭을 확인해 본문에 반영했습니다. approx_count_distinct와 count_distinct의 셔플 비교도 문서 문장 없이 소스로만 확인한 내용입니다.
• 확인하지 못한 것은 Arrow 경로 UDF의 반환 타입 불일치 결과, 내장 함수가 UDF보다 낫다는 권고의 공식 문서 문장, 12-1절 고차 함수의 실제 결과값(실행하지 않음)입니다.

참고 자료

시험

• Databricks, Exam Guide, 2026-10-08판, Exam Guide, Oct 2025 (Section 3 항목과 샘플 문항 3, 4, 6번)

PySpark 4.2.0 API

• DataFrame: withColumn, withColumns, drop, withColumnRenamed, withColumnsRenamed, filter, distinct, dropDuplicates, dropna, fillna, describe, summary, groupBy, join, crossJoin, union, unionAll, unionByName, sort, collect, head, tail, toLocalIterator, toPandas, printSchema, hint
• functions: split, explode, explode_outer, count, count_distinct, approx_count_distinct, mean, from_unixtime, unix_timestamp, to_date, try_to_date, date_format, dayofweek, timestamp_seconds, broadcast, udf, pandas_udf
• 그 밖: GroupedData.agg, DataFrameNaFunctions.fill, StructType, StructField, DataFrameReader.schema, SparkSession.createDataFrame, UDFRegistration.register, SparkContext.broadcast, SparkContext.accumulator, Broadcast, Accumulator
• 2026-10-20 추가 확인: functions.size, functions.try_to_timestamp, Column.try_cast, DataFrame.foreach, DataFrame.foreachPartition, GroupedData.applyInPandasWithState, GroupedData.transformWithStateInPandas, Window, functions.row_number, functions.rank, functions.dense_rank, functions.lag, functions.transform, functions.filter, functions.exists, functions.forall, functions.aggregate, GroupedData.pivot, DataFrame.unpivot, DataFrame.sample, DataFrame.sampleBy, DataFrame.randomSplit
• 3.5.9 비교: functions.countDistinct (3.5.9), functions.unix_timestamp (3.5.9)

Apache Spark 4.2.0 문서

• NULL Semantics, Parquet Files, Column.eqNullSafe, JOIN, ORDER BY
• ANSI Compliance, SQL Migration Guide, PySpark Migration Guide
• Performance Tuning, RDD Programming Guide
• Spark SQL, Built-in Functions: Collection Functions (size, cardinality), Aggregate Functions (approx_count_distinct)
• Spark 3.5.9 비교: Spark SQL, Built-in Functions (3.5.9) (size, try_divide)
• Structured Streaming: APIs on DataFrames and Datasets (State Store 절)

Apache Spark 4.2.0 소스

• sql/core/…/aggregate/AggUtils.scala (planAggregateWithOneDistinct)
• sql/core/…/window/WindowExecBase.scala (파티션 없는 윈도 경고)
• sql/catalyst/…/analysis/TypeCoercionBase.scala (WidenSetOperationTypes), AnsiStringPromotionTypeCoercion.scala, TypeCoercion.scala (union의 타입 맞추기)
• sql/catalyst/…/expressions/arithmetic.scala (DivModLike), datetimeExpressions.scala (UnixTimestamp의 failOnError)
• sql/core/…/joins/HashJoin.scala (semiJoin), sql/core/…/datasources/DataSource.scala (resolveRelation의 asNullable), sql/core/…/python/EvaluatePython.scala (makeFromJava)
• python/pyspark/sql/functions/builtin.py (countDistinct)