Pipeline

laktory.models.Pipeline

Bases: BaseModel, PulumiResource, TerraformResource, PipelineChild

Pipeline model to manage a full-fledged data pipeline including reading from data sources, applying data transformations through Spark and outputting to data sinks.

A pipeline is composed of collections of nodes, each one defining its own source, transformations and optional sink. A node may be the source of another node.

A pipeline may be run manually by using python or the CLI, but it may also be deployed and scheduled using one of the supported orchestrators, such as a Databricks Delta Live Tables or job.

ATTRIBUTE	DESCRIPTION
databricks_job	Defines the Databricks Job specifications when DATABRICKS_JOB is selected as the orchestrator. Requires to add the supporting notebook to the stack. TYPE: Union[DatabricksJobOrchestrator, None]
databricks_dlt	Defines the Databricks DLT specifications when DATABRICKS_DLT is selected as the orchestrator. Requires to add the supporting notebook to the stack. TYPE: Union[DatabricksDLTOrchestrator, None]
dependencies	List of dependencies required to run the pipeline. If Laktory is not provided, it's current version is added to the list. TYPE: list[str]
name	Name of the pipeline TYPE: str
nodes	List of pipeline nodes. Each node defines a data source, a series of transformations and optionally a sink. TYPE: list[PipelineNode]
orchestrator	Orchestrator used for scheduling and executing the pipeline. The selected option defines which resources are to be deployed. Supported options are: DATABRICKS_DLT: When orchestrated through Databricks DLT, each pipeline node creates a DLT table (or view, if no sink is defined). Behind the scenes, PipelineNodeDataSource leverages native dlt read and read_stream functions to defined the interdependencies between the tables as in a standard DLT pipeline. DATABRICKS_JOB: When deployed through a Databricks Job, a task is created for each pipeline node and all the required dependencies are set automatically. If a given task (or pipeline node) uses a PipelineNodeDataSource as the source, the data will be read from the upstream node sink. TYPE: Literal['DATABRICKS_DLT', 'DATABRICKS_JOB', None]
udfs	List of user defined functions provided to the transformer. TYPE: list[PipelineUDF]
root_path	Location of the pipeline node root used to store logs, metrics and checkpoints. TYPE: str

Examples:

This first example shows how to configure a simple pipeline with 2 nodes. Upon execution, raw data will be read from a CSV file and two DataFrames (bronze and silver) will be created and saved as parquet files. Notice how the first node is used as a data source for the second node.

import io

from laktory import models

pipeline_yaml = '''
    name: pl-stock-prices
    nodes:
    - name: brz_stock_prices
      layer: BRONZE
      source:
        format: CSV
        path: ./raw/brz_stock_prices.csv
      sinks:
      - format: PARQUET
        mode: OVERWRITE
        path: ./dataframes/brz_stock_prices

    - name: slv_stock_prices
      layer: SILVER
      source:
        node_name: brz_stock_prices
      sinks:
      - format: PARQUET
        mode: OVERWRITE
        path: ./dataframes/slv_stock_prices
      transformer:
        nodes:
        - with_column:
            name: created_at
            type: timestamp
            expr: data.created_at
        - with_column:
            name: symbol
            expr: data.symbol
        - with_column:
            name: close
            type: double
            expr: data.close
        - func_name: drop
          func_args:
          - value: data
          - value: producer
          - value: name
          - value: description
'''

pl = models.Pipeline.model_validate_yaml(io.StringIO(pipeline_yaml))

# Execute pipeline
# pl.execute()

The next example defines a 3 nodes pipeline (1 bronze and 2 silvers) orchestrated with a Databricks Job. Notice how nodes are used as data sources not only for other nodes, but also for the other keyword argument of the smart join function (slv_stock_prices). Because we are using the DATABRICKS_JOB orchestrator, the job configuration must be declared. The tasks will be automatically created by the Pipeline model. Each task will execute a single node using the notebook referenced in databricks_job.notebook_path the content of this notebook should be similar to laktory.resources.notebooks.job_laktory.pl

import io

from laktory import models

