Saltar a contenido

Estrategias Silver

FullMergeStrategy

DKOps.ingestion.strategies.full_merge.FullMergeStrategy

Bases: BasePromotionStrategy

MERGE INTO con deduplicación — mantiene el registro más reciente por clave.

Source code in src/DKOps/ingestion/strategies/full_merge.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class FullMergeStrategy(BasePromotionStrategy):
    """MERGE INTO con deduplicación — mantiene el registro más reciente por clave."""

    def execute(self) -> int:
        self.log.info(
            f"[{self._contract.name}] FullMerge | "
            f"keys={list(self._contract.merge_keys)} | "
            f"watermark={self._contract.watermark_col}"
        )

        bronze_df = self._read_bronze()

        # Deduplicar: quedarse con el registro más reciente por merge_keys
        deduped = self._dedup(bronze_df)

        # Añadir timestamps Silver si el contrato los pide
        deduped = self._add_silver_timestamps(deduped)

        # Seleccionar solo columnas Silver (excluir metadata Bronze)
        deduped = self._select_for_silver(deduped)

        # MERGE INTO Silver
        self._writer.upsert(
            deduped,
            keys                = list(self._contract.merge_keys),
            insert_only_columns = self._silver_insert_only_columns(),
        )

        count = self._dst_reader.read().count()
        self.log.info(f"[{self._contract.name}] FullMerge completado | silver_rows={count:,}")
        return count

    def _dedup(self, df: DataFrame) -> DataFrame:
        keys = list(self._contract.merge_keys)
        wcol = self._contract.watermark_col

        if wcol and wcol in df.columns:
            window = Window.partitionBy(*keys).orderBy(F.col(wcol).desc())
            return (
                df.withColumn("_row_num", F.row_number().over(window))
                  .filter(F.col("_row_num") == 1)
                  .drop("_row_num")
            )

        # Sin watermark: dedup simple por clave
        return df.dropDuplicates(keys)

Methods:

execute()

Source code in src/DKOps/ingestion/strategies/full_merge.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def execute(self) -> int:
    self.log.info(
        f"[{self._contract.name}] FullMerge | "
        f"keys={list(self._contract.merge_keys)} | "
        f"watermark={self._contract.watermark_col}"
    )

    bronze_df = self._read_bronze()

    # Deduplicar: quedarse con el registro más reciente por merge_keys
    deduped = self._dedup(bronze_df)

    # Añadir timestamps Silver si el contrato los pide
    deduped = self._add_silver_timestamps(deduped)

    # Seleccionar solo columnas Silver (excluir metadata Bronze)
    deduped = self._select_for_silver(deduped)

    # MERGE INTO Silver
    self._writer.upsert(
        deduped,
        keys                = list(self._contract.merge_keys),
        insert_only_columns = self._silver_insert_only_columns(),
    )

    count = self._dst_reader.read().count()
    self.log.info(f"[{self._contract.name}] FullMerge completado | silver_rows={count:,}")
    return count

CdcMergeStrategy

DKOps.ingestion.strategies.cdc_merge.CdcMergeStrategy

Bases: BasePromotionStrategy

Aplica eventos CDC (I/U/D) desde Bronze hacia Silver. Mantiene el estado actual de cada entidad con soft deletes.

