[Spark Ch-06] Spark Connect와 pandas API on Spark: 클라이언트는 계획만 보내고 pandas 코드는 분산으로 돈다

결론부터

1. Spark Connect(3.4에서 도입)는 클라이언트가 DataFrame 연산을 해석 전 논리 계획(unresolved logical plan)으로 만들어 protocol buffers로 인코딩하고 gRPC로 서버에 보내는 구조입니다. 서버가 분석, 최적화, 실행을 하고 결과는 Arrow로 인코딩한 행 배치로 돌려줍니다.
2. 클라이언트가 드라이버 프로세스 밖에 있어서 SparkContext, RDD API, 정적 설정, df._jdf 같은 JVM 내부 접근을 쓸 수 없습니다. 접속 주소는 sc://host:port 형식이고 기본 포트는 15002입니다.
3. 배포 모드는 드라이버가 어디서 뜨는지로 나눕니다. client(기본)는 제출한 쪽, cluster는 클러스터 안, local은 드라이버와 익스큐터가 한 JVM에서 돕니다. local[*]는 논리 코어 수만큼 스레드를 씁니다.
4. pandas API on Spark(import pyspark.pandas as ps)는 pandas 문법으로 Spark에서 분산 실행합니다. Spark DataFrame에서 바꾸면 기본 인덱스가 새로 붙고 기본 종류는 distributed-sequence입니다. to_pandas()는 모든 데이터를 드라이버로 모으니 작은 결과에만 씁니다.
5. pandas UDF는 pandas_udf와 타입 힌트로 종류를 정합니다. Series에서 Series, Series 반복자에서 Series 반복자, Series에서 스칼라(집계)가 있습니다. DataFrame 단위로 함수를 적용할 때는 mapInPandas와 groupBy().applyInPandas()를 씁니다. Arrow 변환(spark.sql.execution.arrow.pyspark.enabled)은 4.2에서 기본 true가 됐고 3.5, 4.0, 4.1에서는 기본 false였습니다.

1. 시험 안내서 항목과 이 글의 순서

안내서 샘플 문항 7번이 6영역에서 나왔습니다. “모든 익스큐터가 워커 노드 하나에서 돌아야 하는 배포 모드는 무엇인가”이고 정답은 Local 모드입니다. 두 영역을 합쳐 10%라서 문항 수로는 4~5개 정도로 짐작합니다(45문항 × 10%로 계산한 값이고 실제 배분은 공개되지 않았습니다).

예제 데이터는 쇼핑몰 주문(orders)과 결제(payments)입니다.

2. Spark Connect의 구조

2-1. 요청이 오가는 순서

Spark Connect Overview 문서의 “How Spark Connect works” 절을 단계로 나누면 이렇습니다.

1. 클라이언트 라이브러리가 DataFrame 연산을 해석 전 논리 계획으로 바꾸고 protocol buffers로 인코딩합니다.
2. 이 계획을 gRPC로 서버에 보냅니다.
3. Spark 서버에 붙은 Spark Connect 엔드포인트가 해석 전 계획을 Spark의 논리 계획 연산자로 옮깁니다. 문서는 SQL을 파싱해 처음 파스 계획을 만드는 것과 비슷하다고 설명합니다.
4. 그다음은 보통의 Spark 실행 과정이라 옵티마이저와 실행 엔진의 최적화를 그대로 씁니다.
5. 결과는 Apache Arrow로 인코딩한 행 배치로 gRPC 스트림을 통해 클라이언트로 돌아옵니다.

해석 전 계획을 보낸다는 점이 시험 보기를 가르는 기준입니다. 클라이언트는 컬럼 이름이 실제로 있는지 모른 채 계획을 보내고 이름 확인(analysis)은 서버가 합니다. 4.2.0 문서의 “Eager vs Lazy: Spark Connect vs Spark Classic”이 이 차이를 예로 듭니다. spark.sql("select 1 as a, 2 as b").filter("c > 1")은 Classic에서는 그 줄에서 컬럼 c가 없다는 오류를 냅니다. Connect에서는 그 줄은 통과하고 df.columns나 df.show()를 부를 때 오류가 납니다. 스키마 조회, explain, persist, 액션처럼 해석된 계획이 필요한 연산이 서버에 분석 요청을 보내기 때문입니다. df["c"]와 df.c로 컬럼을 꺼낼 때 이름을 바로 검사하던 Connect Python 클라이언트 동작은 각각 4.1과 4.2에서 없어졌습니다(PySpark 마이그레이션 가이드).

