Saltar a contenido

TableReader

DKOps.table_governance.readers.table_reader

table_reader.py

Fachada unificada para leer tablas Delta usando el contrato como fuente de configuración (tabla efectiva, columnas de partición, CDF, etc.).

Todos los métodos retornan un pyspark.sql.DataFrame real, por lo que todas las transformaciones, acciones y métodos de escritura nativos de PySpark funcionan sin ningún wrapper adicional.

reader = TableReader(contract)

df = reader.read()                              # tabla completa
df = reader.read(filter="estado = 'ACTIVE'")    # con predicado
df = reader.read_partition({"fecha": "2024-01"})# una partición
df = reader.read_stream()                       # streaming DataFrame
df = reader.read_cdf(starting_version=5)        # Change Data Feed

El Spark y el EnvironmentConfig se obtienen automáticamente del Launcher.current() activo — igual que los writers.

Classes

TableReader

Bases: LoggableMixin

Fachada de lectura para tablas Delta gobernadas por contrato.

Parámetros

contract : TableContract cargado por load_contract().

Notas
  • Todos los métodos retornan un DataFrame de PySpark estándar, por lo que se pueden seguir encadenando .filter(), .join(), .groupBy(), .write, etc. directamente.
  • read_cdf() requiere change_data_feed: true en el contrato. Si no está habilitado lanza ValueError con un mensaje claro.
