6. What’s New in Spark 4.2.0?

CDC·PySpark·Data Source V2·Streaming·Spark Connect 변화

Haram Lee
2026-08-16
studies / Topics / Spark

#spark #distributed-systems #data-engineering

Apache Spark 4.2.0은 2026년 7월 14일 공개된 Spark 4.x 계열의 세 번째 기능 릴리스
1,700개가 넘는 JIRA 이슈가 반영됨
Spark Core의 Driver–Executor 실행 모델을 교체한 버전은 아님
기존 분산 실행 구조 위에서 다음 영역을 확장함
- SQL과 데이터 타입
- Change Data Capture
- Data Source V2
- PySpark 실행 성능
- Structured Streaming 안정성
- Spark Connect
- Kubernetes 운영
- Spark UI와 Observability

flowchart TD
  release["Spark 4.2.0"] --> core["Spark Core"]
  core --> runtime["Driver / Executor"]
  core --> scheduling["Job / Stage / Task"]
  core --> partition["Partition"]
  core --> shuffle["Shuffle"]
  release --> capabilities["New Capabilities"]
  capabilities --> cdc["CDC"]
  capabilities --> dsv2["Data Source V2 Transactions"]
  capabilities --> arrow["Arrow-based PySpark"]
  capabilities --> streaming["Streaming Reliability"]
  capabilities --> connect["Spark Connect Expansion"]
  capabilities --> k8s["Kubernetes Improvements"]
  capabilities --> ui["Modernized Spark UI"]

6.1 Change Data Capture Support

Change Data Capture란?

CDC는 테이블 전체를 매번 다시 읽는 대신, 원본 데이터에서 발생한 행 단위 변경 사항을 읽는 방식임.

text

Original Table

id | name | status
1  | A    | ACTIVE
2  | B    | ACTIVE

다음 변경이 발생했다고 하자.

text

INSERT
id=3, name=C, status=ACTIVE

UPDATE
id=1, status=INACTIVE

DELETE
id=2

CDC는 전체 현재 상태뿐 아니라 다음과 같은 변경 이벤트를 제공함.

text

Change Feed

INSERT  | id=3 | C | ACTIVE
UPDATE  | id=1 | A | INACTIVE
DELETE  | id=2 | B | ACTIVE

기존 방식

Spark에서 CDC 데이터를 처리하려면 일반적으로 외부 시스템이 변경 이벤트를 만들어야 했음.

flowchart TD
  source["Source Database"] -- "Binlog / WAL" --> connector["CDC Connector"]
  connector --> events["Kafka / Object Storage"]
  events --> streaming["Spark Streaming"]
  streaming --> processing["Deduplication / Merge / Upsert"]
  processing --> target["Target Table"]

Spark 자체는 전달된 변경 이벤트를 일반적인 DataFrame으로 읽고, 사용자가 직접 변경 유형과 순서를 해석했음.

text

changes=spark.readStream.format("kafka").load()result= (changes.select(...).filter(...)# 직접 deduplication, upsert 처리
)

Spark 4.2.0의 변화

Spark 4.2.0은 Data Source V2에 CDC Connector API를 추가하고 다음 인터페이스를 제공함.

SQL CHANGES Clause
DataFrame API
PySpark API
Spark Connect API
Batch CDC Read
Streaming CDC Read
Row-level Change Post-processing

text

CDC-capable Table
        ↓
Data Source V2 Changelog API
        ↓
Spark SQL / DataFrame
        ↓
Batch or Streaming DataFrame

예를 들어 PySpark에서는 CDC를 지원하는 테이블을 다음과 같은 형태로 읽을 수 있음.

text

changes= (spark.read.option("startingVersion","10").changes("catalog.database.orders")
)

Streaming으로도 동일한 형태의 API를 사용함.

text

changes= (spark.readStream.option("startingVersion","10").changes("catalog.database.orders")
)

중요한 한계

Spark가 직접 MySQL Binlog나 PostgreSQL WAL을 자동으로 읽어주는 것은 아님.

text

Spark 4.2 CDC의 의미

❌ 모든 Database의 변경 로그를 자동 수집

⭕ CDC를 제공하는 Data Source Connector와
   Spark가 통신할 수 있는 표준 인터페이스 제공

즉, 실제 구조는 다음과 같음.

text

Source System
      ↓
CDC-capable Connector / Table Format
      ↓ loadChangelog()