2-2. 클래식 PySpark와 비교

세션 타입은 문서 예제로 확인할 수 있습니다.

1
>>> type(spark)
2
<class 'pyspark.sql.connect.session.SparkSession'>

2-3. 운영에서 얻는 것

문서가 드는 이점은 셋입니다.

여러 사용자가 클러스터 하나를 함께 쓸 때(멀티테넌트) 생기는 문제를 줄인다는 맥락입니다. Databricks의 서버리스 컴퓨트와 Free Edition도 Spark Connect로 돌아서 Ch-00에서 Free Edition에서는 RDD API를 쓸 수 없다고 적었습니다.

3. Spark Connect의 API 지원 범위

문서의 “How Spark Connect client applications differ from classic Spark applications”와 “What is supported” 절을 합쳤습니다.

문서의 PySpark 항목은 3.4부터 DataFrame, Functions, Column 등 PySpark API 대부분을 지원한다고 적습니다. UDF와 Streaming API 대부분(DataStreamReader, DataStreamWriter, StreamingQuery, StreamingQueryListener)을 지원한다는 문장은 같은 절의 Scala 항목(3.5부터)에 있습니다. Python 쪽 범위는 PySpark API 참조 첫 화면의 안내문이 정리합니다. Spark SQL, pandas API on Spark, Structured Streaming, MLlib(DataFrame 기반)가 Spark Connect를 지원한다는 문장이고 4.0.0 문서부터 있습니다(3.5.6 문서에는 없습니다). API 문서는 함수마다 “Supports Spark Connect” 표시를 달아서 옮기기 전에 확인하라고 안내합니다.

RDD에 기대던 코드는 DataFrame 함수로 바꿔야 합니다.

1
# 클래식에서만 되는 코드
2
n = orders.rdd.getNumPartitions()
3
sc = spark.sparkContext
4
acc = sc.accumulator(0)
5
bc = sc.broadcast({"PAID": 1, "CANCELLED": 0})
6

7
# Connect에서도 되는 대안
8
from pyspark.sql import functions as F
9
part_sizes = orders.groupBy(F.spark_partition_id().alias("pid")).count()
10
status_map = spark.createDataFrame([("PAID", 1), ("CANCELLED", 0)], "status string, flag int")
11
flagged = orders.join(F.broadcast(status_map), "status", "left")

accumulator와 broadcast 변수도 Connect에서 막힙니다. PySpark에서 accumulator()와 broadcast()는 SparkContext의 메서드이고 Connect 세션은 spark.sparkContext에 접근하면 JVM_ATTRIBUTE_NOT_SUPPORTED 오류를 냅니다(4.2.0 소스 pyspark/sql/connect/session.py의 __getattr__). 문서에 이 둘을 따로 적은 문장은 없고 소스로 확인했습니다.

헷갈리는 지점입니다.

• spark.sparkContext.setLogLevel()도 SparkContext를 거쳐서 Connect에서 쓸 수 없습니다.
• 스트리밍의 foreachBatch는 Connect에서 함수가 바깥 변수에 접근하지 못한다고 API 문서가 적습니다(Ch-05 6-2).
• Spark Connect 자체에는 인증이 들어 있지 않습니다. 문서는 gRPC HTTP/2 인터페이스 앞에 인증 프록시를 두는 방식을 설명합니다. 접속 문자열의 token은 bearer 토큰을 보내는 자리입니다.

4. Spark Connect에 연결하기

4-1. 서버 띄우기

1
# 내려받은 Spark 배포판 폴더에서
2
./sbin/start-connect-server.sh
3
# 서버는 localhost:15002에서 받는다

서버 포트 설정은 spark.connect.grpc.binding.port(기본 15002, 3.4.0)입니다. 문서는 클라이언트 패키지 버전을 서버 Spark 버전과 맞추라고 적습니다.

4-2. 클라이언트 세 가지 방법

1
from pyspark.sql import SparkSession
2

3
spark = (SparkSession.builder
4
         .remote("sc://localhost:15002")
5
         .appName("ch06-connect")
6
         .getOrCreate())
7