Source code in src/DKOps/table_governance/readers/table_reader.py 
 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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class TableReader(LoggableMixin):
    """
    Fachada de lectura para tablas Delta gobernadas por contrato.

    Parámetros
    ----------
    contract : TableContract cargado por ``load_contract()``.

    Notas
    -----
    - Todos los métodos retornan un ``DataFrame`` de PySpark estándar,
      por lo que se pueden seguir encadenando ``.filter()``, ``.join()``,
      ``.groupBy()``, ``.write``, etc. directamente.
    - ``read_cdf()`` requiere ``change_data_feed: true`` en el contrato.
      Si no está habilitado lanza ``ValueError`` con un mensaje claro.
    """

    def __init__(self, contract: TableContract) -> None:
        launcher = Launcher.current()
        self._spark    = launcher.spark
        self._env      = launcher.env
        self._contract = contract
        self._table_name: str = (
            contract.full_name
            if self._env._is_databricks
            else f"{contract.schema}.{contract.name}"
        )
        self.log.debug(
            f"TableReader listo | tabla='{self._table_name}' | "
            f"runtime={'databricks' if self._env._is_databricks else 'local-pc'}"
        )

    # ── Lectura completa ──────────────────────────────────────────────────────

    def read(
        self,
        filter:  str | None        = None,
        columns: list[str] | None  = None,
        limit:   int | None        = None,
    ) -> "DataFrame":
        """
        Lee la tabla completa y retorna un DataFrame de PySpark.

        Los parámetros opcionales son atajos de conveniencia; cualquier
        transformación adicional puede aplicarse directamente sobre el DF
        retornado usando la API nativa de Spark.

        Parámetros
        ----------
        filter  : expresión SQL WHERE (sin la palabra WHERE).
                  Ej: ``"estado = 'ACTIVE' AND fecha > '2024-01-01'"``
        columns : lista de columnas a seleccionar. Si ``None``, se retornan
                  todas.
        limit   : número máximo de filas a retornar.

        Devuelve
        --------
        ``pyspark.sql.DataFrame``
        """
        t0 = time.monotonic()
        self.log.info(
            f"▶ read | tabla='{self._table_name}'"
            + (f" | filter='{filter}'" if filter else "")
            + (f" | columns={columns}" if columns else "")
            + (f" | limit={limit}" if limit is not None else "")
        )

        df = self._spark.read.table(self._table_name)

        if columns:
            missing = [c for c in columns if c not in df.columns]
            if missing:
                raise ValueError(
                    f"Columnas no encontradas en '{self._table_name}': {missing}\n"
                    f"Disponibles: {df.columns}"
                )
            df = df.select(*columns)

        if filter:
            df = df.filter(filter)

        if limit is not None:
            if limit < 0:
                raise ValueError(f"limit debe ser >= 0, recibido: {limit}")
            df = df.limit(limit)

        elapsed = time.monotonic() - t0
        self.log.success(
            f"✔ read | tabla='{self._table_name}' | "
            f"elapsed={elapsed:.2f}s"
        )
        return df

    # ── Lectura por partición ─────────────────────────────────────────────────

    def read_partition(self, partition: dict[str, str]) -> "DataFrame":
        """
        Lee una partición específica de la tabla.

        Valida que todas las claves del dict sean columnas de partición
        declaradas en el contrato antes de consultar.

        Parámetros
        ----------
        partition : dict con columna → valor de partición.
                    Ej: ``{"fecha": "2024-01-15"}``
                    Ej: ``{"anio": "2024", "mes": "01"}``

        Devuelve
        --------
        ``pyspark.sql.DataFrame`` filtrado a la partición indicada.

        Lanza
        -----
        ValueError si ``partition`` está vacío o contiene columnas que no
        son de partición en el contrato.
        """
        if not partition:
            raise ValueError("partition no puede estar vacío.")

        contract_parts = set(self._contract.partition_columns)
        invalid = [c for c in partition if c not in contract_parts]
        if invalid:
            raise ValueError(
                f"Columnas no son de partición en '{self._table_name}': {invalid}\n"
                f"Particiones del contrato: {self._contract.partition_columns}"
            )

        conditions = " AND ".join(
            f"`{col}` = '{val}'" for col, val in partition.items()
        )
        self.log.info(
            f"▶ read_partition | tabla='{self._table_name}' | "
            f"filtro='{conditions}'"
        )

        t0 = time.monotonic()
        df  = self._spark.read.table(self._table_name).filter(conditions)
        elapsed = time.monotonic() - t0

        self.log.success(
            f"✔ read_partition | tabla='{self._table_name}' | "
            f"partition={partition} | elapsed={elapsed:.2f}s"
        )
        return df

    # ── Streaming ─────────────────────────────────────────────────────────────

    def read_stream(self) -> "DataFrame":
        """
        Lee la tabla como streaming DataFrame (Delta log incremental).

        El DataFrame retornado es un ``readStream`` estándar de Spark
        Structured Streaming — se puede procesar con ``writeStream``,
        ``.foreachBatch()``, etc.

        Devuelve
        --------
        ``pyspark.sql.DataFrame`` con ``isStreaming == True``.
        """
        self.log.info(f"▶ read_stream | tabla='{self._table_name}'")

        stream = (
            self._spark.readStream
            .format("delta")
            .table(self._table_name)
        )

        self.log.success(
            f"✔ read_stream | tabla='{self._table_name}' | streaming=True"
        )
        return stream

    # ── Change Data Feed ──────────────────────────────────────────────────────

    def read_cdf(
        self,
        starting_version:   int | None = None,
        starting_timestamp: str | None = None,
        ending_version:     int | None = None,
    ) -> "DataFrame":
        """
        Lee el Change Data Feed (CDF) de la tabla.

        El CDF registra cada fila insertada, actualizada o eliminada como
        una entrada separada con columna ``_change_type`` (``insert``,
        ``update_preimage``, ``update_postimage``, ``delete``).

        Requiere ``"change_data_feed": true`` en el contrato —
        el ``CreateWriter`` activa ``delta.enableChangeDataFeed`` en
        ``TBLPROPERTIES`` automáticamente al crear la tabla.

        Parámetros
        ----------
        starting_version   : versión Delta desde la que leer (inclusiva).
        starting_timestamp : timestamp ISO desde el que leer.
                             Solo uno de ``starting_version`` /
                             ``starting_timestamp`` debe pasarse.
        ending_version     : versión Delta hasta la que leer (inclusiva).
                             Si se omite, lee hasta la versión más reciente.

        Devuelve
        --------
        ``pyspark.sql.DataFrame`` con columnas adicionales:
        ``_change_type``, ``_commit_version``, ``_commit_timestamp``.

        Lanza
        -----
        ValueError  si el contrato no tiene ``change_data_feed=True``.
        ValueError  si no se pasa ni ``starting_version`` ni
                    ``starting_timestamp``.
        """
        if not self._contract.change_data_feed:
            raise ValueError(
                f"La tabla '{self._table_name}' no tiene change_data_feed "
                "habilitado en el contrato.\n"
                "Agrega '\"change_data_feed\": true' al contrato JSON y "
                "vuelve a crear la tabla."
            )

        if starting_version is None and starting_timestamp is None:
            raise ValueError(
                "read_cdf requiere 'starting_version' o 'starting_timestamp'."
            )

        if starting_version is not None and starting_timestamp is not None:
            raise ValueError(
                "Pasa solo uno: 'starting_version' o 'starting_timestamp', no ambos."
            )

        self.log.info(
            f"▶ read_cdf | tabla='{self._table_name}' | "
            + (f"starting_version={starting_version}" if starting_version is not None
               else f"starting_timestamp='{starting_timestamp}'")
            + (f" | ending_version={ending_version}" if ending_version is not None else "")
        )

        t0     = time.monotonic()
        reader = (
            self._spark.read
            .format("delta")
            .option("readChangeFeed", "true")
        )

        if starting_version is not None:
            reader = reader.option("startingVersion", starting_version)
        else:
            reader = reader.option("startingTimestamp", starting_timestamp)

        if ending_version is not None:
            reader = reader.option("endingVersion", ending_version)

        df      = reader.table(self._table_name)
        elapsed = time.monotonic() - t0

        self.log.success(
            f"✔ read_cdf | tabla='{self._table_name}' | elapsed={elapsed:.2f}s"
        )
        return df

    def __repr__(self) -> str:
        return f"TableReader({self._contract.full_name!r})"
