ReservationMetricsTask — Developer Reference

cancellation_window = abs(datediff(check_in_date, cancellation_date))

booking_window = abs(datediff(check_in_date, booking_date))

stay_nights = abs(datediff(check_out_date, check_in_date))

def _run(self):
    df = self.get_df_from_input(CleanReservationModel)
    
    # Cancellation window
    if "cancellation_date" in df.columns:
        df = df.withColumn(
            "cancellation_window",
            F.when(
                F.col("cancellation_date").isNotNull(),
                F.abs(F.datediff("check_in_date", "cancellation_date"))
            )
        )
    
    # Booking window
    df = df.withColumn(
        "booking_window",
        F.when(
            (F.col("check_in_date") >= F.current_date()) | 
            F.col("status").isin(["checked_in", "checked_out"]),
            F.abs(F.datediff("check_in_date", "booking_date"))
        ).otherwise(F.lit(None))
    )
    
    # Stay nights
    df = df.withColumn(
        "stay_nights",
        F.when(
            (F.col("check_in_date") >= F.current_date()) | 
            F.col("status").isin(["checked_in", "checked_out"]),
            F.abs(F.datediff("check_out_date", "check_in_date"))
        ).otherwise(F.lit(None))
    )
    
    self.write_to_output(ProcessedReservationModel, df.checkpoint(eager=True))
 
## Incremental Merge
 
When new reservations are computed incrementally, they are written to `ProcessedAddedReservationModel`. During merge, the task aligns schemas between existing processed reservations and added reservations, unions them, and keeps the most recent row per `res_id` based on `last_modified_timestamp`.