8
orders = spark.read.parquet("/data/orders")
9
orders.groupBy("status").count().show()

PySpark Quickstart: Spark Connect 노트북은 원격 세션을 만들기 전에 기존 일반 세션을 멈춥니다. 같은 프로세스에서 일반 세션과 원격 세션이 함께 있을 수 없다고 적습니다.

독립 애플리케이션용으로는 클라이언트만 담은 패키지가 있습니다. 4.2.0 문서는 pip install pyspark-client==4.2.0을 안내합니다. PyPI의 릴리스 목록에서 가장 이른 버전은 4.0.0(2025년 5월 23일 업로드)이라 3.5에는 이 패키지가 없습니다.

4.0.0에서 생긴 spark.api.mode(기본 classic)를 connect로 두면 Spark Classic 애플리케이션이 로컬 Spark Connect 서버를 자동으로 띄워 붙습니다(Configuration 문서의 설명). 같은 4.0.0에서 서버 바인딩 주소 설정 spark.connect.grpc.binding.address(기본 없음)도 생겼습니다. 시험 기준 버전이 3.5라면 둘 다 범위 밖입니다.

4-3. 접속 문자열

Spark 저장소의 클라이언트 접속 문자열 문서 기준입니다. 형식은 sc://host:port/;param1=value;param2=value이고 스킴은 sc://로 고정, 경로는 비워야 하며 매개변수 이름은 대소문자를 구분합니다.

1
sc://localhost
2
sc://spark.example.com:443/;token=ABCDEFG;user_id=analyst

5. 배포 모드와 master URL

5-1. client, cluster, local

Cluster Mode Overview의 용어표는 배포 모드(deploy mode)를 “드라이버 프로세스가 어디서 뜨는지 구분하는 것”으로 정의합니다.

spark-submit --deploy-mode 기본값은 client이고 설정으로는 spark.submit.deployMode(기본 client)입니다. Submitting Applications 문서는 게이트웨이 머신이 워커와 같은 곳에 있으면 client 모드가 맞다고 봅니다. 노트북처럼 워커에서 먼 곳에서 제출하면 cluster 모드가 흔하다고 적습니다. 같은 문서가 Standalone 클러스터는 Python 애플리케이션의 cluster 모드를 지원하지 않는다고 적습니다.

local 모드는 엄밀히 말하면 --deploy-mode 값이 아니고 master URL을 local로 준 실행 방식입니다. 시험 안내서는 Client, Cluster, Local을 나란히 배포 모드로 부르므로 시험 맥락에서는 셋을 같이 외웁니다.

5-2. master URL

1
# 로컬 4스레드로 주문 집계 스크립트 실행
2
./bin/spark-submit --master "local[4]" order_report.py
3

4
# YARN cluster 모드
5
./bin/spark-submit --master yarn --deploy-mode cluster \
6
  --executor-memory 4G --num-executors 10 order_report.py

헷갈리는 지점입니다.

• 로컬 모드에서 spark.default.parallelism은 머신의 코어 수입니다. local[2]로 띄우면 스레드가 둘이라 Task도 동시에 둘만 돕니다.
• sc://host:port는 master URL 표에 없습니다. Submitting Applications 문서의 master URL은 클러스터 매니저를 고르는 값이고 sc://는 --remote나 builder.remote()에 주는 Spark Connect 서버 주소입니다. 보기에 --master sc://...가 나오면 섞어 놓은 오답으로 봅니다.
• Spark Connect 서버도 하나의 Spark 드라이버 애플리케이션입니다. start-connect-server.sh에 --master를 주면 그 서버가 어느 클러스터에서 돌지 정합니다. 4.2.0 스크립트는 받은 인자를 spark-daemon.sh submit으로 넘겨 spark-submit으로 서버 클래스를 띄우고 --help에는 spark-submit 옵션 목록을 보여 줍니다(소스 sbin/start-connect-server.sh).
• client 모드에서 드라이버가 죽으면 애플리케이션 전체가 끝납니다. cluster 모드의 Standalone에는 --supervise(spark.driver.supervise, 기본 false)로 0이 아닌 종료 코드일 때 드라이버를 자동 재시작하는 옵션이 있습니다.

6. pandas API on Spark의 장점

PySpark 문서 첫 화면의 설명을 정리하면 장점은 넷입니다.

