Keyboard shortcuts

Press ← or → to navigate between chapters

Press ? to show this help

Press Esc to hide this help

읽기 쉬운 PySpark 코드 작성하기

PySpark 코드는 분산 실행 계획과 Python 코드의 구조를 함께 관리해야 합니다. 아래 규칙은 읽기와 변경을 쉽게 하기 위한 권장 작성 기준이며, API가 지원하는 연산 자체를 금지하는 규칙은 아닙니다.

주요 저장소는 같은 Pylint 구성을 사용하고 필요한 PySpark 전용 checker를 적용합니다. 단위 테스트와 스타일 검사에 저장소 설정이 있습니다.

열 참조를 계산 위치에 맞추기

일반 변환에서는 현재 DataFrame에서 해석할 이름을 사용합니다. 열 이름을 받는 기본 함수는 F.lower("status")처럼 호출할 수 있습니다. 산술식이나 사용자 helper에 Column이 필요하면 F.col("status")를 사용합니다.

다른 DataFrame에 묶인 df1.colA를 현재 df의 계산에 가져오면 이전 schema나 변수 상태에 의존하기 쉽습니다. F.col("prod_status") == DELIVERED 같은 표현식은 호환되는 여러 DataFrame에 재사용할 수 있습니다.

여러 입력에 같은 이름이 있는 조인에서는 입력에 직접 묶인 Column이나 DataFrame alias로 어느 쪽 열인지 구분합니다. 이름을 통한 일반 참조와 조인 disambiguation은 용도가 다릅니다.

Column 계산을 먼저 구성하기

중간 계산마다 DataFrame에 열을 붙이기보다 Column 식을 만들고 최종 단계에서 이름을 붙입니다. 예제의 새 열 이름은 입력 열과 겹치지 않도록 정합니다.

from pyspark.sql import functions as F

ACTIVE_STATUS = "active"
ACTIVE_CATEGORY = "A"
INACTIVE_CATEGORY = "B"


def status_features(df):
    status_lower = F.lower(F.col("status"))
    is_active = status_lower == ACTIVE_STATUS
    category = F.when(is_active, ACTIVE_CATEGORY).otherwise(INACTIVE_CATEGORY)
    return df.select(
        "*",
        status_lower.alias("status_lower"),
        is_active.alias("is_active"),
        category.alias("category"),
    )

중간 값은 원래 열과 다른 이름으로 구분합니다. 예를 들어 amount를 센트로 바꾼 뒤 같은 amount 이름을 계속 덮어쓰면 어떤 단위인지 추적하기 어렵습니다.

from pyspark.sql import functions as F

CENTS_PER_UNIT = 100


def amount_in_cents(df):
    amount_cents = F.col("amount") * CENTS_PER_UNIT
    amount_rounded = F.round(amount_cents, 2)
    return df.select("*", amount_rounded.alias("amount_cents"))

기존 이름을 유지해야 하는 자료형 변경이나 schema 계약은 마지막 projection에서 명시합니다. 이 원칙은 DataFrame이 실제로 제자리 수정된다는 뜻이 아니라 중간 이름의 재사용에 따른 혼동을 줄이려는 것입니다.

재사용 helper는 Column 입출력으로 만들기

열 값을 변환하는 helper는 이름 문자열이나 DataFrame 전체보다 Column을 받아 Column을 반환하게 구성합니다. 기본 Spark 함수가 열 이름 문자열을 받는 기능과 사용자 helper의 계약을 구분합니다.

from pyspark.sql import functions as F


def make_full_name(first_name, last_name):
    return F.concat(first_name, F.lit(" "), last_name)


def name_length(name):
    return F.length(name)


def normalize(value):
    return F.lower(value)


def person_features(df):
    full_name = make_full_name(F.col("first_name"), F.col("last_name"))
    return df.select(
        "*",
        full_name.alias("full_name"),
        name_length(full_name).alias("name_length"),
        normalize(F.col("status")).alias("status_normalized"),
    )

DataFrame을 결합하거나 필터하는 단계는 상위 함수에 남기고, 재사용할 값 계산은 작은 Column helper로 구성합니다.

여러 열은 한 번에 선택하기

열마다 withColumnRenamed를 반복하면 Python이 projection을 여러 겹으로 구성합니다. 문제가 되는 것은 반복문이라는 문법 자체가 아니라 반복해서 만드는 긴 계획과 그 분석 비용입니다. 같은 작업을 한 select로 표현할 수 있으면 그 형태를 사용합니다.

from pyspark.sql import functions as F


def lowercase_column_names(df):
    return df.select(*[
        F.col("`" + name.replace("`", "``") + "`").alias(name.lower())
        for name in df.columns
    ])

