Saltar a contenido

Partition Writer

DKOps.table_governance.writers.partition_writer

partition_writer.py

Reemplaza una partición específica sin tocar el resto de la tabla.

Databricks → dynamic partition overwrite + saveAsTable Local PC → dynamic partition overwrite + save(path) + refresh catálogo

Classes

PartitionWriter

Bases: BaseWriter

Overwrite de partición específica — idempotente en esa partición.

Uso
PartitionWriter(contract).write(df, partition={"fecha": "2024-01-15"})
Source code in src/DKOps/table_governance/writers/partition_writer.py 
18
19
20
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
65
66
67
68
69
70
71
72
73
74
75
class PartitionWriter(BaseWriter):
    """
    Overwrite de partición específica — idempotente en esa partición.

    Uso
    ---
        PartitionWriter(contract).write(df, partition={"fecha": "2024-01-15"})
    """

    @log_operation("overwrite_partition")
    def write(
        self,
        df:        DataFrame,
        partition: dict[str, str] | None = None,
        **kwargs,
    ) -> None:
        if not partition:
            raise ValueError(
                "PartitionWriter requiere 'partition'. "
                "Ejemplo: writer.write(df, partition={'fecha': '2024-01-15'})"
            )

        declared = set(self._contract.partition_columns)
        for col in partition:
            if col not in declared:
                raise ValueError(
                    f"Columna '{col}' no es columna de partición en "
                    f"'{self._table_name}'.\n"
                    f"Particiones declaradas: {sorted(declared)}"
                )

        self.log.info(
            f"Iniciando OVERWRITE PARTITION | tabla='{self._table_name}' | "
            f"partición={partition}"
        )

        self._validate(df)
        df = self._apply_defaults(df)
        df = self._reorder_columns(df)

        if self._dry_run:
            self._log_dry_run("overwrite_partition")
            return

        # Dynamic partition overwrite — solo reemplaza las particiones del DF
        self._spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

        row_count = df.count()
        self._write_df(df, mode="overwrite")

        self._spark.conf.set("spark.sql.sources.partitionOverwriteMode", "static")

        self.log_write_ok(
            "overwrite_partition",
            rows=row_count,
            target=self._table_name,
            mode=f"overwrite partition={partition}",
        )
Functions
write(df, partition=None, **kwargs)
Source code in src/DKOps/table_governance/writers/partition_writer.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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@log_operation("overwrite_partition")
def write(
    self,
    df:        DataFrame,
    partition: dict[str, str] | None = None,
    **kwargs,
) -> None:
    if not partition:
        raise ValueError(
            "PartitionWriter requiere 'partition'. "
            "Ejemplo: writer.write(df, partition={'fecha': '2024-01-15'})"
        )

    declared = set(self._contract.partition_columns)
    for col in partition:
        if col not in declared:
            raise ValueError(
                f"Columna '{col}' no es columna de partición en "
                f"'{self._table_name}'.\n"
                f"Particiones declaradas: {sorted(declared)}"
            )

    self.log.info(
        f"Iniciando OVERWRITE PARTITION | tabla='{self._table_name}' | "
        f"partición={partition}"
    )

    self._validate(df)
    df = self._apply_defaults(df)
    df = self._reorder_columns(df)

    if self._dry_run:
        self._log_dry_run("overwrite_partition")
        return

    # Dynamic partition overwrite — solo reemplaza las particiones del DF
    self._spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

    row_count = df.count()
    self._write_df(df, mode="overwrite")

    self._spark.conf.set("spark.sql.sources.partitionOverwriteMode", "static")

    self.log_write_ok(
        "overwrite_partition",
        rows=row_count,
        target=self._table_name,
        mode=f"overwrite partition={partition}",
    )

Functions