• pandas 워크로드를 여러 노드에 분산해서 데이터 크기에 맞게 키울 수 있습니다.
• pandas에 익숙하면 코드를 거의 바꾸지 않고 옮길 수 있습니다.
• 코드베이스 하나로 pandas(테스트, 작은 데이터셋)와 Spark(운영, 분산 데이터셋)를 모두 다룹니다.
• pandas API와 pandas API on Spark 사이를 쉽게 오갑니다.

같은 문서는 Spark를 처음 쓰거나 어떤 API를 쓸지 고르는 중이라면 PySpark DataFrame을 권한다고도 적습니다. pandas API on Spark는 pandas에서 넘어오는 길을 쉽게 하려는 도구라는 위치입니다.

1
import pyspark.pandas as ps
2

3
psdf = ps.read_parquet("/data/orders")
4
paid = psdf[psdf["status"] == "PAID"]
5
by_store = paid.groupby("store_id")["amount"].sum().sort_values(ascending=False)
6
print(by_store.head(10))

pandas와 다른 점 세 가지를 같이 기억합니다.

모든 Spark 기능(History Server, Web UI, 배포 모드)을 그대로 쓰고 이미 떠 있는 SparkSession이 있으면 그 세션을 자동으로 씁니다. Arrow 같은 설정은 세션을 만들 때 넣습니다.

Best Practices 문서가 권하는 것을 줄이면 이렇습니다.

• 실행 전에 psdf.spark.explain()으로 계획을 봅니다.
• 계획이 너무 커지면 spark.checkpoint()나 spark.local_checkpoint()로 끊습니다.
• sort_values처럼 셔플이 드는 연산을 줄입니다.
• DataFrame.rank처럼 파티션 없이 Window를 써서 데이터를 파티션 하나로 모으는 API는 큰 데이터에서 GroupBy.rank로 바꿉니다.

7. 기본 인덱스

7-1. 세 가지 종류

pandas-on-Spark DataFrame은 인덱스가 필요합니다. Spark DataFrame에서 바꿀 때처럼 인덱스를 모르면 기본 인덱스를 붙이고 종류는 compute.default_index_type으로 고릅니다.

1
ps.set_option("compute.default_index_type", "distributed")
2
ps.get_option("compute.default_index_type")   # 'distributed'
3
ps.reset_option("compute.default_index_type")

distributed 예제의 인덱스 값은 [25769803776, 60129542144, 94489280512]처럼 큰 수로 나옵니다(문서의 ps.range(3) 예제). 순번이 필요 없으면 이 종류를 쓰라고 문서가 권합니다. 기본값 distributed-sequence는 3.5.6 문서와 4.2.0 문서가 같아서 버전에 따라 답이 갈리지 않습니다.

7-2. distributed-sequence의 주의점

4.2.0 Options and settings 문서에는 경고가 하나 있습니다. distributed-sequence는 인덱스 자체는 전역 순번이지만 각 인덱스에 대응하는 행은 데이터를 읽을 때마다 달라질 수 있습니다. apply(), groupby(), transform() 중에 인덱스를 새로 만들면 원래 행과 계산 값이 어긋날 수 있다고 적습니다. 경고가 연결하는 Best Practices 문서의 “Handling index misalignment with distributed-sequence” 절이 예방법 셋을 제시합니다.

1. 함수를 적용하기 전에 set_index("order_id")로 인덱스를 명시합니다.
2. psdf.spark.persist()로 행 순서를 고정합니다.
3. sequence 인덱스를 씁니다(성능 대가가 있습니다).

7-3. 다른 DataFrame끼리 연산

서로 다른 pandas-on-Spark DataFrame(또는 Series)끼리 연산하면 내부적으로 조인이 일어납니다. 옵션 이름은 compute.ops_on_diff_frames입니다.

4.2.0 옵션 문서 본문은 아직 “기본으로 막는다”고 쓰면서 옵션 표에는 True로 적혀 있어 문서 안에서 서술이 어긋납니다. 시험 기준 버전에 따라 답이 갈릴 수 있는 지점이라 표시해 둡니다. distributed 인덱스를 쓰면서 이 옵션을 켜면 두 DataFrame의 인덱스가 같다는 보장이 없어서 예상과 다른 결과가 나올 가능성이 크다고 문서가 경고합니다.

그 밖의 옵션 기본값입니다.