pipeline_yaml = '''
    name: pl-stock-prices
    orchestrator: DATABRICKS_JOB
    databricks_job:
      name: job-pl-stock-prices
      notebook_path: /Workspace/.laktory/jobs/job_laktory_pl.py
      clusters:
        - name: node-cluster
          spark_version: 14.0.x-scala2.12
          node_type_id: Standard_DS3_v2
    dependencies:
        - laktory==0.3.0
        - yfinance
    nodes:
    - name: brz_stock_prices
      layer: BRONZE
      source:
        path: /Volumes/dev/sources/landing/events/yahoo-finance/stock_price/
      sinks:
      -   path: /Volumes/dev/sources/landing/tables/dev_stock_prices/
          mode: OVERWRITE

    - name: slv_stock_prices
      layer: SILVER
      source:
        node_name: brz_stock_prices
      sinks:
      -   path: /Volumes/dev/sources/landing/tables/slv_stock_prices/
          mode: OVERWRITE
      transformer:
        nodes:
        - with_column:
            name: created_at
            type: timestamp
            expr: data.created_at
        - with_column:
            name: symbol
            expr: data.symbol
        - with_column:
            name: close
            type: double
            expr: data.close
        - func_name: drop
          func_args:
          - value: data
          - value: producer
          - value: name
          - value: description
        - func_name: laktory.smart_join
          func_kwargs:
            'on':
              - symbol
            other:
              node_name: slv_stock_meta

    - name: slv_stock_meta
      layer: SILVER
      source:
        path: /Volumes/dev/sources/landing/events/yahoo-finance/stock_meta/
      sinks:
      - path: /Volumes/dev/sources/landing/tables/slv_stock_meta/
        mode: OVERWRITE

'''
pl = models.Pipeline.model_validate_yaml(io.StringIO(pipeline_yaml))

Finally, we re-implement the previous pipeline, but with a few key differences:

• Orchestrator is DATABRICKS_DLT instead of a DATABRICKS_JOB
• Sinks are Unity Catalog tables instead of storage locations
• Data is read as a stream in most nodes
• slv_stock_meta is simply a DLT view since it does not have an associated sink.

We also need to provide some basic configuration for the DLT pipeline.

import io

from laktory import models

pipeline_yaml = '''
    name: pl-stock-prices
    orchestrator: DATABRICKS_DLT
    databricks_dlt:
        catalog: dev
        target: sandbox
        access_controls:
        - group_name: users
          permission_level: CAN_VIEW

    nodes:
    - name: brz_stock_prices
      layer: BRONZE
      source:
        path: /Volumes/dev/sources/landing/events/yahoo-finance/stock_price/
        as_stream: true
      sinks:
      - table_name: brz_stock_prices

    - name: slv_stock_prices
      layer: SILVER
      source:
        node_name: brz_stock_prices
        as_stream: true
      sinks:
      - table_name: slv_stock_prices
      transformer:
        nodes:
        - with_column:
            name: created_at
            type: timestamp
            expr: data.created_at
        - with_column:
            name: symbol
            expr: data.symbol
        - with_column:
            name: close
            type: double
            expr: data.close
        - func_name: drop
          func_args:
          - value: data
          - value: producer
          - value: name
          - value: description
        - func_name: laktory.smart_join
          func_kwargs:
            'on':
              - symbol
            other:
              node_name: slv_stock_meta

    - name: slv_stock_meta
      layer: SILVER
      source:
        path: /Volumes/dev/sources/landing/events/yahoo-finance/stock_meta/

'''
pl = models.Pipeline.model_validate_yaml(io.StringIO(pipeline_yaml))

References

• Databricks Job
• Databricks DLT

METHOD	DESCRIPTION
push_df_backend	Need to push dataframe_backend which is required to differentiate between spark and polars transformer
execute	Execute the pipeline (read sources and write sinks) by sequentially
dag_figure	[UNDER DEVELOPMENT] Generate a figure representation of the pipeline

Attributes

is_orchestrator_dlt property

is_orchestrator_dlt

If True, pipeline orchestrator is DLT

nodes_dict property

nodes_dict

Nodes dictionary whose keys are the node names.

RETURNS	DESCRIPTION
dict[str, PipelineNode]	Nodes

dag property

dag

Networkx Directed Acyclic Graph representation of the pipeline. Useful to identify interdependencies between nodes.

RETURNS	DESCRIPTION
DiGraph	Directed Acyclic Graph

sorted_nodes property

sorted_nodes

Topologically sorted nodes.

RETURNS	DESCRIPTION
list[PipelineNode]	List of Topologically sorted nodes.

resource_type_id property

resource_type_id

pl

additional_core_resources property

additional_core_resources

if orchestrator is DLT:

• DLT Pipeline

