apply_changes

laktory.dlt.apply_changes

apply_changes(*args, node=None, **kwargs)

When is_debug() is True read source CDC table from storage, else run native Databricks dlt.apply_changes

RETURNS	DESCRIPTION
	Output dataframe

Examples:

from laktory import dlt
from laktory import models

dlt.spark = spark

def define_table(node):
    dlt.create_streaming_table(name=node.name)
    df = dlt.apply_changes(**node.apply_changes_kwargs)
    return df

define_table(
    models.PipelineNode(
        name="slv_stock_prices",
        source={
            "table_name": "brz_stock_prices",
            "cdc": {
                "primary_keys": ["asset_symbol"],
                "sequence_by": "change_id",
                "scd_type": 2,
            },
        },
        sink={
            "table_name": "brz_stock_prices",
        },
    )
)

Source code in laktory/dlt/__init__.py

def apply_changes(*args, node=None, **kwargs):
    """
    When `is_debug()` is `True` read source CDC table from storage, else run
    native Databricks `dlt.apply_changes`

    Returns
    -------
    :
        Output dataframe

    Examples
    --------
    ```py
    from laktory import dlt
    from laktory import models

    dlt.spark = spark

    def define_table(node):
        dlt.create_streaming_table(name=node.name)
        df = dlt.apply_changes(**node.apply_changes_kwargs)
        return df

    define_table(
        models.PipelineNode(
            name="slv_stock_prices",
            source={
                "table_name": "brz_stock_prices",
                "cdc": {
                    "primary_keys": ["asset_symbol"],
                    "sequence_by": "change_id",
                    "scd_type": 2,
                },
            },
            sink={
                "table_name": "brz_stock_prices",
            },
        )
    )
    ```
    """
    if is_debug():
        if node is None:
            return
        df = node.source.read(spark=spark)
        # TODO: Apply changes
        logger.warning(
            "Laktory does not currently support applying CDC changes. Returned dataframe is CDC source."
        )
        return df
    else:
        return _apply_changes(*args, **kwargs)