compute.max_rows와 compute.shortcut_limit의 설명은 옵션 표 문장을 줄여 옮겼고 어떤 연산이 이 지름길을 쓰는지는 문서에 목록이 없었습니다.

8. 변환과 비용

1
DataFrame.pandas_api(index_col=None)              # Spark DataFrame -> pandas-on-Spark (3.2.0)
2
pyspark.pandas.DataFrame.to_spark(index_col=None) # pandas-on-Spark -> Spark DataFrame
3
pyspark.pandas.DataFrame.to_pandas()              # pandas-on-Spark -> pandas
4
pyspark.pandas.from_pandas(pdf)                   # pandas -> pandas-on-Spark
5
DataFrame.toPandas()                              # Spark DataFrame -> pandas

인덱스를 지키면서 오가는 문서 예제를 주문 데이터로 바꾸면 이렇습니다.

1
psdf = ps.DataFrame({"amount": [12000, 5300, 87000]}, index=[101, 102, 103])
2

3
sdf = psdf.to_spark(index_col="order_id")          # 인덱스를 order_id 컬럼으로
4
sdf = sdf.filter("amount > 10000")
5
psdf2 = sdf.pandas_api(index_col="order_id")       # order_id를 다시 인덱스로

pandas_api 문서는 pandas-on-Spark에서 Spark로 갔다가 돌아오면 인덱스 정보를 잃고 원래 인덱스가 일반 컬럼이 된다고 적습니다. 4.0에서 예전 이름 to_koalas와 to_pandas_on_spark가 제거되고 pandas_api만 남았습니다.

헷갈리는 지점입니다.

• to_pandas()(pandas-on-Spark 메서드)와 toPandas()(Spark DataFrame 메서드)는 이름 표기가 다르고 둘 다 드라이버로 모읍니다.
• Spark DataFrame과 pandas-on-Spark DataFrame 사이 변환은 데이터를 옮기지 않습니다. 시험 보기의 “to_spark()는 데이터를 드라이버로 모은다”는 틀린 문장입니다.

9. Arrow 설정

Arrow는 JVM과 Python 사이에서 데이터를 컬럼 형식으로 주고받는 데 씁니다. pandas UDF는 추가 설정 없이 Arrow로 데이터를 옮긴다고 문서가 적습니다. toPandas()와 pandas DataFrame으로 createDataFrame()을 할 때는 아래 설정을 따릅니다.

값은 각 버전 Configuration 문서의 표에서 옮겼습니다. 두 기본값이 true로 바뀐 버전은 4.2이고 4.0과 4.1은 3.5와 같이 false입니다.

4.2 마이그레이션 가이드가 두 기본값 변경을 적습니다. PySpark와 JVM 사이 컬럼 데이터 교환이 기본으로 Arrow를 쓰게 됐고 일반 Python UDF도 기본으로 Arrow 최적화가 됐습니다. 예전 동작이 필요하면 각각 false로 돌립니다. 같은 4.2에서 PyArrow 최소 버전이 18.0.0으로 올랐습니다.

1
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")   # 3.5에서는 직접 켠다
2
pdf = orders.filter("status = 'PAID'").limit(10000).toPandas()

Arrow를 켜도 toPandas()는 모든 레코드를 드라이버로 모읍니다. 바뀌는 것은 직렬화 방식입니다. 문서가 Arrow 변환에서 지원하지 않는다고 적은 타입은 TimestampType의 ArrayType이고 MapType과 중첩 StructType의 ArrayType은 PyArrow 2.0.0 이상에서만 됩니다. 파티션을 Arrow 배치로 바꾸면서 JVM 메모리를 순간적으로 많이 쓸 수 있어서 컬럼이 많으면 maxRecordsPerBatch를 낮추라고 문서가 권합니다.

10. pandas UDF

10-1. 시그니처와 종류

1
pyspark.sql.functions.pandas_udf(f=None, returnType=None, functionType=None)

데코레이터나 함수 감싸기로 만들고 추가 설정은 필요 없습니다. 3.0부터 Python 타입 힌트로 종류를 정하고 functionType(PandasUDFType)은 호환용으로 남아 있으며 앞으로 deprecated 예정입니다. returnType은 DataType 객체나 DDL 문자열입니다.

구조체(StructType) 컬럼을 입력이나 출력으로 쓸 때만 pd.Series 자리에 pd.DataFrame을 씁니다.

