로그, 실행 계획과 제한된 데이터 조사

PySpark의 query를 만드는 driver 코드와 실제 행을 계산하는 executor 코드의 실행 위치를 구분합니다. 로그나 디버그 열에 민감한 원문과 인증 정보를 기록하지 않도록 주의합니다.

Code Workbook의 print

Python print는 코드 편집기 오른쪽 Output 영역에 나타납니다. 출력 데이터셋도 반환하는 함수에서는 return을 유지합니다.

def new_dataset(some_input_dataset):
    print("example log output")
    return some_input_dataset

Code Repositories의 logging

Python 표준 logging으로 ERROR, WARNING, INFO 등의 레벨을 구분합니다. 로그는 출력 데이터셋의 Details > Files > Log Files와 Builds의 Build > Job status logs > Driver logs에서 확인할 수 있습니다.

import logging

from transforms.api import Input, Output, transform_df

logger = logging.getLogger(__name__)


@transform_df(
    Output("/Example Project/Outputs/logged_output"),
    source=Input("/Example Project/Inputs/source"),
)
def compute(source):
    logger.info("Input columns: %s", source.columns)
    return source

레벨과 화면의 로그 필터에 따라 표시되는 메시지가 달라집니다. Logger 이름으로 어떤 모듈에서 기록했는지 구분할 수 있습니다.

UDF 안의 값 조사

UDF는 executor에서 실행되므로 그 안의 print/logging이 driver 최상위 로그에 나타나는 것으로 기대하지 않습니다. 필요한 값을 디버그 열로 반환하여 결과를 살펴보는 방법을 사용할 수 있습니다.

import logging

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType, StringType
from transforms.api import Input, Output, transform_df

logger = logging.getLogger(__name__)


@transform_df(
    Output("/Example Project/Outputs/debugged_integers"),
    source=Input("/Example Project/Inputs/integers"),
)
def compute(source):
    logger.info("Constructing integer transformation")

    @F.udf(returnType=IntegerType())
    def add_five(value):
        return None if value is None else value + 5

    @F.udf(returnType=StringType())
    def describe_original(value):
        return f"Original integer was {value} before adding 5"

    return (
        source
        .withColumn("new_integer", add_five(F.col("example_integer_col")))
        .withColumn("debugging", describe_original(F.col("example_integer_col")))
    )

예제는 정수 열을 받으며 +5 결과가 IntegerType 범위에 들어가야 합니다. NULL 계산은 NULL을 유지합니다. 디버그 열도 저장되어 다른 사용자가 읽을 수 있으므로 조사에 필요한 정보만 포함합니다.

열 구조 확인

df.columns는 열 이름 목록입니다. 다음 함수는 조인 전후 이름을 출력하며 결과 DataFrame을 반환합니다.

def employee_phone_numbers(employee_df, phone_number_df):
    print("employee columns are {}".format(employee_df.columns))
    print("phone columns are {}".format(phone_number_df.columns))
    joined = employee_df.join(phone_number_df, "employee_id", "left")
    print("joined columns are {}".format(joined.columns))
    return joined

예를 들어 왼쪽에 name/employee_id, 오른쪽에 phone_number/employee_id가 있으면 공통 키가 하나인 조인 결과에서 name, employee_id와 phone_number를 사용할 수 있습니다. 실제 출력 열 순서는 API가 반환한 columns로 확인합니다.

일대일 조인을 예상할 때 행 수 보기

def inspect_left_join(employee_df, phone_number_df):
    before = employee_df.count()
    print("Incoming employee rows {}".format(before))
    joined = employee_df.join(phone_number_df, "employee_id", "left")
    after = joined.count()
    print("Final employee rows {}".format(after))
    if after > before:
        print("Some employee keys matched multiple right-side rows")
    else:
        print("The join did not increase the row count")
    return joined

가령 입력이 100행이고 조인 결과가 105행이라면 오른쪽 키가 여러 행과 일치한 경우를 조사합니다. 행 수가 같다는 사실만으로 모든 값과 조인 관계가 올바르다고 결론 내리지 않습니다. count는 실제 Spark action이므로 조사를 위해 수행하는 추가 계산 비용도 고려합니다.

실행 계획

df.explain()으로 Spark가 선택한 물리 계획을 볼 수 있습니다.

from pyspark.sql import functions as F


def explain_birthday_join(employee, phone):
    selected = employee.filter(
        F.month("birthday") == F.month(F.current_date())
    )
    joined = selected.join(phone, "employee_id", "left")
    joined.explain()
    return joined

계획에서는 다음 항목을 살펴봅니다.

입력의 크기와 통계 및 설정에 따라 물리 계획은 달라집니다. 계획의 특정 조인 방식이 모든 실행에 고정된다고 가정하지 않습니다.

일부 행만 Python으로 가져오기

다음 함수는 직원별 NULL이 아닌 전화번호 레코드 수를 세고, 둘 이상인 직원 중 상위 세 결과를 출력합니다. 같은 개수에서는 employee_id를 추가 정렬 키로 사용합니다.

from pyspark.sql import functions as F


def multiple_numbers(phone_numbers):
    result = (
        phone_numbers.groupBy("employee_id")
        .agg(F.count("phone_number").alias("numbers"))
        .where(F.col("numbers") > 1)
        .orderBy(F.col("numbers").desc(), F.col("employee_id").asc())
    )
    print(result.take(3))
    return result

가상의 집계 값이 employee_id 70에 4개, 90과 25에 각각 2개라면 위 정렬 기준에서는 70, 25, 90 순서입니다. take(3)는 최대 세 Row를 Python 목록으로 가져옵니다. collect()는 전체 행을 가져오므로 작은 결과에만 사용합니다. 큰 데이터를 수집하면 driver 메모리가 부족해질 수 있습니다.