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, sink):
    dlt.create_streaming_table(name=sink.table_name)
    df = dlt.apply_changes(
        source=node.source.table_name, **sink.dlt_apply_changes_kwargs
    )
    return df


node = models.PipelineNode(
    name="slv_stock_prices",
    source={
        "table_name": "brz_stock_prices",
    },
    sinks=[
        {
            "table_name": "brz_stock_prices",
            "mode": "MERGE",
            "merge_cdc_options": {
                "primary_keys": ["asset_symbol"],
                "order_by": "change_id",
                "scd_type": 2,
            },
        }
    ],
)

define_table(node, node.primary_sink)

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, sink):
        dlt.create_streaming_table(name=sink.table_name)
        df = dlt.apply_changes(
            source=node.source.table_name, **sink.dlt_apply_changes_kwargs
        )
        return df


    node = models.PipelineNode(
        name="slv_stock_prices",
        source={
            "table_name": "brz_stock_prices",
        },
        sinks=[
            {
                "table_name": "brz_stock_prices",
                "mode": "MERGE",
                "merge_cdc_options": {
                    "primary_keys": ["asset_symbol"],
                    "order_by": "change_id",
                    "scd_type": 2,
                },
            }
        ],
    )

    define_table(node, node.primary_sink)
    ```
    """
    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)