Python 반복문에서 PySpark 표현식으로

일반 Python에서는 list의 값을 하나씩 바꾸는 절차를 작성할 수 있습니다. PySpark에서는 원하는 결과를 DataFrame과 Column 표현식으로 기술하고 Spark가 분산 실행 계획을 만듭니다. Python의 표현력과 SQL의 선언적 연산을 함께 사용하지만 Python 코드의 모든 반복이 데이터의 행을 직접 처리하는 것은 아닙니다.

각 숫자에 5 더하기

작은 Python 목록의 값을 처리하는 예제입니다.

old_list = [1, 2, 3]
new_list = []
for number in old_list:
    new_list.append(number + 5)
print(new_list)

새 목록은 [6, 7, 8]입니다. 같은 행별 계산을 Spark에서는 열 표현식으로 만듭니다.

from pyspark.sql import functions as F


def add_five(old_dataframe):
    return old_dataframe.withColumn("added_number", F.col("number") + 5)

표는 각 입력값과 계산값의 대응이며 분산 결과의 행 순서를 보장하는 표시가 아닙니다.

계획과 실행 구분하기

함수를 호출하면 새 DataFrame이 계산 계획을 나타냅니다. 결과를 Foundry에 쓰는 등 action을 요청하면 Spark가 중간 연산을 최적화하고 cluster에서 실행합니다. 전체 데이터를 한 대의 Python 프로세스에 올리지 않고도 처리할 수 있는 이유입니다.

데이터의 값을 일반 변수, list나 dict로 직접 꺼내 반복하기보다 필터, 열 연산, 조인과 집계를 사용합니다. 반대로 열 이름 목록이나 몇 개의 조건식처럼 작은 메타데이터를 Python 반복문으로 구성하는 것은 별개의 작업입니다.

사용자 Python 로직이 필요한 경우

기본 함수로 로직을 표현할 수 없다면 UDF를 사용하여 실제 행 값을 Python 함수에 전달할 수 있습니다. UDF에는 실행과 데이터 전달 비용이 있으므로 같은 계산을 제공하는 기본 함수가 있으면 그 함수를 우선 선택합니다.

열 참조에서는 Column과 select의 차이 및 cast를, 필터에서는 Python 불리언과 Spark 불리언 표현식의 차이를 설명합니다.