PySpark - 2023년 한 해를 돌아보며

작성자: Hyukjin Kwon, Takuya Ueshin, Allison Wang, Ruifeng Zheng, Xinrong Meng, Haejoon Lee, Amanda Liu

April 10, 2024 in 산업

이 포스트 공유하기

(번역: Youngkyong Ko) Original Post

2023년에 출시된 Apache Spark 3.4와 3.5에서는 PySpark의 성능, 유연성, 사용 편의성을 개선하는 데 중점을 두었습니다. 이 블로그 게시물에서는 주요 개선 사항을 안내합니다.

다음은 2023년에 Apache Spark 3.4와 3.5에 추가된 가장 중요한 기능 중 몇 가지를 요약한 것입니다:

Spark Connect는 분리된 클라이언트-서버 아키텍처를 도입하여 모든 애플리케이션에서 Spark 클러스터에 원격으로 연결할 수 있습니다. 따라서 안정성과 업그레이드 가능성, 관찰 가능성이 향상되는 동시에, Spark as a service가 활성화됩니다.
Arrow에 최적화된 Python 사용자 정의 함수(UDF)를 통해 Arrow 컬럼 기반 형식을 활용하여 일반 Python UDF의 성능을 두 배로 향상시킴으로써 효율성의 비약적인 발전을 보여주었습니다.
Python 사용자 정의 테이블 함수(UDTF)를 통해 이제 사용자는 PySpark에서 기본적으로 테이블 기반 변환을 수행할 수 있습니다.
GROUP BY ALL 및 ORDER BY ALL과 같은 새로운 Spark SQL 기능이 도입되었으며, 이 기능들은 모두 PySpark에서 기본적으로 사용할 수 있습니다.
파이썬 임의 상태 저장 처리는 스트리밍 쿼리에서 임의의 상태를 유지할 수 있는 기능을 제공합니다.
TorchDistributor는 아파치 스파크 클러스터에서 분산된 파이토치 트레이닝을 지원합니다.
새로운 테스트 API는 PySpark 애플리케이션을 효과적으로 테스트하고 개발자가 고품질 코드를 생성할 수 있도록 도와줍니다.
English SDK는 일반 영어 명령을 PySpark 및 SQL로 변환하여 개발자의 생산성을 높일 수 있는 LLM 기반 프로그래밍 접근 방식입니다.

다음 섹션에서는 각각을 살펴보고 몇 가지 주목할 만한 추가 개선 사항에 대해 설명합니다.

Apache Spark 3.5 및 3.4: 기능 심층 분석

Spark Connect: : Apache Spark의 원격 연결

Apache Spark 3.4에서 처음 선보인 Spark Connect는 모든 애플리케이션에서 어디서나 실행 중이든지 Spark 클러스터에 원격으로 연결할 수 있는 분리된 클라이언트-서버 아키텍처를 도입했습니다. 이렇게 클라이언트와 서버를 분리함으로써 최신 데이터 애플리케이션, IDE, 노트북, 프로그래밍 언어가 대화형으로 Spark에 액세스할 수 있습니다. 또한, 분리된 아키텍처는 안정성, 업그레이드 가능성, 디버깅 가능성 및 관찰 가능성을 향상시킵니다.

Apache Spark 3.5에서는 Scala 지원이 완료되었으며, 구조화된 스트리밍(SPARK-42938), ML 및 PyTorch(SPARK-42471), Spark의 Pandas API(SPARK-42497) 등 주요 Spark 구성 요소에 대한 지원도 추가되었습니다.

데이터브릭스에서 Spark Connect를 사용하려면 Databricks Connect를 사용하거나 또는 Apache Spark에 직접 Spark Connect를 사용하세요.

Arrow에 최적화된 Python UDF: Python UDF의 성능 향상하기

Arrow에 최적화된 Python UDF(SPARK-40307)는 Arrow 컬럼 기반 형식을 활용하여 상당한 성능 최적화를 가능하게 합니다. 예를 들어, 동일한 클러스터에서 UDF를 체인으로 연결할 경우, 32GB 데이터 세트에서 Arrow에 최적화된 Python UDF는 피클된 Python UDF보다 약 1.9배 더 빠르게 실행됩니다.

파이썬 UDTF

Apache Spark 3.5에서는 단일 스칼라 결과 값 대신 테이블을 출력으로 반환하는 사용자 정의 테이블 함수를 통해 PySpark의 UDF 지원을 확장했습니다. 이 UDTF를 등록하면 SQL 쿼리의 FROM 절에서 사용할 수 있습니다. 예를 들어, UDTF SquareNumbers는 입력값과 그 제곱값을 테이블로 출력합니다:

from pyspark.sql.functions import udtf

@udtf(returnType="num: int, squared: int")
class SquareNumbers:
    def eval(self, start: int, end: int):
        for num in range(start, end + 1):
            yield (num, num * num)

from pyspark.sql.functions import lit

SquareNumbers(lit(1), lit(3)).show()

+---+-------+
|num|squared|
+---+-------+
|  1|      1|
|  2|      4|
|  3|      9|
+---+-------+

새로운 SQL 기능

PySpark 주요 이점 중 하나는 Spark SQL이 PySpark 데이터프레임과 원활하게 작동한다는 것입니다. 2023년에 Spark SQL은 spark.sql을 통해 직접 활용할 수 있는 많은 새로운 기능을 도입했습니다. GROUP BY ALL 및 ORDER BY ALL, 일반 테이블 값 함수 지원, INSERT BY NAME, PIVOT 및 MELT, ANSI 준수 등이 그 예입니다. 다음은 GROUP BY ALL 및 ORDER BY ALL 사용 예시입니다:

spark.sql("""
    SELECT name, firstname, level, sum(comp) as totalcomp 
    FROM {table}
    GROUP BY ALL
    ORDER BY ALL
""", table=df)

파이썬 임의 상태 저장 처리(arbitrary stateful processing)

구조화된 스트리밍에서의 파이썬 임의 상태 저장 작업은 스트리밍 쿼리 실행 전반에 걸쳐 상태 처리를 허용함으로써 PySpark에서 수많은 실시간 분석 및 머신 러닝 사용 사례를 가능하게 합니다. 다음 예제는 임의 상태 저장 처리를 보여줍니다:

# Group the data by word, and compute the count of each group
output_schema = "session STRING, count LONG"
state_schema = "count LONG"
sessions = events.groupBy(events["session"]).applyInPandasWithState(
    func,
    output_schema,
    state_schema,
    "append",
    GroupStateTimeout.ProcessingTimeTimeout,
)

# Start running the query that prints the windowed word counts to the console
query = sessions.writeStream.foreachBatch(
    lambda df, _: df.show()).start()

TorchDistributor: PyTorch 네이티브 통합

TorchDistributor는 PySpark에서 PyTorch에 대한 네이티브 지원을 제공하여, Spark 클러스터에서 딥러닝 모델의 분산 학습을 가능하게 합니다. 이 도구는 PyTorch 프로세스를 시작하고 분산처리 메커니즘은 PyTorch에 맡겨 프로세스가 코디네이트 되도록 하는 역할만 수행합니다.

TorchDistributor 사용이 간편하며 몇 가지 주요 설정만 고려하면 됩니다:

from pyspark.ml.torch.distributor import TorchDistributor

model = TorchDistributor(
    num_processes=2,
    local_mode=True, 
    use_gpu=True,
).run(<function_or_script>, <args>)

Testing API: PySpark 데이터프레임에 대한 더 쉬운 테스트

pyspark.testing 패키지의 새로운 테스트 API(SPARK-44042)는 PySpark 애플리케이션을 테스트하는 개발자를 위해 크게 개선된 기능을 제공합니다. 자세한 오류 메시지와 함께 동등성 테스트를 위한 유틸리티 함수를 제공하여 데이터프레임 스키마와 데이터의 불일치를 더 쉽게 식별할 수 있습니다. 아래 예제 출력은 이를 보여줍니다:

*** actual ***
  Row(name='Amy', languages=['C++', 'Rust'])
! Row(name='Jane', languages=['Scala', 'SQL', 'Java'])

*** expected ***
  Row(name='Amy', languages=['C++', 'Rust'])
! Row(name='Jane', languages=['Scala', 'Java'])

English SDK: 프로그래밍 언어로서의 영어

Apache Spark의 English SDK는 사용자가 일반 영어로 명령을 입력한 다음 이를 PySpark 및 Spark SQL 코드로 변환할 수 있도록 하여 사용을 간소화합니다. 이를 통해 특히 데이터 프레임 변환 작업, 데이터 수집 및 UDF와 관련된 코드에서 PySpark 프로그래밍에 더 쉽게 접근할 수 있으며 캐싱 덕분에 생산성을 더욱 높일 수 있습니다. English SDK는 개발 프로세스를 간소화하고 코드 복잡성을 최소화하며 Spark 커뮤니티의 범위를 확장할 수 있는 큰 잠재력을 가지고 있습니다. 직접 사용해 보세요!

기타 주목할 만한 개선 사항

Apache Spark 3.4와 3.5에 도입된 몇 가지 다른 기능들입니다. 아직 익숙하지 않으시다면 한번 살펴보시기 바랍니다:

Parameterized queries with PySpark
SQL function parity: 150 new functions (SPARK-43907), including sketch-based approximate distinct counting
Lateral column alias support
Dropping duplicates within watermarks for easier deduplication (SPARK-42931)
DeepSpeed Distributor
Advanced job cancellation API (SPARK-44194)
Python memory profilers
Better autocompletion in IPython (SPARK-43892)
Documentation improvements (SPARK-42374, SPARK-42493, and SPARK-42642)

회고와 전망

2023년에는 오픈 소스 커뮤니티의 활발한 혁신이 PySpark와 Apache Spark를 크게 강화하여 데이터 전문가가 사용할 수 있는 도구 키트를 넓히고 분석 워크플로우를 간소화했습니다. 곧 출시될 Apache Spark 4.0에서, PySpark는 새로운 기능과 향상된 성능을 통해 데이터 처리를 더욱 혁신하고 데이터 엔지니어링 및 데이터 사이언스 커뮤니티 내에서 데이터 분석을 발전시키기 위한 노력을 더욱 공고히 할 준비가 되어 있습니다.

새로운 기능 시작하기

이 포스팅에서는 2023년에 Apache Spark 3.4와 3.5에서 사용 편의성, 성능 및 유연성을 향상시키기 위해 개선된 가장 중요한 기능들에 대한 간략한 개요를 살펴 보았습니다. 이 모든 기능은 데이터브릭스 런타임 13과 14에서 사용할 수 있습니다. 지금 바로 이 새로운 기능들을 직접 사용해 보시는 것은 어떨까요?