Methods:
__init__(contract)
Source code in src/DKOps/table_governance/readers/table_reader.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(self, contract: TableContract) -> None:
    launcher = Launcher.current()
    self._spark    = launcher.spark
    self._env      = launcher.env
    self._contract = contract
    self._table_name: str = (
        contract.full_name
        if self._env._is_databricks
        else f"{contract.schema}.{contract.name}"
    )
    self.log.debug(
        f"TableReader listo | tabla='{self._table_name}' | "
        f"runtime={'databricks' if self._env._is_databricks else 'local-pc'}"
    )
read(filter=None, columns=None, limit=None)

Lee la tabla completa y retorna un DataFrame de PySpark.

Los parámetros opcionales son atajos de conveniencia; cualquier transformación adicional puede aplicarse directamente sobre el DF retornado usando la API nativa de Spark.

Parámetros

filter : expresión SQL WHERE (sin la palabra WHERE). Ej: "estado = 'ACTIVE' AND fecha > '2024-01-01'" columns : lista de columnas a seleccionar. Si None, se retornan todas. limit : número máximo de filas a retornar.

Devuelve

pyspark.sql.DataFrame

Source code in src/DKOps/table_governance/readers/table_reader.py 
 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
def read(
    self,
    filter:  str | None        = None,
    columns: list[str] | None  = None,
    limit:   int | None        = None,
) -> "DataFrame":
    """
    Lee la tabla completa y retorna un DataFrame de PySpark.

    Los parámetros opcionales son atajos de conveniencia; cualquier
    transformación adicional puede aplicarse directamente sobre el DF
    retornado usando la API nativa de Spark.

    Parámetros
    ----------
    filter  : expresión SQL WHERE (sin la palabra WHERE).
              Ej: ``"estado = 'ACTIVE' AND fecha > '2024-01-01'"``
    columns : lista de columnas a seleccionar. Si ``None``, se retornan
              todas.
    limit   : número máximo de filas a retornar.

    Devuelve
    --------
    ``pyspark.sql.DataFrame``
    """
    t0 = time.monotonic()
    self.log.info(
        f"▶ read | tabla='{self._table_name}'"
        + (f" | filter='{filter}'" if filter else "")
        + (f" | columns={columns}" if columns else "")
        + (f" | limit={limit}" if limit is not None else "")
    )

    df = self._spark.read.table(self._table_name)

    if columns:
        missing = [c for c in columns if c not in df.columns]
        if missing:
            raise ValueError(
                f"Columnas no encontradas en '{self._table_name}': {missing}\n"
                f"Disponibles: {df.columns}"
            )
        df = df.select(*columns)

    if filter:
        df = df.filter(filter)

    if limit is not None:
        if limit < 0:
            raise ValueError(f"limit debe ser >= 0, recibido: {limit}")
        df = df.limit(limit)

    elapsed = time.monotonic() - t0
    self.log.success(
        f"✔ read | tabla='{self._table_name}' | "
        f"elapsed={elapsed:.2f}s"
    )
    return df
