열 선택, 생성과 자료형 변환

pyspark.sql.Column은 열이나 계산식을 나타냅니다. F.col("name")과 F.column("name")으로 표현식을 참조하는 것과, select()로 출력에 남길 열과 순서를 정하는 것은 서로 다른 작업입니다.

이름과 자료형 확인

df.columns는 열 이름의 Python 목록을, df.dtypes는 이름과 자료형 문자열의 튜플 목록을 반환합니다. 예를 들어 age가 Integer이고 name이 String인 표에서는 각각 ["age", "name"], [("age", "int"), ("name", "string")] 형태입니다. 전체 StructType은 df.schema로 확인합니다.

select

가상의 입력 표를 사용합니다.

select("phone_number")는 전화번호 열만 남깁니다. 여러 열을 선택하는 다음 방식들은 이름이 일의적으로 해석되는 입력에서 같은 열과 순서를 선택합니다.

from pyspark.sql import functions as F


def select_by_names(df):
    return df.select("id", "first_name", "last_name")


def select_by_expressions(df):
    return df.select(F.col("id"), F.col("first_name"), F.col("last_name"))


def select_by_list(df):
    names = ["id", "first_name", "last_name"]
    return df.select(names)


def select_by_unpacking(df):
    names = ["id", "first_name", "last_name"]
    return df.select(*names, "phone_number")

마지막 함수의 *names는 목록을 여러 인자로 펼치므로 전화번호까지 네 열을 선택합니다. 출력 열은 선택한 순서대로 배치되고 선택하지 않은 열은 빠집니다.

기존 열 외에도 계산한 표현식을 바로 선택할 수 있습니다. 다음 함수는 string1/string2/string3 열을 받습니다.

from pyspark.sql import functions as F


def select_derived(df):
    combined = F.concat_ws(":", F.col("string1"), F.col("string2"))
    return df.select("string3", combined.alias("derived"))

실제 선택 결과는 string3와 derived 두 열입니다. 열 이름의 대소문자 해석은 Spark 설정에 영향을 받으므로 대소문자만 다른 이름에 의존하지 않습니다.

withColumn

df.withColumn(name, expression)은 새 열을 만들거나 같은 이름의 열을 교체한 DataFrame을 반환합니다. 원래 DataFrame 객체의 행을 제자리에서 수정하지 않습니다.

from pyspark.sql import functions as F


def derive_columns(df):
    return (
        df.withColumn("times_two", F.col("number") * 2)
        .withColumn("concat", F.concat(F.col("string1"), F.col("string2")))
        .withColumn("start_date", F.col("start_timestamp").cast("date"))
        .withColumn("string", F.lower(F.col("string")))
    )

각 식은 이름이 같은 현재 열에 적용됩니다. 예제 입력은 number 숫자 열, string/string1/string2 문자열 열과 start_timestamp 시각 열을 가져야 합니다.

이름 변경과 alias

withColumnRenamed("old_name", "new_name")은 열 이름을 바꿉니다. 새 이름이 기존 열과 충돌하지 않는 경우에는 같은 값을 새 열로 만든 뒤 이전 열을 지우는 방식도 사용할 수 있습니다.

from pyspark.sql import functions as F


def rename_column(df):
    return df.withColumnRenamed("old_name", "new_name")


def copy_name_then_drop(df):
    return df.withColumn("new_name", F.col("old_name")).drop("old_name")

Column.alias()와 그 별칭 Column.name()은 계산식이나 집계 결과의 이름을 정합니다.

from pyspark.sql import functions as F


def named_aggregates(df):
    return df.groupBy("group").agg(
        F.sum("number").alias("sum_of_numbers"),
        F.count("*").name("count"),
    )

여러 이름을 바꿀 때는 원래 이름을 기준으로 한 번 선택하면 이름 변경이 연쇄적으로 적용되는 것을 피할 수 있습니다.

from pyspark.sql import functions as F


def rename_columns(df):
    renames = {"column": "column_renamed", "data": "data_renamed"}
    return df.select(*[
        F.col("`" + name.replace("`", "``") + "`").alias(renames.get(name, name))
        for name in df.columns
    ])

drop

drop(*cols)는 지정 열을 제거한 DataFrame을 반환합니다. 문자열 이름이 schema에 없으면 그 이름에 대한 삭제는 아무 변화 없이 끝납니다. 남길 열을 select하는 방식과 반대로 버릴 열을 지정합니다.

from pyspark.sql import functions as F


def drop_phone(df):
    return df.drop("phone_number")


def drop_identity_by_name(df):
    return df.drop("id", "first_name", "last_name")


def drop_identity_by_column(df):
    return df.drop(F.col("id"), F.col("first_name"), F.col("last_name"))


def drop_from_list(df):
    names = ["id", "first_name", "last_name"]
    return df.drop(*names, "phone_number")

drop에 목록을 전달할 때는 *로 펼칩니다. 마지막 예제의 결과에는 zip_code와 is_active_member가 남습니다.

문자열 drop은 이름 자체를 문자 그대로 해석하고, Column drop은 표현식을 비교합니다. 조인 뒤 같은 이름의 열이 여러 개 있으면 문자열 이름으로 모두 제거할 수 있지만 F.col("name")은 모호한 참조 오류가 날 수 있습니다. 점이 포함된 실제 열 이름과 중첩 필드 표현식도 이 차이에 영향을 받습니다.

cast

Column.cast()는 자료형 객체나 문자열 타입 이름을 받아 새 표현식을 만듭니다.

from pyspark.sql import functions as F
from pyspark.sql.types import StringType


def age_as_string(df):
    return df.select(F.col("age").cast(StringType()).alias("age"))


def age_as_named_type(df):
    return df.select(F.col("age").cast("string").alias("age"))

Integer 값 2와 5는 각각 문자열 "2", "5"가 됩니다. 숫자의 범위를 줄이거나 시각을 날짜로 바꾸면 정보가 줄어들 수 있습니다. 변환 가능성 및 잘못된 값의 오류/NULL 처리는 자료형과 ANSI 설정에 따라 확인합니다.

자주 사용하는 타입은 NullType, StringType, BinaryType, BooleanType, DateType, TimestampType, DecimalType, DoubleType, FloatType, ByteType, ShortType, IntegerType, LongType, ArrayType, MapType, StructType입니다. StructField는 구조체 필드의 이름/타입/null 허용 여부를 정의하며 일반적인 열 cast 대상과 구분합니다.

when과 otherwise

F.when(condition, value).otherwise(other_value)는 행별 조건식을 만듭니다. condition은 불리언 Column이고 값은 리터럴이나 Column입니다. otherwise를 생략하면 조건과 일치하지 않는 행에 NULL을 반환합니다. when을 여러 개 연결하면 앞에서부터 일치하는 분기를 사용합니다.

from pyspark.sql import functions as F


def classify_people(df):
    valid_last_name = F.when(F.col("last_name") != "", F.col("last_name"))
    age_class = (
        F.when(F.col("age") >= 35, "A")
        .when(F.col("age") >= 21, "B")
        .otherwise("C")
    )
    return (
        df.withColumn("last_name", valid_last_name)
        .withColumn("at_least_21", F.when(F.col("age") >= 21, True).otherwise(False))
        .withColumn("switch", age_class)
        .where(F.col("last_name").isNotNull())
    )

예제는 빈 성을 NULL로 바꾼 뒤 제외하며 숫자 age를 A/B/C로 나눕니다. 열 계산과 함께 사용할 행 필터, 조인, 집계, window, UDF와 로그에서 다음 단계의 처리 방식을 설명합니다. 기본 중복 제거와 정렬은 행 선택을 참고합니다.