콘솔 탐색과 Global Code

콘솔은 그래프의 입력과 변환을 대화형으로 살펴보는 REPL입니다. 현재 브랜치에서 활성화한 언어마다 콘솔이 하나씩 있습니다.

아래 Variables에서 각 노드를 콘솔에 어떤 입력 유형으로 전달할지 정합니다. Pandas와 Spark DataFrame을 혼동하지 않도록 명령을 실행하기 전에 확인합니다.

DataFrame을 반환한 명령은 + Add to graph로 새 변환에 저장할 수 있습니다. 코드 노드의 선택 부분은 Windows의 Ctrl+Shift+Enter, macOS의 Cmd+Shift+Enter로 콘솔에 보냅니다.

Global Code의 범위

오른쪽 Global Code에서 변수와 함수를 정의하면 Workbook의 같은 언어 코드 변환들이 사용할 수 있습니다. 반복 사용하는 상수나 helper 함수에 적합합니다.

승객 나이 구간 함수

다음은 Titanic 입력의 수치형 Age를 표시 구간으로 바꾸는 Python 함수입니다. Global Code에 정의합니다.

def return_age_bracket(age):
    if age is None:
        return "Not specified"
    if age <= 12:
        return "12 and under"
    if 13 <= age < 19:
        return "Between 13 and 19"
    if 19 <= age < 65:
        return "Between 19 and 65"
    if age >= 65:
        return "65 and over"
    return "N/A"

레이블 문자열과 실제 경계는 별개입니다. 이 예시는 13 이상 19 미만과 19 이상 65 미만을 사용합니다. 12보다 크고 13보다 작은 값이나 NaN은 N/A가 됩니다.

새 Python 변환의 입력을 Spark dataframe으로 지정하고 다음을 작성합니다.

def passengers_by_age_bracket_udf(titanic_dataset):
    from pyspark.sql import functions as F
    from pyspark.sql.types import StringType

    bracket = F.udf(return_age_bracket, StringType())
    return (
        titanic_dataset
        .withColumn("age_bracket", bracket(F.col("Age")))
        .select("Name", "age_bracket")
    )

반환값은 이름과 나이 구간의 두 열입니다.

Spark 기본 표현식으로 계산

이 조건은 Python UDF 없이 Spark의 when으로도 표현할 수 있습니다. Python 실행 경계의 비용을 줄이고 Spark가 처리할 수 있는 표현식으로 유지합니다.

def passengers_by_age_bracket(titanic_dataset):
    from pyspark.sql import functions as F

    age = F.col("Age")
    bracket = (
        F.when(age.isNull(), "Not specified")
        .when(F.isnan(age), "N/A")
        .when(age <= 12, "12 and under")
        .when((age >= 13) & (age < 19), "Between 13 and 19")
        .when((age >= 19) & (age < 65), "Between 19 and 65")
        .when(age >= 65, "65 and over")
        .otherwise("N/A")
    )
    return (
        titanic_dataset
        .withColumn("age_bracket", bracket)
        .select("Name", "age_bracket")
    )

두 방식은 같은 표시 구간을 계산합니다. 실행 시간은 입력과 환경에 따라 다르지만, Spark 기본 표현식으로 표현할 수 있는 단순 조건은 UDF 대신 사용하는 것이 일반적으로 유리합니다.

전역 상태 변경의 격리

변환 안에서 Global Code의 변수나 함수를 바꿔도 그 변경이 다른 변환으로 전파되지는 않습니다. 재현 가능한 결과를 위해 각 변환이 전역 상태의 원래 정의를 사용합니다.

Global Code에 다음 목록을 정의했다고 가정합니다.

my_list = [1, 2, 3, 4]

입력 DataFrame을 그대로 반환하면서 목록을 바꾸는 노드는 다음과 같습니다.

def inspect_global_state(input_df):
    my_list.append(5)
    print(my_list)
    return input_df

이 실행의 콘솔에서는 [1, 2, 3, 4, 5]를 볼 수 있지만 다른 변환에는 [1, 2, 3, 4]가 제공됩니다. 한 노드의 전역 변수 변경을 다른 노드로 값을 전달하는 수단으로 사용하지 않습니다.