소문자화 뒤 이름이 서로 충돌하지 않는 입력에 사용합니다. 반복문으로 작은 Python 조건 목록이나 expression 목록을 만드는 것 자체가 Spark action을 반복 실행하는 것은 아닙니다.

복잡한 불리언 조건에 이름 붙이기

한 논리 블록의 복합 표현식은 세 개 정도를 기준으로 나눕니다. 조건이 더 길어지면 의미 있는 변수나 helper로 분리합니다. 다음은 배송 날짜가 지났고 등록 또는 운영 정보가 있거나, 이미 Delivered 상태이면 In Service로 분류하는 가상의 규칙입니다.

from pyspark.sql import functions as F

DELIVERED = "Delivered"
IN_SERVICE = "In Service"


def service_status(prod_status, delivery_date, current_date,
                   registration, original_operator, current_operator):
    has_operator = original_operator.rlike(".+") | current_operator.rlike(".+")
    delivery_date_passed = F.datediff(delivery_date, current_date) < 0
    has_registration = registration.rlike(".+")
    is_delivered = prod_status == DELIVERED
    is_active = has_registration | has_operator
    return F.when(is_delivered | (delivery_date_passed & is_active), IN_SERVICE)

날짜 비교는 날짜 단위이며 같은 날은 passed에 포함하지 않습니다. 중복 계산을 하나의 변수로 모으면 조건의 관계와 재사용 범위가 드러납니다. 설명되지 않은 나머지 상태를 임의의 정상 값으로 바꾸는 otherwise는 넣지 않습니다.

select로 schema 계약 표현하기

입력의 시작과 출력 경계에서 필요한 열을 명시합니다. 새로운 원천 열이 추가되어도 예상하지 않은 열이 출력에 섞이는 것을 막을 수 있습니다.

select의 각 열은 단순 선택이나 한 단계 변환과 alias 정도로 유지합니다. 이런 변환이 세 개를 넘거나 긴 when/여러 입력 참조가 섞이면 계산을 먼저 이름 있는 expression이나 clean 함수로 분리합니다.

from pyspark.sql import functions as F


def clean_aircraft(aircraft):
    return aircraft.select(
        "aircraft_id", "aircraft_msn", "aircraft_type", "operator_code",
        F.col("aircraft_registration").alias("registration"),
        F.col("number_of_economy_seats").cast("long"),
        F.col("number_of_business_seats").cast("long"),
    )


def comments_contract(df):
    return df.select("key", F.col("comments").cast("double").alias("num_comments"))

단순한 새 열 하나를 계산 단계로 보여 줄 때는 withColumn도 사용할 수 있습니다. 여러 중간 계산을 별도 이름의 Column으로 준비하여 마지막 select에 붙이는 방식과 일관되게 사용합니다.

긴 계산을 select에 직접 넣는 대신 분리합니다. 아래 함수의 created_at/closed_at은 해당 함수가 파싱할 수 있는 시각 열이고, closed_at이 없으면 현재 시각까지 열린 기간을 계산합니다.

from pyspark.sql import functions as F

SECONDS_PER_DAY = 86400


def days_open(df):
    closed_or_now = F.coalesce(F.unix_timestamp("closed_at"), F.unix_timestamp())
    elapsed_seconds = closed_or_now - F.unix_timestamp("created_at")
    return df.withColumn("days_open", elapsed_seconds / SECONDS_PER_DAY)

여기서 coalesce의 현재 시각 대체는 열려 있는 항목의 기간이라는 명시적인 업무 규칙입니다.

빈 열은 자료형을 가진 NULL로 만들기

schema를 맞추기 위한 빈 열에는 빈 문자열이나 NA 같은 표시 문자열 대신 NULL을 사용합니다.

from pyspark.sql import functions as F


def add_optional_note(df):
    return df.withColumn("foo", F.lit(None).cast("string"))

None 자체는 구체적인 자료형이 없으므로 목적에 맞는 타입으로 cast합니다. 이후 isNull 같은 조건도 일관되게 사용할 수 있습니다.

주석은 데이터의 이유를 설명하기

연산을 그대로 읽어 주는 주석보다 왜 그 단위나 예외 처리가 필요한지 기록합니다. 다음은 원천 설명과 실제 저장 단위가 다른 가상의 사례이며 PySpark 3.5 이상의 timestamp_millis를 사용합니다.

from pyspark.sql import functions as F

MILLISECOND_COLUMNS = {"start_date", "delivery_date"}


def normalize_source_timestamps(df):
    # 원천 명세는 날짜라고 부르지만 실제 값은 epoch 밀리초이며 소비자는 Timestamp를 받습니다.
    return df.select(*[
        F.timestamp_millis(F.col(name)).alias(name)
        if name in MILLISECOND_COLUMNS else F.col(name)
        for name in df.columns
    ])

