PySpark 문법 빠른 참조

각 식의 df는 실제 Spark DataFrame입니다. 표의 열 이름과 입력 자료형을 자신의 schema에 맞추며, 아래 함수들은 서로 독립적인 사용 예입니다.

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

Code Workbook은 print로 Output 영역에 기록할 수 있습니다. Code Repositories는 logging.getLogger(__name__)에서 얻은 logger의 info/warning/error를 사용합니다. 로그 참조에 전체 예제가 있습니다.

행 필터와 정렬

조건마다 괄호를 사용합니다. 필터는 NULL, 리터럴, LIKE와 경계 포함 여부를 설명합니다.

조인과 lookup

다른 키 이름으로 조인한 뒤에는 중복되거나 충돌한 열을 선택/이름 변경합니다. Cross join은 N×M개 조합을 만들며 Foundry에서는 CROSS_JOIN_ENABLED 프로필을 가져와 적용합니다. 완전한 교차 조인 선언에 예제가 있습니다.

다음 helper는 왼쪽 키를 lookup의 값으로 바꾸되 일치가 없거나 대체 값이 NULL이면 원래 키를 유지합니다. lookup 키가 유일하고 키/대체 값이 호환되는 자료형이어야 합니다. 예시의 코드와 설명은 문자열입니다.

from pyspark.sql import functions as F


def lookup_and_replace(df1, df2, df1_key, df2_key, df2_value):
    left = df1.alias("original")
    right_names = list(dict.fromkeys([df2_key, df2_value]))
    right = df2.select(*right_names).alias("lookup")
    joined = left.join(right, left[df1_key] == right[df2_key], how="left")
    return joined.select(*[
        F.coalesce(right[df2_value], left[name]).alias(name)
        if name == df1_key else left[name]
        for name in df1.columns
    ])


def replace_pay_codes(people, pay_codes):
    return lookup_and_replace(people, pay_codes, "id", "pay_code_id", "pay_code_desc")

결과에는 왼쪽 표의 열만 남습니다. 양쪽 키 이름이 같아도 출력 키를 잘못 drop하지 않습니다. 오른쪽 키에 여러 행이 있으면 행이 증식하므로 임의 중복 제거로 숨기지 않습니다.

열 생성, 선택과 정리

PASS, Tom, 0과 N/A는 이 문법 예제의 명시적 값입니다. PASS가 실제 검사의 성공을 뜻하거나 Tom이 결측 이름의 추정값이라는 의미는 아닙니다. 실제 처리에서는 대체값의 업무 의미를 정하고, 단순 빈 열은 자료형을 가진 NULL로 만듭니다.

from pyspark.sql import functions as F


def display_full_name(df):
    complete = F.col("fname").isNotNull() & F.col("lname").isNotNull()
    display = F.when(complete, F.concat("fname", "lname")).otherwise("N/A")
    return df.withColumn("full_name", display)


def shared_columns(df, reference):
    existing = set(df.columns)
    names = [name for name in reference.columns if name in existing]
    return df.select(*[
        F.col("`" + name.replace("`", "``") + "`")
        for name in names
    ])


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

shared_columns는 참조 표의 열 순서로 정확한 이름의 교집합을 선택하며 자료형까지 같다는 검사는 아닙니다. 이름 정리 후 충돌이 생기지 않는 입력에 사용합니다. 상세 API는 열 참조에 있습니다.

문자열 필터

각 조건을 filter/where에 전달합니다.

문자열 값 변환

문자열 concat은 인자 중 NULL이 있으면 NULL이며 concat_ws는 NULL 인자를 건너뜁니다. substr 위치는 1부터 시작합니다. 정규식 [0-9]*는 길이 0의 매치도 가능하므로 숫자가 없으면 빈 문자열이 나올 수 있습니다. 문자열 참조에 인코딩과 다른 문자열 함수가 있습니다.

숫자 연산

표현식에 alias를 붙이거나 withColumn에 연결합니다. 수학 참조는 반올림 방식과 로그/난수를 다룹니다.

날짜와 Timestamp

from datetime import date
from pyspark.sql import functions as F


def birth_dates_in_range(df):
    parsed = df.withColumn("date_of_birth", F.to_date("date_of_birth", "yyyy-MM-dd"))
    return parsed.filter(
        (F.col("date_of_birth") >= F.lit(date(2017, 5, 10)))
        & (F.col("date_of_birth") <= F.lit(date(2018, 7, 21)))
    )

양쪽 날짜 경계를 포함합니다. epoch 단위와 시간대 및 소수 초는 날짜 참조에서 구분합니다.

