언어, 실행 환경과 입출력 자료형

Code Workbook은 Python, R과 Spark SQL을 지원합니다. 같은 Workbook에서도 노드별로 언어를 선택할 수 있으며, 입력 유형이 다르면 필요한 변환을 수행합니다.

지원 언어와 활성화

Python은 현재 환경에 vector-spark-module-py가 있어야 합니다. Control Panel의 프로필에서 Python을 켜면 자동 추가되며 Add package로 직접 넣을 수도 있습니다.

R은 플랫폼 인스턴스의 별도 활성화와 환경의 vector-spark-module-r 패키지가 모두 필요합니다. R 활성화는 self-service로 제공하지 않으므로 Palantir 담당자에게 요청합니다. 패키지는 프로필의 R 체크박스 또는 Add package로 추가합니다.

Python이나 R을 사용하지 않는 프로필에서는 해당 vector-spark-module 패키지를 빼서 환경을 줄일 수 있습니다. 필요할 때 다시 추가합니다.

입력 유형 변환

변환의 Inputs에서 원하는 유형을 선택합니다.

상위가 Pandas DataFrame을 반환하고 하위가 Spark DataFrame을 요구하면 Workbook이 변환합니다. Pandas 결과를 R data.frame으로 전달할 때는 Pandas에서 Spark로, 다시 R로 변환합니다.

Python 입력

R 입력

SQL은 입력과 출력 모두 Spark DataFrame만 지원하며 입력을 SQL 테이블로 읽습니다.

반환값 제한

입력은 여러 개이고 각각 다른 유형일 수 있지만 변환 하나의 반환 출력은 하나뿐입니다.

다른 유형을 반환하면 unsupported type 오류가 발생합니다. None 또는 NULL을 반환하는 노드는 실행할 수 있지만 이를 하위 변환의 입력으로 사용할 수는 없습니다. 파일 출력의 명시적 TransformOutput API는 반환값 대신 출력 핸들을 사용하는 별도 경로입니다.

Python 코드

함수의 인자 이름을 상위 노드 alias와 같게 쓰면 그 결과가 입력으로 전달됩니다. 다음은 Spark 입력의 A 값으로 필터하고 상수 열을 추가합니다.

def child(input_spark_dataframe):
    from pyspark.sql import functions as F

    return (
        input_spark_dataframe
        .filter(F.col("A") == "value")
        .withColumn("new_column", F.lit(2))
    )

Spark와 Pandas 변환

Workbook이 제공하는 Spark session인 spark로 작은 Pandas 데이터를 Spark로 바꿀 수 있습니다.

def pandas_to_spark(pandas_input):
    return spark.createDataFrame(pandas_input)

Spark에서 Pandas로의 toPandas()는 모든 결과를 드라이버로 모읍니다. 큰 입력을 먼저 Spark로 줄여야 합니다. 다음은 name, age만 선택하고 18세 이하로 줄인 뒤 평균과 차이를 계산합니다.

def filtering_before_pandas(input_spark_dataframe):
    from pyspark.sql import functions as F

    small = (
        input_spark_dataframe.select("name", "age")
        .filter(F.col("age") <= 18)
    )
    local = small.toPandas()
    mean_age = local["age"].mean()
    local["age_difference_to_mean"] = local["age"] - mean_age
    return local

정렬한 Pandas 결과를 단일 Spark 파티션으로 바꾸는 구성도 가능합니다.

def single_partition_output(pandas_input):
    import pyspark.pandas as ps

    return ps.from_pandas(pandas_input).to_spark().coalesce(1)

단일 파티션은 저장 후 다른 데이터셋 읽기에서도 행 순서가 보장된다는 계약이 아닙니다. 순서가 필요하면 정렬에 사용할 열을 보존하고 소비 단계에서 정렬합니다.

R 코드

R 변환은 function으로 정의합니다. 아래 입력은 Boolean col_A와 col_B가 있는 R data.frame이며 환경에 dplyr가 설치되어 있어야 합니다.

child <- function(r_dataframe) {
    selected <- dplyr::select(r_dataframe, col_A, col_B)
    filtered <- dplyr::filter(selected, col_A == TRUE)
    dplyr::mutate(filtered, new_column = 1000)
}

SparkR과 로컬 R 변환

다음은 드라이버에 들어갈 만큼 작은 Spark 입력을 R data.frame으로 변환합니다.

as_local_frame <- function(spark_input) {
    SparkR::collect(spark_input)
}

R data.frame을 분산 계산에 쓰려면 SparkR::as.DataFrame(r_df)로 바꿉니다. 큰 결과는 SparkR로 필터와 집계를 수행하고 마지막에 작은 결과만 collect합니다.

output_dataset <- function(spark_input) {
    selected <- SparkR::select(spark_input, "column_A", "column_B")
    filtered <- SparkR::filter(selected, selected$column_A == TRUE)
    local <- SparkR::collect(filtered)
    dplyr::mutate(local, new_column = 1000)
}

입력에는 Boolean column_A가 있어야 합니다. SparkR 연산은 분산 DataFrame의 필터, 열 계산과 집계를 설명합니다.

R의 메모리와 실행 제약

R data.frame으로 읽는 입력은 드라이버 메모리에 수집합니다. 큰 입력은 SparkR, Python 또는 SQL에서 먼저 줄여야 합니다.

R 수집에는 대개 r-arrow를 사용하며 이 경로에서 Long, Array, Map, Struct와 Datetime 유형의 변환 제한이 있습니다. UI에 경고가 나타나면 해당 열을 빼거나 String 등 지원되는 유형으로 바꾸는 방법을 검토합니다.

한 Spark 모듈의 R 실행은 단일 스레드이므로 여러 R job을 동시에 시작해도 순서대로 실행합니다. 대기 중인 job은 Queueing in Code Workbook으로 표시됩니다. 오래 걸리는 작업이나 저장된 데이터셋의 처리는 별도 Spark 모듈을 쓰는 batch build로 실행하면 현재 Workbook과 모듈을 공유한 다른 Workbook의 대화형 작업을 계속할 수 있습니다.

SQL 코드

SQL 노드의 입력은 코드에 alias를 적기만 해서는 연결되지 않습니다. 입력 막대에서 추가하거나 부모 노드의 +로 SQL 하위를 만들어야 합니다.

두 입력 alias가 table_a, table_b이고 양쪽에 col_C가 있다고 가정하면 다음과 같이 조인합니다.

SELECT table_b.col_A, table_b.col_B, table_a.*
FROM table_a
JOIN table_b ON table_a.col_C = table_b.col_C

SQL의 결과는 Spark DataFrame입니다. 원시 파일이나 모델 객체를 SQL 테이블 입력으로 직접 사용할 수는 없습니다.