조인 방식과 이름을 명시하기

기본 inner를 쓰더라도 how="inner"처럼 키워드로 의도를 표시합니다. 오른쪽 조인을 읽기 어렵게 만드는 상황에서는 입력 순서를 바꾸어 왼쪽 조인으로 표현합니다.

같은 이름의 공통 키는 문자열 on으로 지정하면 키가 중복되지 않습니다. 여러 입력의 같은 이름을 구분할 때는 무조건 전체 열을 바꾸기보다 DataFrame alias와 마지막 선택을 사용합니다.

from pyspark.sql import functions as F


def flight_parking_times(flights, parking):
    flights = flights.alias("flights")
    parking = parking.alias("parking")
    joined = flights.join(parking, on="flight_code", how="left")
    return joined.select(
        F.col("flights.start_time").alias("flight_start_time"),
        F.col("flights.end_time").alias("flight_end_time"),
        F.col("parking.total_time").alias("client_parking_total_time"),
    )

필요 없는 겹치는 열은 조인 전에 제거할 수 있고 양쪽을 모두 써야 하면 필요한 쪽을 이름 변경하거나 alias로 선택합니다. 출력 전에 모호한 이름을 해소합니다.

오른쪽 키에 여러 일치 행이 있으면 왼쪽 행이 증식합니다. 의도한 관계인지 확인하고, 예상 밖 중복을 숨기려고 distinct나 dropDuplicates를 추가하지 않습니다. 의도적으로 중복을 제거하는 요구사항과 조인 오류를 구분합니다.

체인을 의미별로 나누기

열 선택/필터, 값 계산, 다른 입력과 조인하는 단계를 별도 블록으로 나눕니다. 한 블록의 체인은 보통 3~5개 문장을 기준으로 검토하며, 세 개를 넘는 긴 의미 단위는 이름 있는 함수로 분리할 후보입니다.

from pyspark.sql import functions as F

REQUIRED_VALUE = "truthiness"


def join_customers_with_shipping_address(customers, lookup):
    customers = (
        customers.select("a", "b", "c", "key")
        .filter(F.col("a") == REQUIRED_VALUE)
    )
    customers = customers.withColumn("boverc", F.col("b") / F.col("c"))
    return customers.join(lookup, "key", how="inner")

예제의 b/c는 숫자이며 c는 0이 아니어야 합니다. lookup의 키 외 열은 결과 이름과 충돌하지 않게 준비합니다. 의미별 블록을 만들면 검토할 계산을 찾거나 IDE에서 함수를 추출하고 이동하기 쉬워집니다.

여러 줄은 전체 표현식을 괄호로 감싸고 줄 끝의 역슬래시를 사용하지 않는 방식으로 통일합니다.

from pyspark.sql import functions as F


def executed_with_tests(df):
    return (
        df.filter(F.col("event") == "executing")
        .filter(F.col("has_tests") == F.lit(True))
        .drop("has_tests")
    )

수집, 테스트와 일반 기준

  • 처리 로직은 가능한 한 분산 DataFrame/Column 연산으로 유지합니다. collect, first, head, take, show는 driver로 결과를 가져오는 호출이며, 특히 전체 collect는 메모리 부족을 일으킬 수 있습니다.
  • 개발 중에는 Preview & Debug를 우선 사용합니다. 작은 제어 결과나 제한된 조사 표본을 수집해야 하면 상한과 action 비용을 명확히 합니다.
  • 같은 로직을 기본 함수로 표현할 수 있으면 UDF보다 우선 사용합니다.
  • 파일 250행, 함수 70행을 검토 기준으로 삼고 커진 변환은 자연스러운 처리 단계로 나눕니다. 실행 가능한 언어의 한도가 아니라 코드 구성 기준입니다.
  • 로컬 테스트를 실행하고 새 로직을 검사합니다. 로컬 실행이 어려운 경우에는 자신의 브랜치에서 빌드한 데이터를 직접 확인합니다.
  • 매핑에서 알 수 없는 키를 일반 otherwise 값으로 합치면 누락을 감출 수 있습니다. 나머지 분기의 의미와 오류 정책을 정합니다.
  • 주석 처리한 코드, 사용하지 않는 함수/모듈을 저장소에 남겨 두기보다 Git 이력과 브랜치를 사용합니다.
  • 변수와 함수 이름은 실제 계산을 설명하게 붙입니다. 익숙한 functions as F, types as T 같은 별칭을 유지하고 불필요한 새 import 별칭을 늘리지 않습니다.
  • 필터 값과 상태 문자열 및 계산 상수에는 의미 있는 이름을 붙여 저장소 전반에서 같은 정의를 재사용합니다.