배열, map과 struct

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


def nested_values(df):
    return (
        df.withColumn("full_name", F.array("fname", "lname"))
        .withColumn("empty_array_column", F.array().cast(T.ArrayType(T.StringType())))
        .withColumn("guardians", F.array("guardian_1", "guardian_2"))
        .withColumn("properties", F.struct("hair_color", "eye_color"))
    )


def first_guardian_and_color(df):
    return df.select(
        F.element_at("guardians", 1).alias("first_guardian"),
        F.col("properties.hair_color").alias("hair_color"),
    )


def guardian_rows(df):
    return df.select("child_name", F.explode("guardians").alias("guardian"))


def property_columns(df):
    return df.select("child_name", "properties.*")


def property_rows(df):
    properties = F.create_map(
        F.lit("hair_color"), F.col("properties.hair_color"),
        F.lit("eye_color"), F.col("properties.eye_color"),
    )
    return df.select("child_name", F.explode(properties).alias("property", "value"))

빈 배열은 []이며 빈 문자열을 원소로 가진 [""]와 다릅니다. element_at의 배열 인덱스는 1부터 시작하고 음수는 끝에서 셉니다. 0은 오류입니다. 범위 밖 배열 접근은 ANSI 설정에 따라 NULL 또는 오류이며, map에 없는 키는 NULL입니다.

explode는 array/map을 행으로 펼칩니다. struct는 properties.*로 열을 펼치거나 위처럼 실제 map으로 만들어 행으로 바꿉니다. 중첩 자료와 JSON에서 관련 함수를 설명합니다.

집계

from pyspark.sql import functions as F


def residents_by_address(df):
    return df.groupBy("address").agg(
        F.count("uuid").alias("num_residents"),
        F.max("age").alias("oldest_age"),
        F.first("city", ignorenulls=True).alias("city"),
    )


def resident_names(df):
    return df.groupBy("address").agg(F.collect_set("name").alias("resident_names"))

첫 예제는 같은 address의 city가 일관된다는 전제입니다. count는 대상 열의 비NULL 값 수이며, sum/mean/max/min은 각각 한 열을 집계합니다. 여러 열 조합의 고유 개수는 countDistinct로 구분합니다. collect_list는 중복을 유지하고 collect_set은 제거하며 배열 순서는 보장하지 않습니다. 집계와 피벗에 전체 함수가 있습니다.

Window

from pyspark.sql import functions as F
from pyspark.sql.window import Window


def customer_windows(df):
    customer = Window.partitionBy("customer")
    ordered = customer.orderBy("spend")
    return df.select(
        "*",
        F.avg("spend").over(customer).alias("average_spend"),
        F.max("spend").over(customer).alias("max_spend"),
        F.rank().over(ordered).alias("rank"),
        F.dense_rank().over(ordered).alias("dense_rank"),
        F.row_number().over(ordered).alias("row_number"),
        F.lead("spend", 1).over(ordered).alias("next_value"),
        F.lag("spend", 1).over(ordered).alias("prev_value"),
    )

그룹 전체 집계와 순서가 있는 계산을 다른 window로 정의합니다. 지출이 같은 행 사이의 row_number 및 이전/다음 행 순서는 추가 동률 기준이 없으면 정해지지 않습니다. Window 참조에 frame과 안정적인 순서 조건이 있습니다.

파티션과 UDF

df.repartition(1)은 셔플하여 파티션을 하나로 재구성합니다. 큰 데이터의 병렬성을 줄일 수 있으며 물리 출력 파일 수까지 항상 하나라는 계약은 아닙니다.

UDF는 반환형을 지정합니다. 다음 age는 정수이고 배수 결과가 LongType 범위에 들어간다고 가정합니다. 이름 생성 예제는 실제 사람의 이름을 추정하는 기능이 아니라 가상의 네 값 중 무작위로 선택하는 함수입니다.

import random

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


times_two_udf = F.udf(
    lambda value: None if value is None else value * 2,
    T.LongType(),
)
random_name_udf = F.udf(
    lambda: random.choice(["Bob", "Tom", "Amy", "Jenna"]),
    T.StringType(),
).asNondeterministic()


def custom_columns(df):
    return (
        df.withColumn("age", times_two_udf(F.col("age")))
        .withColumn("name", random_name_udf())
    )

F.udf의 기본 반환형은 StringType이므로 숫자 반환을 명시합니다. 무작위 함수에는 asNondeterministic을 적용합니다. 같은 곱셈은 기본 열 연산으로 더 간단하게 표현할 수 있으며, UDF 참조와 작성 스타일에 선택 기준이 있습니다.