Source code in src/DKOps/ingestion/strategies/cdc_merge.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class CdcMergeStrategy(BasePromotionStrategy):
    """
    Aplica eventos CDC (I/U/D) desde Bronze hacia Silver.
    Mantiene el estado actual de cada entidad con soft deletes.
    """

    def __init__(
        self,
        spark:        SparkSession,
        contract:     IngestionContract,
        src_contract: TableContract,
        dst_contract: TableContract,
        op_col:       str  = "op_type",
        soft_delete:  bool = True,
    ) -> None:
        super().__init__(spark, contract, src_contract, dst_contract)
        self._op_col      = op_col
        self._soft_delete = soft_delete

    def execute(self) -> int:
        self.log.info(
            f"[{self._contract.name}] CdcMerge | "
            f"keys={list(self._contract.merge_keys)} | "
            f"op_col={self._op_col} | soft_delete={self._soft_delete}"
        )

        bronze_df = self._read_bronze()
        latest_df = self._keep_latest_per_key(bronze_df)

        # Separar inserts/updates de deletes
        deletes = latest_df.filter(F.col(self._op_col) == "D")
        upserts = latest_df.filter(F.col(self._op_col).isin("I", "U"))

        if upserts.count() > 0:
            # Añadir timestamps Silver
            upserts = self._add_silver_timestamps(upserts)

            # Garantizar is_deleted no nulo en los eventos I/U
            upserts = self._ensure_is_deleted(upserts, deleted=False)

            # Seleccionar solo columnas Silver (excluir metadata Bronze)
            upserts = self._select_for_silver(upserts)

            self._writer.upsert(
                upserts,
                keys                = list(self._contract.merge_keys),
                insert_only_columns = self._silver_insert_only_columns(),
            )

        if deletes.count() > 0:
            self._apply_deletes(deletes)

        count = self._dst_reader.read().count()
        self.log.info(f"[{self._contract.name}] CdcMerge completado | silver_rows={count:,}")
        return count

    def _ensure_is_deleted(self, df: DataFrame, deleted: bool) -> DataFrame:
        """
        Garantiza que `is_deleted` llegue a Silver como true o false, nunca NULL.

        Antes solo se rellenaba cuando la columna **faltaba** en el DataFrame.
        Bastaba con que llegara desde Bronze —por ejemplo porque Auto Loader la
        incorporó al esquema— para que el default dejara de aplicarse y los
        nulos se propagaran.

        Un NULL aquí no rompe nada de forma visible, y ese es el problema: el
        filtro natural aguas abajo es `WHERE NOT is_deleted`, y con lógica
        ternaria `NOT NULL` no es `TRUE`, así que esas filas desaparecen del
        resultado sin error alguno.

        Parámetros
        ----------
        deleted : True  → evento D. La baja se marca pase lo que pase en origen.
                  False → evento I/U. Se respeta el valor del origen si lo trae,
                          y solo se rellenan los nulos.
        """
        if "is_deleted" not in self._dst_contract.column_names:
            return df

        if deleted:
            return df.withColumn("is_deleted", F.lit(True))

        if "is_deleted" not in df.columns:
            return df.withColumn("is_deleted", F.lit(False))

        return df.withColumn(
            "is_deleted", F.coalesce(F.col("is_deleted"), F.lit(False))
        )

    def _keep_latest_per_key(self, df: DataFrame) -> DataFrame:
        """Retiene solo el evento más reciente por clave de negocio."""
        keys  = list(self._contract.merge_keys)
        wcol  = self._contract.watermark_col

        if wcol and wcol in df.columns:
            window = Window.partitionBy(*keys).orderBy(F.col(wcol).desc())
            return (
                df.withColumn("_row_num", F.row_number().over(window))
                  .filter(F.col("_row_num") == 1)
                  .drop("_row_num")
            )
        return df.dropDuplicates(keys)

    def _apply_deletes(self, deletes: DataFrame) -> None:
        keys      = list(self._contract.merge_keys)
        dst_table = self._dst_contract.effective_name

        if self._soft_delete and "is_deleted" in self._dst_contract.column_names:
            # Soft delete: marcar is_deleted=True vía upsert
            soft = self._ensure_is_deleted(deletes, deleted=True)

            # Añadir timestamps Silver también en soft-deletes
            soft = self._add_silver_timestamps(soft)

            # Seleccionar solo columnas Silver (excluir metadata Bronze)
            soft = self._select_for_silver(soft)

            self._writer.upsert(
                soft,
                keys                = keys,
                insert_only_columns = self._silver_insert_only_columns(),
            )
        else:
            # Hard delete: DELETE WHERE key IN (...)
            key_col   = keys[0]
            key_vals  = [str(row[key_col]) for row in deletes.select(key_col).collect()]
            quoted    = ", ".join(f"'{v}'" for v in key_vals)
            condition = f"{key_col} IN ({quoted})"
            self._writer.delete(condition)

Methods:

__init__(spark, contract, src_contract, dst_contract, op_col='op_type', soft_delete=True)

Source code in src/DKOps/ingestion/strategies/cdc_merge.py 
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    spark:        SparkSession,
    contract:     IngestionContract,
    src_contract: TableContract,
    dst_contract: TableContract,
    op_col:       str  = "op_type",
    soft_delete:  bool = True,
) -> None:
    super().__init__(spark, contract, src_contract, dst_contract)
    self._op_col      = op_col
    self._soft_delete = soft_delete