read_partition(partition)

Lee una partición específica de la tabla.

Valida que todas las claves del dict sean columnas de partición declaradas en el contrato antes de consultar.

Parámetros

partition : dict con columna → valor de partición. Ej: {"fecha": "2024-01-15"} Ej: {"anio": "2024", "mes": "01"}

Devuelve

pyspark.sql.DataFrame filtrado a la partición indicada.

Lanza

ValueError si partition está vacío o contiene columnas que no son de partición en el contrato.

Source code in src/DKOps/table_governance/readers/table_reader.py 
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def read_partition(self, partition: dict[str, str]) -> "DataFrame":
    """
    Lee una partición específica de la tabla.

    Valida que todas las claves del dict sean columnas de partición
    declaradas en el contrato antes de consultar.

    Parámetros
    ----------
    partition : dict con columna → valor de partición.
                Ej: ``{"fecha": "2024-01-15"}``
                Ej: ``{"anio": "2024", "mes": "01"}``

    Devuelve
    --------
    ``pyspark.sql.DataFrame`` filtrado a la partición indicada.

    Lanza
    -----
    ValueError si ``partition`` está vacío o contiene columnas que no
    son de partición en el contrato.
    """
    if not partition:
        raise ValueError("partition no puede estar vacío.")

    contract_parts = set(self._contract.partition_columns)
    invalid = [c for c in partition if c not in contract_parts]
    if invalid:
        raise ValueError(
            f"Columnas no son de partición en '{self._table_name}': {invalid}\n"
            f"Particiones del contrato: {self._contract.partition_columns}"
        )

    conditions = " AND ".join(
        f"`{col}` = '{val}'" for col, val in partition.items()
    )
    self.log.info(
        f"▶ read_partition | tabla='{self._table_name}' | "
        f"filtro='{conditions}'"
    )

    t0 = time.monotonic()
    df  = self._spark.read.table(self._table_name).filter(conditions)
    elapsed = time.monotonic() - t0

    self.log.success(
        f"✔ read_partition | tabla='{self._table_name}' | "
        f"partition={partition} | elapsed={elapsed:.2f}s"
    )
    return df
read_stream()

Lee la tabla como streaming DataFrame (Delta log incremental).

El DataFrame retornado es un readStream estándar de Spark Structured Streaming — se puede procesar con writeStream, .foreachBatch(), etc.

Devuelve

pyspark.sql.DataFrame con isStreaming == True.

Source code in src/DKOps/table_governance/readers/table_reader.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def read_stream(self) -> "DataFrame":
    """
    Lee la tabla como streaming DataFrame (Delta log incremental).

    El DataFrame retornado es un ``readStream`` estándar de Spark
    Structured Streaming — se puede procesar con ``writeStream``,
    ``.foreachBatch()``, etc.

    Devuelve
    --------
    ``pyspark.sql.DataFrame`` con ``isStreaming == True``.
    """
    self.log.info(f"▶ read_stream | tabla='{self._table_name}'")

    stream = (
        self._spark.readStream
        .format("delta")
        .table(self._table_name)
    )

    self.log.success(
        f"✔ read_stream | tabla='{self._table_name}' | streaming=True"
    )
    return stream
read_cdf(starting_version=None, starting_timestamp=None, ending_version=None)

Lee el Change Data Feed (CDF) de la tabla.

