Spark로 Python 변환 실행하기

PySpark는 Python에서 Apache Spark의 DataFrame과 분산 계산을 사용하는 API입니다. 큰 구조화 데이터를 여러 executor에서 처리할 수 있습니다. 단일 노드로 충분한 작업과 Spark가 필요한 작업은 계산 엔진 선택에서 구분합니다.

DataFrame, RDD와 공유 변수

DataFrame은 이름과 자료형이 있는 열로 구성한 행의 컬렉션입니다. 관계형 연산과 SQL 표현식을 사용하며, 실제 데이터를 일반 Python list나 dict처럼 전부 보관하는 컨테이너와는 다릅니다. Python 객체는 분산 데이터와 계산 계획을 참조합니다.

DataFrame의 논리적인 내용은 불변입니다. select, filter와 withColumn은 새 DataFrame을 반환하고 변수 이름을 다시 지정해 다음 계산에 사용할 수 있습니다. 특정 물리 행의 주소를 찾아 제자리 수정하는 모델은 아니지만, 키 조건으로 행을 선택하거나 조건에 따른 새 열 값을 계산할 수 있습니다.

RDD는 더 낮은 수준의 분산 컬렉션입니다. 데이터를 여러 파티션으로 나누어 병렬 처리하며, 구조화된 DataFrame 연산에서는 엔진이 많은 분할과 실행 계획을 관리합니다.

일반 Python 변수는 작업에 전달될 때 별도 사본으로 사용됩니다. 여러 작업이 하나의 mutable Python 객체를 공유한다고 가정하지 않습니다. Spark는 읽기 전용 값을 executor에 배포하는 broadcast 변수와, 합계/카운터처럼 추가하는 accumulator를 제공합니다. 이는 일반 전역 변수의 공유와 다른 인터페이스입니다.

지연 실행과 Foundry 출력

DataFrame 변환은 보통 계획을 구성하고, 출력 쓰기나 count/collect 같은 action에서 실제 계산합니다. Spark는 중간 계획을 최적화하여 실행합니다. 모든 메서드가 실행 시점까지 오류를 미루는 것은 아니며, 열 해석이나 계획 분석 단계에서도 자료형/열 오류가 드러날 수 있습니다.

새 DataFrame을 만들었다고 즉시 새 Foundry 데이터셋이 저장되는 것은 아닙니다. 선언한 출력에 쓰거나 반환한 결과를 빌드가 저장하면 데이터셋 트랜잭션과 계보가 만들어집니다. 보존된 버전은 권한과 보존 정책 범위 안에서 조회할 수 있습니다.

계보는 여러 입력을 결합할 수 있는 방향성 그래프입니다. 중간 출력 데이터셋을 저장하면 권한이 있는 다른 사용자가 재사용할 수 있고, 어떤 입력과 코드로 결과를 만들었는지 추적할 수 있습니다.

열 중심으로 생각하기

행의 물리적인 순서를 전제로 하지 않고 열 선택, 필터, 집계와 조인으로 필요한 결과를 표현합니다. 정렬은 필요할 때 명시하며 비용을 고려합니다. 적은 수의 제어 값이나 조사 표본은 Python으로 가져올 수 있지만 큰 결과를 collect하면 driver 메모리를 초과할 수 있습니다.

입력 schema를 확인하고 계산 전에 필요한 cast를 적용합니다. 숫자처럼 보이는 문자열이나 날짜에 산술 연산을 적용할 때 암묵적 자료형 변환에 의존하지 않습니다.

Foundry 출력 열은 소문자와 밑줄을 사용하는 이름으로 통일하고 공백, 괄호와 특수문자를 피하는 편이 안전합니다. 집계가 자동으로 만든 이름에는 alias를 붙입니다. Spark의 대소문자 해석 설정과 저장 형식 제약을 구분하며 대소문자만 다른 열 이름에 의존하지 않습니다.

같은 계산을 단계별 또는 체인으로 표현하기

다음 함수는 firstName과 정수로 해석할 수 있는 age 열을 받습니다. age가 21을 초과하는 행만 남기고 이름을 first_name으로 바꿉니다.

from pyspark.sql import functions as F


def adults_step_by_step(old_frame):
    selected = old_frame.select("firstName", "age")
    typed = selected.withColumn("age", F.col("age").cast("integer"))
    adults = typed.filter(F.col("age") > 21)
    return adults.withColumnRenamed("firstName", "first_name")


def adults_chained(old_frame):
    return (
        old_frame.select("firstName", "age")
        .withColumn("age", F.col("age").cast("integer"))
        .filter(F.col("age") > 21)
        .withColumnRenamed("firstName", "first_name")
    )

두 결과는 first_name과 age 두 열만 가지며 21세 이하는 제외합니다. 계산 단계마다 새 DataFrame을 받아도 원래 입력의 값을 직접 바꾸는 것은 아닙니다.

Foundry의 편집 환경

Code Repositories에서는 Python 파일에 필요한 import를 명시합니다.

from pyspark.sql import functions as F, types as T

Code Workbook은 F를 기본 전역 환경에 제공하므로 편집 환경의 기본 import를 사용할 수 있습니다. 저장소로 옮기는 코드는 필요한 import를 명시하여 환경에 대한 의존을 분명히 합니다. 기본 함수로 표현하기 어려운 행별 Python 계산에는 UDF를 사용할 수 있지만 비용과 실행 환경을 고려합니다.

기초와 전체 참조

Python에서 PySpark로 사고방식 바꾸기는 리스트 반복과 지연 Column 계산을 비교합니다.
문법 빠른 참조는 필터, 조인, 타입, 중첩 자료, 집계와 UDF의 자주 쓰는 패턴입니다.
작성 스타일은 Column helper, schema 계약, 조인과 체인 길이 기준을 설명합니다.
열과 관계 연산은 필터, UDF, 조인, 집계, window와 로그의 상세 참조에 연결됩니다.
행 선택과 값 변환은 중복 제거, 날짜, 문자열, 수학, JSON과 체크포인트에 연결됩니다.

Foundry 실행 기능

window는 원래 행을 유지하면서 집계와 순위를 붙이고, groupBy는 그룹별 결과로 행을 줄입니다. 이 차이는 window 참조에서 전체 partition과 누적 frame으로 구체화합니다.

Keyboard shortcuts

팔란티어 플랫폼 한국어 해설