if orchestrator is DATABRICKS_JOB:

• Databricks Job

Functions

push_df_backend classmethod

push_df_backend(data)

Need to push dataframe_backend which is required to differentiate between spark and polars transformer

Source code in laktory/models/pipeline/pipeline.py

@model_validator(mode="before")
@classmethod
def push_df_backend(cls, data: Any) -> Any:
    """Need to push dataframe_backend which is required to differentiate between spark and polars transformer"""
    df_backend = data.get("dataframe_backend", None)
    if df_backend:
        if "nodes" in data.keys():
            for n in data["nodes"]:
                if isinstance(n, dict):
                    n["dataframe_backend"] = n.get("dataframe_backend", df_backend)
                else:
                    n.dataframe_backend = n.dataframe_backend or df_backend
    return data

execute

execute(spark=None, udfs=None, write_sinks=True, full_refresh=False)

Execute the pipeline (read sources and write sinks) by sequentially executing each node. The selected orchestrator might impact how data sources or sinks are processed.

PARAMETER	DESCRIPTION
spark	Spark Session DEFAULT: None
udfs	List of user-defined functions used in transformation chains. DEFAULT: None
write_sinks	If False writing of node sinks will be skipped DEFAULT: True
full_refresh	If True all nodes will be completely re-processed by deleting existing data and checkpoints before processing. TYPE: bool DEFAULT: False

Source code in laktory/models/pipeline/pipeline.py

def execute(
    self, spark=None, udfs=None, write_sinks=True, full_refresh: bool = False
) -> None:
    """
    Execute the pipeline (read sources and write sinks) by sequentially
    executing each node. The selected orchestrator might impact how
    data sources or sinks are processed.

    Parameters
    ----------
    spark:
        Spark Session
    udfs:
        List of user-defined functions used in transformation chains.
    write_sinks:
        If `False` writing of node sinks will be skipped
    full_refresh:
        If `True` all nodes will be completely re-processed by deleting
        existing data and checkpoints before processing.
    """
    logger.info("Executing Pipeline")

    for inode, node in enumerate(self.sorted_nodes):
        node.execute(
            spark=spark,
            udfs=udfs,
            write_sinks=write_sinks,
            full_refresh=full_refresh,
        )

dag_figure

dag_figure()

[UNDER DEVELOPMENT] Generate a figure representation of the pipeline DAG.

RETURNS	DESCRIPTION
Figure	Plotly figure representation of the pipeline.

Source code in laktory/models/pipeline/pipeline.py

def dag_figure(self) -> Figure:
    """
    [UNDER DEVELOPMENT] Generate a figure representation of the pipeline
    DAG.

    Returns
    -------
    :
        Plotly figure representation of the pipeline.
    """
    import networkx as nx
    import plotly.graph_objs as go

    dag = self.dag
    pos = nx.spring_layout(dag)

    # ------------------------------------------------------------------- #
    # Edges                                                               #
    # ------------------------------------------------------------------- #

    edge_traces = []
    for e in dag.edges:
        edge_traces += [
            go.Scatter(
                x=[pos[e[0]][0], pos[e[1]][0]],
                y=[pos[e[0]][1], pos[e[1]][1]],
                line={
                    "color": "#a006b1",
                },
                marker=dict(
                    symbol="arrow-bar-up",
                    angleref="previous",
                    size=30,
                ),
                mode="lines+markers",
                hoverinfo="none",
                showlegend=False,
            )
        ]

    # ------------------------------------------------------------------- #
    # Nodes                                                               #
    # ------------------------------------------------------------------- #

    nodes_trace = go.Scatter(
        x=[_p[0] for _p in pos.values()],
        y=[_p[1] for _p in pos.values()],
        text=list(dag.nodes),
        name="pl nodes",
        mode="markers+text",
        marker=dict(
            size=50,
            color="#06d49e",
            line=dict(
                width=2,
                color="#dff2ed",
            ),
        ),
        textfont=dict(
            color="#a006b1",
        ),
    )

    return go.Figure(data=[nodes_trace] + edge_traces)

---

laktory.models.pipeline.pipeline.PipelineUDF

Bases: BaseModel

Pipeline User Define Function

ATTRIBUTE	DESCRIPTION
module_name	Name of the module from which the function needs to be imported. TYPE: str
function_name	Name of the function. TYPE: str
module_path	Workspace filepath of the module, if not in the same directory as the pipeline notebook TYPE: str