Solve the following pyspark and delta script due to the next error:

AnalysisException: [UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name `target`.`last_period_wallet_day` cannot be resolved. Did you mean one of the following? [`target`.`period_wallet_day`, `source`.`period_wallet_day`, `target`.`period_id`, `target`.`wallet_id`, `target`.`processed_date`].; line 3 pos 51;
'Join LeftOuter, ((wallet_id#675 = wallet_id#51) AND (period_wallet_day#1321 = 'target.last_period_wallet_day))
:- SubqueryAlias source
:  +- Aggregate [wallet_id#675, period_wallet_day#1321], [wallet_id#675, period_wallet_day#1321, first(period_id#694, false) AS period_id#1349, first(closing_day#1298, false) AS closing_day#1351, coalesce(sum(quantity_payments#1322), cast(0 as bigint)) AS quantity_payments#1353L, coalesce(sum(total_value_payment#1323), cast(0 as double)) AS total_value_payment#1355, coalesce(sum(quantity_consumptions#1324), cast(0 as bigint)) AS quantity_consumptions#1357L, coalesce(sum(total_value_consumption#1325), cast(0 as double)) AS total_value_consumption#1359, coalesce(sum(quantity_commissions#1326), cast(0 as bigint)) AS quantity_commissions#1361L, coalesce(sum(total_value_commission#1327), cast(0 as double)) AS total_value_commission#1363, current_timestamp() AS processed_date#1364]
:     +- Project [period_id#694, wallet_id#675, closing_day#1298, to_date(concat(period_id#694, lpad(cast(closing_day#1298 as string), 2, cast(0 as string))), Some(yyyyMMdd), Some(GMT), false) AS period_wallet_day#1321, CASE WHEN ((transaction_type#687 = PAYMENT) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = PAYMENT))) THEN 1 ELSE 0 END AS quantity_payments#1322, CASE WHEN ((transaction_type#687 = PAYMENT) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = PAYMENT))) THEN amount#690 END AS total_value_payment#1323, CASE WHEN ((transaction_type#687 = CONSUMPTION) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = CONSUMPTION))) THEN 1 ELSE 0 END AS quantity_consumptions#1324, CASE WHEN ((transaction_type#687 = CONSUMPTION) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = CONSUMPTION))) THEN amount#690 ELSE cast(0 as double) END AS total_value_consumption#1325, CASE WHEN ((transaction_type#687 = COMMISSION) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = COMMISSION))) THEN 1 ELSE 0 END AS quantity_commissions#1326, CASE WHEN ((transaction_type#687 = COMMISSION) OR ((transaction_type#687 = ADJUSTMENT) AND (origin#693 = COMMISSION))) THEN amount#690 ELSE cast(0 as double) END AS total_value_commission#1327]
:        +- Project [uuid#674, wallet_id#675, transaction_id#676, attention_point_id#677L, adjustment_id#678L, created_date#679, application_date#680, reference#681, movement_reference#682, movement_description#683, movement_description_direction#684, movement_description_tag#685, movement_description_attention_point#686, transaction_type#687, fly_key#688, fly_key_type#689, amount#690, commission#691, consumption_type#692, origin#693, period_id#694, dayofmonth(cast(application_date#680 as date)) AS closing_day#1298]
:           +- Deduplicate [reference#681, adjustment_id#678L, period_id#694, commission#691, created_date#679, movement_description_tag#685, uuid#674, fly_key#688, consumption_type#692, application_date#680, amount#690, transaction_id#676, movement_description#683, movement_description_attention_point#686, origin#693, movement_reference#682, fly_key_type#689, attention_point_id#677L, movement_description_direction#684, transaction_type#687, wallet_id#675]
:              +- Relation [uuid#674,wallet_id#675,transaction_id#676,attention_point_id#677L,adjustment_id#678L,created_date#679,application_date#680,reference#681,movement_reference#682,movement_description#683,movement_description_direction#684,movement_description_tag#685,movement_description_attention_point#686,transaction_type#687,fly_key#688,fly_key_type#689,amount#690,commission#691,consumption_type#692,origin#693,period_id#694] parquet
+- SubqueryAlias target
   +- Relation [wallet_id#51,period_id#52,period_wallet_day#53,closing_balance#54,total_value_payment#55,total_value_consumption#56,total_value_commission#57,processed_date#58,closing_day#59] parquet


script:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Crear la especificación de ventana para el particionamiento
window_spec = Window.partitionBy("wallet_id").orderBy(F.col("period_wallet_day").desc())

# Paso 1: Obtener el `last_period_wallet_day` asegurándonos de seleccionar solo las columnas requeridas
previous_wallet_by_period = (
    aggregated_df.alias("source")
    .join(
        raw_wallet_dto_by_period_flypass.alias("target"),
        (F.col("source.wallet_id") == F.col("target.wallet_id")),
        "left"
    )
    .selectExpr(
        "target.*",
        "source.period_wallet_day as last_period_wallet_day"  # Renombrar la columna relacionada con "last_period_wallet_day"
    )
    # Crear una columna "rank" por cada partición en "window_spec"
    .withColumn("rank", F.row_number().over(window_spec))
    # seleccionar el último valor del periodo (rank = 1)
    .filter(F.col("rank") == 1)
    .drop("rank")
)
 
aggregated_df_with_closing_balance = (
            aggregated_df.alias("source") 
            .join(
                raw_wallet_dto_by_period_flypass.alias("target"),
                f.expr("""
                    source.wallet_id = target.wallet_id
                    AND source.period_wallet_day = target.last_period_wallet_day
                """),
                "left"
            )
            .withColumn(
                "closing_balance",
                f.coalesce(f.col("target.closing_balance"), f.lit(0)) +
                f.col("source.total_value_payment") -
                f.col("source.total_value_commission") -
                f.col("source.total_value_consumption")
            )
            .select("source.*", "closing_balance"))