El CDF registra cada fila insertada, actualizada o eliminada como una entrada separada con columna _change_type (insert, update_preimage, update_postimage, delete).

Requiere "change_data_feed": true en el contrato — el CreateWriter activa delta.enableChangeDataFeed en TBLPROPERTIES automáticamente al crear la tabla.

Parámetros

starting_version : versión Delta desde la que leer (inclusiva). starting_timestamp : timestamp ISO desde el que leer. Solo uno de starting_version / starting_timestamp debe pasarse. ending_version : versión Delta hasta la que leer (inclusiva). Si se omite, lee hasta la versión más reciente.

Devuelve

pyspark.sql.DataFrame con columnas adicionales: _change_type, _commit_version, _commit_timestamp.

Lanza

ValueError si el contrato no tiene change_data_feed=True. ValueError si no se pasa ni starting_version ni starting_timestamp.

Source code in src/DKOps/table_governance/readers/table_reader.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def read_cdf(
    self,
    starting_version:   int | None = None,
    starting_timestamp: str | None = None,
    ending_version:     int | None = None,
) -> "DataFrame":
    """
    Lee el Change Data Feed (CDF) de la tabla.

    El CDF registra cada fila insertada, actualizada o eliminada como
    una entrada separada con columna ``_change_type`` (``insert``,
    ``update_preimage``, ``update_postimage``, ``delete``).

    Requiere ``"change_data_feed": true`` en el contrato —
    el ``CreateWriter`` activa ``delta.enableChangeDataFeed`` en
    ``TBLPROPERTIES`` automáticamente al crear la tabla.

    Parámetros
    ----------
    starting_version   : versión Delta desde la que leer (inclusiva).
    starting_timestamp : timestamp ISO desde el que leer.
                         Solo uno de ``starting_version`` /
                         ``starting_timestamp`` debe pasarse.
    ending_version     : versión Delta hasta la que leer (inclusiva).
                         Si se omite, lee hasta la versión más reciente.

    Devuelve
    --------
    ``pyspark.sql.DataFrame`` con columnas adicionales:
    ``_change_type``, ``_commit_version``, ``_commit_timestamp``.

    Lanza
    -----
    ValueError  si el contrato no tiene ``change_data_feed=True``.
    ValueError  si no se pasa ni ``starting_version`` ni
                ``starting_timestamp``.
    """
    if not self._contract.change_data_feed:
        raise ValueError(
            f"La tabla '{self._table_name}' no tiene change_data_feed "
            "habilitado en el contrato.\n"
            "Agrega '\"change_data_feed\": true' al contrato JSON y "
            "vuelve a crear la tabla."
        )

    if starting_version is None and starting_timestamp is None:
        raise ValueError(
            "read_cdf requiere 'starting_version' o 'starting_timestamp'."
        )

    if starting_version is not None and starting_timestamp is not None:
        raise ValueError(
            "Pasa solo uno: 'starting_version' o 'starting_timestamp', no ambos."
        )

    self.log.info(
        f"▶ read_cdf | tabla='{self._table_name}' | "
        + (f"starting_version={starting_version}" if starting_version is not None
           else f"starting_timestamp='{starting_timestamp}'")
        + (f" | ending_version={ending_version}" if ending_version is not None else "")
    )

    t0     = time.monotonic()
    reader = (
        self._spark.read
        .format("delta")
        .option("readChangeFeed", "true")
    )

    if starting_version is not None:
        reader = reader.option("startingVersion", starting_version)
    else:
        reader = reader.option("startingTimestamp", starting_timestamp)

    if ending_version is not None:
        reader = reader.option("endingVersion", ending_version)

    df      = reader.table(self._table_name)
    elapsed = time.monotonic() - t0

    self.log.success(
        f"✔ read_cdf | tabla='{self._table_name}' | elapsed={elapsed:.2f}s"
    )
    return df
__repr__()
Source code in src/DKOps/table_governance/readers/table_reader.py 
296
297
def __repr__(self) -> str:
    return f"TableReader({self._contract.full_name!r})"