10-2. Series에서 Series

내부적으로 컬럼을 배치로 나눠 배치마다 함수를 부르고 결과를 이어 붙입니다. 출력 길이는 항상 입력 길이와 같아야 합니다.

1
import pandas as pd
2
from pyspark.sql.functions import pandas_udf, col
3

4
@pandas_udf("double")
5
def with_vat(amount: pd.Series, rate: pd.Series) -> pd.Series:
6
    return amount * (1 + rate)
7

8
priced = orders.withColumn("amount_vat", with_vat(col("amount"), col("vat_rate")))

같은 함수는 로컬 pandas 데이터로 먼저 시험할 수 있습니다. 문서도 pandas UDF로 쓸 함수는 로컬 pandas 데이터로도 돌아가야 한다고 적습니다.

10-3. Series 반복자

비싼 초기화를 한 번만 하고 모든 배치에 재사용할 때 씁니다.

1
from typing import Iterator
2

3
@pandas_udf("string")
4
def normalize_coupon(batches: Iterator[pd.Series]) -> Iterator[pd.Series]:
5
    table = load_coupon_alias_table()      # 한 번만 읽는 별칭 표(가정한 함수)
6
    for s in batches:
7
        yield s.str.upper().map(lambda c: table.get(c, c))
8

9
orders.select(normalize_coupon("coupon_code"))

입력 컬럼이 여러 개면 Iterator[Tuple[pd.Series, pd.Series]]로 받고 for a, b in batches:로 풉니다.

10-4. Series에서 스칼라

PySpark 집계 함수처럼 동작합니다.

1
@pandas_udf("double")
2
def trimmed_mean(v: pd.Series) -> float:
3
    lo, hi = v.quantile(0.05), v.quantile(0.95)
4
    return float(v[(v >= lo) & (v <= hi)].mean())
5

6
payments.groupBy("method").agg(trimmed_mean("amount").alias("amount_tm"))

문서의 제약입니다. 부분 집계를 하지 않아서 그룹이나 윈도의 데이터가 전부 메모리에 올라갑니다. 윈도에는 경계가 없는(unbounded) 윈도만 됩니다. 반환 타입은 기본 데이터 타입이어야 합니다.

10-5. 헷갈리는 지점

• pandas UDF 안에서는 조건식이나 불리언 단축 평가가 지원되지 않아 모든 행에서 함수가 실행됩니다. 특정 행에서 실패할 수 있으면 조건을 함수 안에 넣으라고 API 문서가 적습니다.
• 호출하는 쪽에서 키워드 인자를 쓸 수 없다고 적혀 있습니다. 4.0.0에서 SCALAR와 GROUPED_AGG 종류는 키워드 인자를 지원하게 바뀌었다는 표시도 같은 문서에 있어서 버전에 따라 다릅니다.
• 같은 API 문서에 4.1.0에서 GROUPED_MAP 종류가, 4.2.0에서 GROUPED_AGG 종류가 반복자 API를 지원하게 됐다는 변경 표시가 있습니다. 시험 기준 버전이 이전이면 범위 밖입니다.
• ArrayType의 TimestampType과 중첩 StructType은 출력 타입으로 지원하지 않는다고 API 문서가 적습니다.
• 반환한 Series의 타입이 returnType과 다르면 Spark가 변환을 시도하고 그 변환이 맞다는 보장이 없습니다. 4.1부터 spark.sql.execution.pandas.convertToArrowArraySafely가 기본으로 켜져서 정수 넘침 같은 안전하지 않은 변환에 오류를 냅니다.

11. pandas Function API

pandas UDF가 컬럼(Column) 단위라면 pandas Function API는 DataFrame 단위로 Python 함수를 적용합니다. 내부적으로 Arrow와 pandas를 쓰는 점은 같습니다. Arrow 문서는 Function API의 타입 힌트가 선택이고 지금은 동작에 영향을 주지 않는다고 적습니다. 다만 4.2.0 applyInPandas API 문서는 4.1.0에 추가된 반복자 형태를 타입 힌트(Iterator[pd.DataFrame])로 받는 예제를 보여서 4.1 이후에는 힌트가 형태를 가를 수 있습니다.