execute()

Source code in src/DKOps/ingestion/strategies/cdc_merge.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def execute(self) -> int:
    self.log.info(
        f"[{self._contract.name}] CdcMerge | "
        f"keys={list(self._contract.merge_keys)} | "
        f"op_col={self._op_col} | soft_delete={self._soft_delete}"
    )

    bronze_df = self._read_bronze()
    latest_df = self._keep_latest_per_key(bronze_df)

    # Separar inserts/updates de deletes
    deletes = latest_df.filter(F.col(self._op_col) == "D")
    upserts = latest_df.filter(F.col(self._op_col).isin("I", "U"))

    if upserts.count() > 0:
        # Añadir timestamps Silver
        upserts = self._add_silver_timestamps(upserts)

        # Garantizar is_deleted no nulo en los eventos I/U
        upserts = self._ensure_is_deleted(upserts, deleted=False)

        # Seleccionar solo columnas Silver (excluir metadata Bronze)
        upserts = self._select_for_silver(upserts)

        self._writer.upsert(
            upserts,
            keys                = list(self._contract.merge_keys),
            insert_only_columns = self._silver_insert_only_columns(),
        )

    if deletes.count() > 0:
        self._apply_deletes(deletes)

    count = self._dst_reader.read().count()
    self.log.info(f"[{self._contract.name}] CdcMerge completado | silver_rows={count:,}")
    return count

IncrementalReplaceStrategy

DKOps.ingestion.strategies.incremental_replace.IncrementalReplaceStrategy

Bases: BasePromotionStrategy

Reemplaza la partición más reciente de Silver con el snapshot de Bronze.

Source code in src/DKOps/ingestion/strategies/incremental_replace.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class IncrementalReplaceStrategy(BasePromotionStrategy):
    """Reemplaza la partición más reciente de Silver con el snapshot de Bronze."""

    def execute(self) -> int:
        partition_col = (
            self._contract.watermark_col
            or (self._dst_contract.partition_columns[0] if self._dst_contract.partition_columns else "_ingested_date")
        )

        self.log.info(
            f"[{self._contract.name}] IncrementalReplace | partition_col={partition_col}"
        )

        bronze_df = self._read_bronze()

        # Fecha de la última partición disponible
        max_val_row = bronze_df.select(F.max(F.col(partition_col)).alias("max_val")).collect()
        if not max_val_row or max_val_row[0]["max_val"] is None:
            self.log.warning(f"[{self._contract.name}] Bronze vacío, nada que procesar")
            return 0

        max_val   = max_val_row[0]["max_val"]
        latest_df = bronze_df.filter(F.col(partition_col) == max_val)

        # Añadir timestamps Silver antes de filtrar columnas.
        # Ojo: esta estrategia reescribe la partición entera, así que
        # `_silver_created_at` marca la última reconstrucción de la partición,
        # no la primera vez que se vio la fila.
        latest_df = self._add_silver_timestamps(latest_df)

        # Seleccionar solo columnas Silver (excluir metadata Bronze)
        latest_df = self._select_for_silver(latest_df)

        self._writer.overwrite_partition(
            latest_df,
            partition = {partition_col: str(max_val)},
        )

        count = latest_df.count()
        self.log.info(
            f"[{self._contract.name}] IncrementalReplace completado | "
            f"partition={max_val} | rows={count:,}"
        )
        return count

Methods:

execute()

