DataFrame 조인

left.join(right, on, how)는 다른 DataFrame이나 자기 자신과 행을 연결합니다. on에는 공통 열 이름, 열 이름 목록 또는 불리언 조건식을 전달할 수 있습니다.

같은 이름의 키

def left_join(left, right):
    return left.join(right, "key", "left")


def multi_key_join(left, right):
    return left.join(right, ["column1", "column2", "column3"], "left")

문자열 이름으로 공통 키를 지정하면 결과의 키 열은 하나로 합쳐집니다. 두 입력에서 키 이름이 다르면 먼저 이름을 맞춘 뒤 이 형식을 사용할 수 있습니다. 왼쪽 조인은 일치하는 오른쪽 행이 없어도 왼쪽 행을 남깁니다. 오른쪽에 같은 키가 여러 개면 왼쪽 행이 여러 결과와 결합할 수 있습니다.

키 외 열 이름 충돌

키 이외의 같은 이름은 자동으로 하나로 합쳐지지 않습니다. 조인 전에 필요 없는 열을 제거하거나 이름을 바꿉니다. 전체 열에 접두사를 붙이는 다음 함수는 하나의 select에서 이름을 정합니다.

from pyspark.sql import functions as F


def prefix_columns(df, prefix="some_prefix_"):
    return df.select(*[
        F.col("`" + name.replace("`", "``") + "`").alias(prefix + name)
        for name in df.columns
    ])

키까지 이름이 바뀌므로 조인 조건에서도 새 이름을 사용해야 합니다. 이미 조인해 중복 이름이 생긴 뒤에는 단순 이름 참조가 모호할 수 있습니다.

여러 관계를 결합한 조인

다음 조건은 ID가 같고 왼쪽 start가 오른쪽 end보다 빠르며, 왼쪽 X가 some_value이면 오른쪽 Y가 some_other_value여야 한다는 규칙입니다. X가 그 값이 아닌 경우 Y에 대한 조건은 True로 둡니다.

from pyspark.sql import functions as F


def conditional_left_join(left, right):
    left = left.alias("l")
    right = right.alias("r")
    same_id = F.col("l.ID") == F.col("r.ID")
    date_order = F.col("l.start") < F.col("r.end")
    conditional_match = F.when(
        F.col("l.X") == "some_value",
        F.col("r.Y") == "some_other_value",
    ).otherwise(True)
    joined = left.join(right, same_id & date_order & conditional_match, "left")
    return joined.select(
        "l.*", F.col("r.end").alias("right_end"), F.col("r.Y").alias("right_y")
    )

입력에는 예제의 ID/start/X 및 ID/end/Y 열이 있어야 합니다. 결과는 왼쪽 열 전체와 필요한 오른쪽 end/Y를 충돌 없는 이름으로 선택합니다. 불리언 조건으로 조인한 뒤에는 결과에 남길 키와 열을 명시하는 편이 분명합니다.

Cross join

crossJoin()은 두 입력의 모든 행 조합을 만듭니다. 행 수가 각각 N과 M이면 결과는 N×M이므로 메모리와 계산량이 크게 늘 수 있습니다. 조인 직후 조건으로 대부분의 행을 제거하려는 목적이면 그 조건을 join의 on에 넣습니다.

Foundry 저장소의 Spark 설정에서 CROSS_JOIN_ENABLED 프로필을 가져오고 configure에 적용합니다. 다음 예제에서 상품 표와 지역 표의 열 이름은 서로 겹치지 않아야 합니다.

from transforms.api import Input, Output, configure, transform_df


@configure(profile=["CROSS_JOIN_ENABLED"])
@transform_df(
    Output("/Example Project/Outputs/product_regions"),
    products=Input("/Example Project/Inputs/products"),
    regions=Input("/Example Project/Inputs/regions"),
)
def compute(products, regions):
    return products.crossJoin(regions)