1
# 고객별 주문 금액에서 고객 평균을 빼기(split-apply-combine)
2
def center_amount(pdf: pd.DataFrame) -> pd.DataFrame:
3
    return pdf.assign(amount=pdf.amount - pdf.amount.mean())
4

5
centered = (orders.select("customer_id", "order_id", "amount")
6
                  .groupBy("customer_id")
7
                  .applyInPandas(center_amount,
8
                                 schema="customer_id string, order_id string, amount double"))
9

10
# 배치마다 큰 주문만 남기기
11
def big_orders(batches):
12
    for pdf in batches:
13
        yield pdf[pdf.amount >= 100000]
14

15
large = orders.mapInPandas(big_orders, schema=orders.schema)

헷갈리는 지점입니다.

• applyInPandas는 함수를 적용하기 전에 그룹 하나의 데이터를 전부 메모리에 올립니다. 그룹 크기가 쏠리면 OOM이 날 수 있고 maxRecordsPerBatch는 그룹에 적용되지 않는다고 문서가 적습니다. Ch-04 10절의 익스큐터 OOM 원인 표에 이 API를 넣은 이유입니다. 4.1.0부터는 pandas DataFrame 반복자를 받는 형태도 지원하고 API 문서는 이 형태로 OOM 위험을 줄일 수 있다고 적습니다. cogroup 쪽은 코그룹 하나의 데이터를 전부 메모리에 올린다는 같은 주의가 있습니다.
• 반환 DataFrame의 컬럼 라벨은 schema를 문자열로 주면 필드 이름과 맞아야 하고 정수 인덱스처럼 문자열이 아니면 위치로 맞춥니다.
• mapInPandas의 각 pandas DataFrame 크기는 maxRecordsPerBatch로 조절됩니다.
• 3.0 이전의 grouped map pandas UDF(PandasUDFType.GROUPED_MAP과 groupby().apply())는 applyInPandas로 옮기라고 문서가 권합니다.

12. 세 가지 Python UDF 비교

Arrow Python UDF는 따로 있는 함수가 아니고 udf()에 useArrow를 준 것입니다. 시그니처가 udf(f=None, returnType=StringType(), *, useArrow=None)이라 기본 반환 타입은 둘 다 StringType입니다. 4.2부터 일반 Python UDF가 기본으로 Arrow 최적화를 쓰면서 타입 변환(coercion) 동작이 달라졌습니다. 반환 값이 지정 타입과 맞지 않을 때 결과가 3.5와 다를 수 있다는 뜻입니다.

4.2.0 udf API 문서에는 4.1.0 변경 표시가 하나 더 있습니다. 함수에 pandas UDF의 타입 힌트 패턴(예: pd.Series -> pd.Series)을 달고 useArrow를 따로 주지 않으면 udf로 만들어도 pandas UDF로 처리됩니다. 3.5 기준 보기에서는 @udf가 행 단위라는 판단이 맞지만 4.1 이후 문서로 공부했다면 이 차이를 같이 기억합니다.

13. 확인 문제

시험 안내서 항목을 보고 직접 만든 문제입니다. 덤프나 문제 은행에서 가져온 것은 없습니다.

문제 1. Spark Connect 클라이언트가 서버로 보내는 것은 무엇입니까?

• A. Python 바이트코드
• B. 해석 전 논리 계획(protocol buffers, gRPC)
• C. 최적화된 물리 계획
• D. 직렬화한 RDD

문제 2. Spark Connect 세션에서 실패하는 코드는 무엇입니까?

• A. orders.groupBy("status").count().show()
• B. orders.rdd.map(lambda r: r.amount).sum()
• C. orders.select(F.spark_partition_id()).distinct().count()
• D. orders.createOrReplaceTempView("orders")

문제 3. 노트북에서 spark-submit으로 YARN에 제출했고 드라이버가 클러스터 안의 노드에서 뜨기를 원합니다. 어떤 옵션입니까?

문제 4. local[*]과 local의 차이는 무엇입니까?

문제 5. 1억 행 Spark DataFrame을 pandas_api()로 바꿨더니 느립니다. 행 순서와 연속 번호는 필요 없습니다. 무엇을 바꾸면 됩니까?

문제 6. 주문 금액 컬럼에 같은 길이의 결과를 내는 벡터화 함수를 적용하려고 합니다. 맞는 정의는 무엇입니까?