Spark
      ↓
Standard CDC DataFrame

Auto CDC와 SCD Type 1

Spark 4.2.0은 Spark Declarative Pipelines에 Auto CDC도 추가함.

대표적인 활용은 SCD Type 1임.

text

Incoming Changes

id=1, name="Rami", team="Data"
id=1, name="Rami", team="Platform"

SCD Type 1에서는 과거 값을 별도로 남기지 않고 최신 값으로 덮어씀.

text

Target Table

id | name | team
1  | Rami | Platform

기존에는 사용자가 직접 다음 로직을 구현해야 했음.

text

Read Changes
→ Order by Sequence
→ Remove Duplicates
→ Match Target Rows
→ Insert or Update
→ Manage Checkpoint

Auto CDC에서는 Declarative Pipeline에 변경 처리 규칙을 선언하고, Spark가 SCD Type 1 Streaming Write를 구성함. Spark 4.2.0은 이를 위한 Python 및 Spark Connect API도 추가함.

의미

text

Before

Raw Change Events
→ Application-specific CDC Logic
→ Target

text

Spark 4.2.0

CDC-capable Source
→ Standard Changelog API
→ Declarative Change Processing
→ Target

Spark 4.2.0 does not become a database log collector. It provides a standard execution model for reading and processing changes exposed by CDC-capable Data Source V2 connectors.

6.2 PySpark and Arrow Optimizations

PySpark에는 기본적으로 두 개의 실행 환경이 존재함.

text

Python Process
      ↕
JVM Process
      ↓
Spark Core
      ↓
Executors

Spark의 Scheduler와 대부분의 실행 엔진은 JVM에서 동작하지만, 사용자가 작성한 Python 함수는 Python Worker에서 실행됨.

기존 데이터 교환

기존 PySpark에서는 JVM 객체와 Python 객체 사이에서 데이터 형식을 변환해야 했음.

text

JVM Row
   ↓ Serialization
Python Object
   ↓ Python Function
Python Object
   ↓ Serialization
JVM Row

레코드 단위 객체 변환은 다음 비용을 만들 수 있음.

Serialization
Deserialization
JVM–Python Process Communication
Python Object 생성
Garbage Collection

text

Row 1 → Serialize → Python → Deserialize
Row 2 → Serialize → Python → Deserialize
Row 3 → Serialize → Python → Deserialize

Apache Arrow

Arrow는 데이터를 언어별 객체로 반복 변환하는 대신, 여러 시스템이 함께 읽을 수 있는 컬럼형 메모리 표현을 제공함.

text

Columnar Batch

id     [1, 2, 3, 4]
amount [100, 200, 300, 400]
status [A, A, B, A]

text

JVM
  │
  │ Arrow Columnar Batch
  ▼
Python

Spark 4.2.0의 변화

Spark 4.2.0에서는 다음 기능이 기본적으로 활성화됨.

Arrow-optimized Python UDF
PySpark와 JVM 사이의 Arrow 기반 Columnar Data Exchange

text

Before

JVM Rows
→ Object Serialization
→ Python Objects

text

Spark 4.2.0 Default

JVM Columnar Batch
→ Apache Arrow
→ Python Columnar Batch

Spark 4.2부터 PySpark와 JVM 사이의 컬럼형 데이터 교환에 Arrow가 기본으로 사용되며, Arrow 최적화 Python UDF도 기본 활성화됨.

Python UDF 실행

text

frompyspark.sql.functionsimportudf@udf("long")defmultiply_by_two(value):returnvalue*2

개념적으로 기존 실행은 다음과 같음.

text

Executor JVM
    ↓ Rows
Python Worker
    ↓ Python UDF
Executor JVM

Arrow를 사용하면:

text

Executor JVM
    ↓ Columnar Batch
Python Worker
    ↓ Arrow-optimized UDF
Executor JVM

형태로 데이터 교환 비용을 낮출 수 있음.

Grouped Aggregation UDF

Spark 4.2.0은 Arrow 및 Pandas Grouped Aggregation UDF를 위한 Iterator API도 추가함.

text

Group 1 Batch ─┐
Group 2 Batch ─┼→ Iterator-based UDF
Group 3 Batch ─┘

Iterator 방식에서는 전체 입력을 한꺼번에 Python 메모리로 가져오기보다 Batch 단위로 처리할 수 있음.

text