Source code in src/DKOps/ingestion/strategies/incremental_replace.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def execute(self) -> int:
    partition_col = (
        self._contract.watermark_col
        or (self._dst_contract.partition_columns[0] if self._dst_contract.partition_columns else "_ingested_date")
    )

    self.log.info(
        f"[{self._contract.name}] IncrementalReplace | partition_col={partition_col}"
    )

    bronze_df = self._read_bronze()

    # Fecha de la última partición disponible
    max_val_row = bronze_df.select(F.max(F.col(partition_col)).alias("max_val")).collect()
    if not max_val_row or max_val_row[0]["max_val"] is None:
        self.log.warning(f"[{self._contract.name}] Bronze vacío, nada que procesar")
        return 0

    max_val   = max_val_row[0]["max_val"]
    latest_df = bronze_df.filter(F.col(partition_col) == max_val)

    # Añadir timestamps Silver antes de filtrar columnas.
    # Ojo: esta estrategia reescribe la partición entera, así que
    # `_silver_created_at` marca la última reconstrucción de la partición,
    # no la primera vez que se vio la fila.
    latest_df = self._add_silver_timestamps(latest_df)

    # Seleccionar solo columnas Silver (excluir metadata Bronze)
    latest_df = self._select_for_silver(latest_df)

    self._writer.overwrite_partition(
        latest_df,
        partition = {partition_col: str(max_val)},
    )

    count = latest_df.count()
    self.log.info(
        f"[{self._contract.name}] IncrementalReplace completado | "
        f"partition={max_val} | rows={count:,}"
    )
    return count

AppendDedupStrategy

DKOps.ingestion.strategies.append_dedup.AppendDedupStrategy

Bases: BasePromotionStrategy

Inserta registros de Bronze que no existen en Silver (por merge_keys).

Source code in src/DKOps/ingestion/strategies/append_dedup.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class AppendDedupStrategy(BasePromotionStrategy):
    """Inserta registros de Bronze que no existen en Silver (por merge_keys)."""

    def execute(self) -> int:
        keys = list(self._contract.merge_keys)
        self.log.info(
            f"[{self._contract.name}] AppendDedup | keys={keys}"
        )

        bronze_df = self._read_bronze()

        # Leer Silver (destino) para el anti-join.
        # En el primer run la tabla aún no existe → todos los registros de Bronze son nuevos.
        try:
            silver_df   = self._dst_reader.read()
            new_records = self._filter_new(bronze_df, silver_df, keys)
        except Exception:
            self.log.info(
                f"[{self._contract.name}] Silver aún no existe — "
                f"insertando todos los registros de Bronze como registros nuevos"
            )
            new_records = bronze_df

        count = new_records.count()

        if count == 0:
            self.log.info(f"[{self._contract.name}] AppendDedup: sin registros nuevos")
            return 0

        # Añadir timestamps Silver antes de filtrar columnas.
        # Aquí solo hay inserciones (anti-join), así que created == modified.
        new_records = self._add_silver_timestamps(new_records)

        # Seleccionar solo columnas Silver (excluir metadata Bronze)
        new_records = self._select_for_silver(new_records)
        self._writer.append(new_records)
        self.log.info(f"[{self._contract.name}] AppendDedup completado | new_rows={count:,}")
        return count

    @staticmethod
    def _filter_new(
        source: DataFrame,
        target: DataFrame,
        keys:   list[str],
    ) -> DataFrame:
        """Anti-join: retorna filas de source que no existen en target por keys."""
        target_keys = target.select(*keys).withColumn("_exists", F.lit(True))
        joined = source.join(target_keys, on=keys, how="left")
        return joined.filter(F.col("_exists").isNull()).drop("_exists")

Methods:

execute()

Source code in src/DKOps/ingestion/strategies/append_dedup.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def execute(self) -> int:
    keys = list(self._contract.merge_keys)
    self.log.info(
        f"[{self._contract.name}] AppendDedup | keys={keys}"
    )

    bronze_df = self._read_bronze()

    # Leer Silver (destino) para el anti-join.
    # En el primer run la tabla aún no existe → todos los registros de Bronze son nuevos.
    try:
        silver_df   = self._dst_reader.read()
        new_records = self._filter_new(bronze_df, silver_df, keys)
    except Exception:
        self.log.info(
            f"[{self._contract.name}] Silver aún no existe — "
            f"insertando todos los registros de Bronze como registros nuevos"
        )
        new_records = bronze_df

    count = new_records.count()

    if count == 0:
        self.log.info(f"[{self._contract.name}] AppendDedup: sin registros nuevos")
        return 0

    # Añadir timestamps Silver antes de filtrar columnas.
    # Aquí solo hay inserciones (anti-join), así que created == modified.
    new_records = self._add_silver_timestamps(new_records)

    # Seleccionar solo columnas Silver (excluir metadata Bronze)
    new_records = self._select_for_silver(new_records)
    self._writer.append(new_records)
    self.log.info(f"[{self._contract.name}] AppendDedup completado | new_rows={count:,}")
    return count