두 입력을 증분으로 조인하기

주문과 배송이 다른 배치로 도착하면 두 입력의 added끼리만 조인해서는 모든 일치를 찾지 못합니다. 지난 배치의 주문과 이번 배치의 배송이 일치할 수 있기 때문입니다. 한쪽을 전체 참조 표로 읽는 방식이나, 처리하지 않은 키 범위를 양쪽에서 다시 읽는 방식이 필요합니다.

단조 증가하는 주문 ID의 예

가상의 입력은 다음과 같습니다.

완료된 배송 기간은 1001번의 2일입니다. 1002번과 1003번은 아직 배송과 일치하지 않으므로 완료 결과에 기록하지 않습니다.

다음 알고리즘은 양쪽 입력에서 orderId가 엄격하게 증가하는 순서로 추가되고, 각 입력에서 주문 ID가 유일하다는 조건을 사용합니다. 이미 완료 처리한 최대 ID 이하의 주문이나 배송이 나중에 들어오지 않아야 합니다. 단순히 ID가 유일하다는 조건만으로는 충분하지 않습니다.

이 조건을 만족하면 이전 출력의 완료된 최대 ID를 경계로 삼아 양쪽의 더 큰 ID만 조인할 수 있습니다. 입력은 snapshot_inputs로 지정하여 현재 전체 표를 읽고, 코드에서 경계 이후의 범위를 선택합니다.

from pyspark.sql import functions as F
from pyspark.sql import types as T
from transforms.api import Input, Output, incremental, transform

RESULT_SCHEMA = T.StructType([
    T.StructField("orderId", T.IntegerType(), True),
    T.StructField("deliveryDuration", T.IntegerType(), True),
])


@incremental(snapshot_inputs=["orders", "deliveries"])
@transform.spark.using(
    orders=Input("/Example Project/Inputs/orders"),
    deliveries=Input("/Example Project/Inputs/deliveries"),
    delivery_duration=Output("/Example Project/Outputs/delivery_duration"),
)
def compute(orders, deliveries, delivery_duration):
    previous = delivery_duration.dataframe("previous", schema=RESULT_SCHEMA)
    boundary = previous.agg(F.max("orderId").alias("last_id")).first()["last_id"]
    pending_orders = orders.dataframe().select("orderId", "submittedDate")
    pending_deliveries = deliveries.dataframe().select("orderId", "deliveryDate")
    if boundary is not None:
        pending_orders = pending_orders.filter(F.col("orderId") > boundary)
        pending_deliveries = pending_deliveries.filter(F.col("orderId") > boundary)
    completed = (
        pending_orders.join(pending_deliveries, "orderId", "inner")
        .withColumn(
            "deliveryDuration",
            F.datediff(F.col("deliveryDate"), F.col("submittedDate")),
        )
        .select(RESULT_SCHEMA.fieldNames())
    )
    delivery_duration.write_dataframe(completed)

예제의 orderId는 Spark IntegerType이고 날짜 열은 날짜 값입니다. 다른 정수 자료형을 사용하는 입력은 이전 출력 schema도 같은 자료형으로 선언합니다. 첫 실행의 빈 이전 출력에서는 경계를 적용하지 않습니다.

완료된 inner join 결과만 쓰므로 배송이 아직 없는 주문은 다음 실행에서 다시 후보가 됩니다. 미완료 주문을 NULL 배송 기간과 함께 출력해 최대 ID를 올리면 나중에 도착한 배송을 건너뛸 수 있습니다.

양쪽 입력의 ID 도착 순서를 보장할 수 없다면 이 최대 ID 알고리즘을 사용하지 않습니다. 처리하지 않은 키를 별도 상태로 유지하거나 전체 입력과 이전 결과를 키 단위로 비교하는 등 지연 도착을 보존하는 계산이 필요합니다. 변경되거나 삭제되는 주문/배송까지 이 추가 전용 알고리즘으로 처리할 수는 없습니다.