defaggregate_batches(iterator):forbatchiniterator:yieldprocess(batch)

Spark 4.2.0은 Arrow와 Pandas의 Grouped Aggregation UDF를 위한 Iterator API 및 SQL 등록 기능을 추가함.

추가적인 Python 개선

Python UDF Processing Time Metric 추가
Pandas API on Spark의 describe() Job 수를 컬럼 수에 비례하는 구조에서 한 번의 Job으로 줄임
Pandas UDF의 Nullable Integer 및 Extension Type 지원 개선
Python Data Source Streaming에서 Admission Control과 AvailableNow 지원

이 중 Python UDF Processing Time Metric은 Python 함수에서 실제로 소비된 시간을 확인하는 데 도움을 줌.

항상 빨라지는 것은 아님

Arrow는 주로 JVM과 Python 사이의 데이터 교환 비용을 줄임.

text

Arrow가 개선하는 영역

JVM ↔ Python Serialization
Columnar Data Exchange
Python UDF Input / Output

다음 비용까지 자동으로 없애는 것은 아님.

text

여전히 남는 비용

Shuffle
Network I/O
Disk I/O
Data Skew
느린 Python 함수 자체
잘못된 Join Strategy
지나치게 많은 Partition

text

Total Execution Time

Data Scan
+ Shuffle
+ JVM Execution
+ JVM–Python Exchange
+ Python Computation

Arrow는 이 가운데 JVM–Python Exchange를 주로 개선함.

Arrow makes Python execution less expensive, but it does not remove the distributed-system costs of shuffle, skew, scheduling, and data movement.

6.3 Data Source V2 Improvements

Data Source란?

Data Source는 Spark가 외부 시스템의 데이터를 읽고 쓰는 인터페이스임.

text

Spark
  ↕
Data Source Connector
  ↕
Parquet / JDBC / Iceberg / Kafka / Custom Storage

사용자 입장에서는 다음과 같은 API로 나타남.

text

df= (spark.read.format("custom-source").option("...").load()
)

Data Source V1

초기의 Data Source V1은 비교적 단순한 읽기·쓰기 인터페이스에 집중함.

text

Spark
→ Read Data
→ Write Data

하지만 Catalog, Transaction, Row-level Operation, Streaming, Schema Evolution 같은 기능을 일관된 방식으로 확장하기 어려웠음.

Data Source V2

Data Source V2는 기능을 여러 Capability Interface로 나눔.

text

Data Source V2

├─ Catalog
├─ Table
├─ Scan
├─ Batch
├─ Streaming
├─ Write
├─ Row-level Operations
├─ Statistics
└─ Pushdown

Connector는 자신이 지원하는 기능을 선택적으로 구현할 수 있음.

text

Connector A
├─ Filter Pushdown
├─ Column Pruning
└─ Batch Read

Connector B
├─ Batch Read
├─ Streaming Read
├─ Transactions
├─ CDC
└─ Schema Evolution

Transaction Management

Spark 4.2.0은 Data Source V2 Transaction Management를 도입함.

기존에는 여러 쓰기 연산이 서로 독립적으로 Commit될 수 있었음.

text

Write Operation A → Commit 성공
Write Operation B → 실패

그 결과 일부 변경만 적용된 상태가 남을 수 있음.

text

Expected

A + B 모두 반영
또는
A + B 모두 취소

Transaction을 지원하는 Connector에서는 여러 연산을 하나의 Transaction으로 묶을 수 있음.

text

Begin Transaction
├─ Operation A
├─ Operation B
└─ Operation C
        ↓
    Commit All

실패 시:

text

Begin Transaction
├─ Operation A
├─ Operation B → Failure
└─ Abort Transaction

Spark 4.2.0의 Data Source V2 Transaction Management는 Connector가 여러 작업을 원자적으로 Commit할 수 있는 기반을 제공함. 실제 보장 수준은 Connector의 구현에 따라 달라짐.

Schema Evolution for INSERT

데이터를 적재할 때 Source와 Target Schema가 달라질 수 있음.

text

Target

id       LONG
name     STRING

text

Source

id       LONG
name     STRING
region   STRING

기존에는 Target Table을 먼저 변경하거나 Schema를 수동으로 맞춰야 했음.

text

ALTER TABLE
→ Schema 변경
→ INSERT

Spark 4.2.0은 Data Source V2 INSERT에 Schema Evolution 기능을 확장함.