• A. @pandas_udf("double") + def f(s: pd.Series) -> float
• B. @pandas_udf("double") + def f(s: pd.Series) -> pd.Series
• C. udf(lambda a: a * 1.1, "double")
• D. orders.mapInPandas(f, "double")

문제 7. orders.groupBy("customer_id").applyInPandas(func, schema)가 특정 고객 몇 명 때문에 익스큐터 OOM을 냅니다. spark.sql.execution.arrow.maxRecordsPerBatch를 1000으로 낮추면 해결됩니까?

문제 8. Spark 3.5에서 toPandas()가 Arrow를 쓰게 하려면 무엇을 합니까? 4.2에서는 어떻습니까?

14. 범위와 한계

• 동작과 기본값은 2026년 10월 19일에 연 Spark 4.2.0 문서(Spark Connect Overview, Cluster Mode Overview, Submitting Applications, Configuration)와 PySpark 4.2.0 문서(Apache Arrow in PySpark, pandas API on Spark의 Options and settings, From/to pandas and PySpark DataFrames, Best Practices, Upgrading PySpark, API 참조)를 기준으로 했습니다. 3.5 기본값은 3.5.6 Configuration 문서로, 4.0과 4.1의 Arrow 기본값은 4.0.0과 4.1.0 Configuration 문서로 2026년 10월 20일에 다시 확인했습니다. pyspark-client의 첫 버전은 PyPI 릴리스 목록으로 확인했습니다.
• 이 글의 코드는 아직 실행하지 않았습니다. 이 노트북에 Java가 없어서 PySpark와 로컬 Connect 서버를 띄우지 못했습니다. Python 문법 검사(ast.parse)만 했습니다. load_coupon_alias_table()은 설명용으로 가정한 함수입니다.
• 2026년 10월 21일에 Connect의 오류 시점(Eager vs Lazy 문서), accumulator와 broadcast 변수(Connect 세션 소스), start-connect-server.sh의 옵션 처리(스크립트 소스), pandas API on Spark의 Connect 지원(PySpark API 참조 안내문)을 확인해 본문에 넣었습니다. Arrow 관련 기본값 넷은 3.5.6, 4.0.0, 4.1.0, 4.2.0 태그의 SQLConf.scala로 한 번 더 맞춰 봤습니다.
• 1절의 영역별 문항 수 짐작은 비중으로 계산한 값이고 Connect 위 pandas API on Spark의 메서드별 지원 표는 찾지 못했습니다.

참고 자료

공식 자료

• Databricks, Exam Guide, 2026-10-08판, Exam Guide, Oct 2025 (6, 7영역 항목과 샘플 문항 7)
• Apache Spark 4.2.0, Spark Connect Overview, Eager vs Lazy: Spark Connect vs Spark Classic
• Apache Spark 저장소 v4.2.0, python/pyspark/sql/connect/session.py, sbin/start-connect-server.sh, SQLConf.scala
• PySpark 4.2.0, API Reference (Spark Connect 지원 안내문)
• PySpark 4.2.0, Quickstart: Spark Connect
• Apache Spark 저장소, client-connection-string.md (master 브랜치)
• Apache Spark 4.2.0, Cluster Mode Overview, Submitting Applications
• Apache Spark 4.2.0, Spark Configuration, 4.1.0 Spark Configuration, 4.0.0 Spark Configuration, 3.5.6 Spark Configuration
• PySpark 3.5.6, Options and settings
• PyPI, pyspark-client (릴리스 목록)
• PySpark 4.2.0, PySpark Overview (pandas API on Spark 소개)
• PySpark 4.2.0, Options and settings, From/to pandas and PySpark DataFrames, Best Practices
• PySpark 4.2.0, Apache Arrow in PySpark
• PySpark 4.2.0, Upgrading PySpark
• PySpark 4.2.0 API: functions.pandas_udf, DataFrame.mapInPandas, GroupedData.applyInPandas, PandasCogroupedOps.applyInPandas, functions.udf, DataStreamWriter.foreachBatch, DataFrame.pandas_api, pyspark.pandas.DataFrame.to_spark, pyspark.pandas.DataFrame.to_pandas, SparkSession.builder.remote

이 시리즈

• Spark Ch-00: 시험 구조와 공부 계획, Spark Ch-04: 문제 해결과 튜닝, Spark Ch-05: Structured Streaming