text

INSERTINTO targetWITHSCHEMA EVOLUTIONSELECT*FROM source;

지원 Connector에서는 Source의 필드 변화에 맞춰 Target Schema Evolution을 처리할 수 있음. Spark 4.2.0은 Source에 더 적은 Column이 있는 경우와 INSERT INTO … WITH SCHEMA EVOLUTION 문법도 지원함.

Partition Statistics Filtering

Partitioned Table에는 Partition별 통계가 존재할 수 있음.

text

Table

dt=2026-07-01
├─ min_id=1
└─ max_id=1000

dt=2026-07-02
├─ min_id=1001
└─ max_id=2000

다음 쿼리가 있다고 하자.

text

SELECT*FROM ordersWHERE id>1500;

Partition Statistics를 이용하면 조건에 맞을 가능성이 없는 Partition을 제외할 수 있음.

text

dt=2026-07-01 → max_id=1000 → Skip
dt=2026-07-02 → max_id=2000 → Read

text

Without Filtering

All Partitions
→ Read
→ Filter

text

With Partition Statistics Filtering

Partition Statistics
→ Candidate Partitions
→ Read

Spark 4.2.0은 Data Source V2의 Partition Statistics Filtering과 Runtime Filtering을 확장함.

Operation Metrics

Connector가 수행한 Read와 Write 작업에 대한 세부 Metric도 확장됨.

text

Connector Operation

├─ Rows Read
├─ Bytes Read
├─ Rows Written
├─ Files Created
├─ Commit Time
└─ Custom Metrics

이를 통해 Spark UI와 Observability 시스템에서 Connector 내부 동작을 더 구체적으로 확인할 수 있음.

Data Source V2와 CDC의 관계

Spark 4.2.0의 CDC도 Data Source V2 위에서 구현됨.

text

Data Source V2
├─ Catalog
├─ Transactions
├─ Schema Evolution
├─ Partition Statistics
└─ Changelog / CDC

즉, CDC와 Transaction은 별개의 임시 기능이 아니라 Spark가 Connector를 확장하는 공통 구조에 포함됨.

Data Source V2 is becoming the contract through which Spark understands not only how to read data, but also how to manage tables, transactions, schemas, changes, and connector-level metrics.

6.4 Structured Streaming Enhancements

flowchart TD
  streaming["Structured Streaming"] --> input["Input Stream"]
  input --> trigger["Micro-batch / Real-time Trigger"]
  trigger --> operations["Stateful Operations"]
  operations --> stateStore["State Store"]
  stateStore --> sink["Sink"]
  sink --> checkpoint["Checkpoint"]

Stream–Stream Join in Update Mode

두 개의 Stream을 Join하면 양쪽에서 데이터가 계속 들어오기 때문에 이전 Row를 State로 보관해야 함.

text

Orders Stream ─┐
               ├→ Stream–Stream Join
Payments Stream┘

text

State Store

Orders State
Payments State

Spark 4.2.0은 Non-outer Stream–Stream Join을 Update Output Mode에서 사용할 수 있도록 확장함. 또한 Stream–Stream Join State Format V4가 추가됨.

Stable Source and Sink Naming

Streaming Query는 Checkpoint를 통해 Source Offset과 Sink Commit 상태를 기억함.

기존 구조에서는 Source의 순서가 중요한 식별 정보로 사용될 수 있었음.

text

Source 0 → Kafka A
Source 1 → Kafka B

코드에서 Source 순서를 바꾸면:

text

Source 0 → Kafka B
Source 1 → Kafka A

기존 Checkpoint와 새 Query의 Source가 잘못 대응될 위험이 있음.

Spark 4.2.0은 Source에 안정적인 이름을 부여할 수 있는 API를 추가함.

text

events= (spark.readStream.format("kafka").name("order-events").load()
)

개념적으로:

text

Before

Source Index 0
Source Index 1

text

Spark 4.2.0

Source Name: order-events
Source Name: payment-events

Stable Name을 사용하면 Source를 추가·제거하거나 순서를 바꿀 때 Checkpoint가 Source를 더 안정적으로 식별할 수 있음.

Sink에도 이름을 부여할 수 있으며, Sink Name은 새로운 Commit Log에 저장됨.

State Store Reliability

Stateful Streaming 연산은 이전 Batch의 정보를 State Store에 저장함.

text

Batch 1
→ State v1

Batch 2
→ Read State v1
→ Update
→ State v2

State Store가 손상되면 Query를 계속 실행하기 어려울 수 있음.

Spark 4.2.0은 다음 기능을 추가함.

State Store Snapshot 자동 복구
State Store Row Checksum
Snapshot Upload가 지연될 때 추가 Snapshot Trigger
일관되지 않은 Checkpoint Metadata 감지
RocksDB 및 HDFS State Store 안정성 개선

text

State Store Write
       ↓
Checksum
       ↓
Snapshot
       ↓
Corruption Detection / Repair

이전에는 손상된 상태가 나중에 모호한 오류로 나타날 수 있었다면, 4.2.0에서는 Checksum과 Metadata 검증으로 문제를 더 일찍 발견하고 일부 Snapshot을 자동 복구할 수 있음.

Python Streaming Improvements

Spark 4.2.0은 Python Data Source Streaming Reader에 다음 기능을 추가함.

Admission Control
Trigger.AvailableNow
PySpark Real-time Mode Trigger

Admission Control은 Source가 한 번의 Trigger에 지나치게 많은 데이터를 전달하지 않도록 조절하는 기능임.

text

Source Backlog
████████████████████

Without Admission Control
→ 한 번에 대량 처리
→ 긴 Batch / Memory Pressure

With Admission Control
→ 처리 가능한 양만 입력
→ 다음 Trigger에서 계속 처리

AvailableNow는 Query 시작 시점까지 도착한 데이터를 처리한 뒤 종료하는 방식임.

text

Current Backlog
      ↓
Process All Available Data
      ↓
Stop Query

Real-time Mode 자체가 4.2.0에서 처음 등장한 것은 아니며, 4.2.0에서는 이를 PySpark에서도 사용할 수 있도록 Trigger 지원이 확장됨.

6.5 Spark Connect Expansion

Spark Connect의 기본 구조는 다음과 같음.

flowchart TD
  client["Client Application"]
  client -- "Unresolved Logical Plan · gRPC" --> server["Spark Connect Server"]
  server --> analyzer["Analyzer"]
  analyzer --> optimizer["Optimizer"]
  optimizer --> physical["Physical Plan"]
  physical --> core["Spark Core"]
  core --> executors["Executors"]

Spark Connect는 Spark 3.4에서 도입된 Client–Server Architecture이며, Client와 Driver를 분리하고 DataFrame의 Logical Plan을 Protocol로 사용함.

Spark 4.2.0의 방향

Spark 4.2.0에서는 Spark Connect를 단순한 원격 DataFrame API에서 운영 가능한 원격 실행 인터페이스로 확장함.

text

Earlier Spark Connect

Client
→ Submit Logical Plan
→ Receive Result

text

Spark 4.2.0

Client
├─ Submit Plan
├─ Check Execution Status
├─ Release Session
├─ Send Code Location
└─ Inspect History

GetStatus API

Client와 Server 양쪽에 Execution Status 조회 API가 추가됨.

text

Client
  │ GetStatus
  ▼
Connect Server
  │
  ├─ Running
  ├─ Completed
  ├─ Failed
  └─ Cancelled

원격 Application에서 실행 상태를 Polling하거나 UI에 표시하기 쉬워짐.

History Server Integration

Spark History Server에 Spark Connect Tab이 추가됨.

text

Spark History Server

├─ Jobs
├─ Stages
├─ SQL
├─ Executors
└─ Spark Connect

종료된 Connect Session과 원격 실행 정보를 History Server에서 분석할 수 있는 기반이 마련됨.

Session Lifecycle

Spark Connect Client Process가 종료돼도 서버 Session이 남으면 자원이 누적될 수 있음.

text

Client 종료
      ↓
Server Session 유지
      ↓
Resource Leak 가능

Spark 4.2.0은 설정에 따라 Client Process 종료 시 Remote Session을 해제할 수 있도록 함.

text

Client Exit
      ↓
Release Remote Session
      ↓
Server Resource Cleanup

Code Location and Telemetry

Client 코드의 위치를 Server에 전달할 수 있게 됨.

text

Client Code

orders.py:42
df.groupBy(...).count()

text

Logical Plan
+ Client Code Location
        ↓
Server Logging / Telemetry

Server Log만 보더라도 어떤 Client 코드가 Action을 실행했는지 추적하기 쉬워짐.

Discussion