Pipeline

from datazone import transform

@transform
def say_hello():
    print("Hello, World!")

@transform(depends=[say_hello])
def say_goodbye():
    print("Goodbye, World!")

Each pipeline should have a unique alias and should be defined in the different files.
A pipeline should have at least one transform function.
You can define dependencies between pipelines using the depends or input_mapping parameter to create a directed acyclic graph (DAG).

Complex Pipeline Example

from datazone import transform

@transform
def prepare():
    print("Preparing data...")

@transform(depends=[prepare])
def build_project():
    print("Building project...")

@transform(depends=[prepare])
def build_report():
    print("Building report...")

@transform(depends=[build_project, build_report])
def notify_email():
    print("Sending email notification...")

Data Flow Management

The @transform decorator enables you to define data transformation functions efficiently. Each function should:

Accept input data as arguments
Process the data
Return the transformed data

Data is handled as PySpark DataFrames both for input and output operations.

from datazone import transform

@transform(input_mapping={'data': Input(Dataset(alias='input_data')})
def clean_data(data):
    return data.filter(data['column'] > 0)

@transform(input_mapping={'clean_data': Input(clean_data)}, materialized=True)
def aggregate_data(clean_data):
    return clean_data.groupBy('column').agg({'column': 'sum'})

In above example,

clean_data function takes input_data as input. You can check the dataset alias in the Datazone UI or use the datazone dataset list command to list all datasets.
After cleaning the data, the clean_data function returns the cleaned data PySpark DataFrame as lazy evaluation.
The aggregate_data function takes the cleaned data as input and aggregates it and returns the aggregated data.
Since the materialized parameter is set to True, the aggregate_data function will be materialized and create a new dataset in Datazone.

Check the Transform section for more information on how to define a transform decorator.

Invoking LLM Models

Transforms can call an LLM through one of your Model Accounts with the Agent class — no provider SDK or API key in your pipeline code. The call is proxied by Datazone and its token usage is attributed to the pipeline.

import pandas as pd

from datazone import transform, Agent, logger

TEXT = """
Datazone connects data sources to a lakehouse and provides tools for transformation,
orchestration and exploration.
"""

@transform(materialized=True, engine="pandas")
def summarize_text():
    agent = Agent(model="CLAUDE_45_SONNET")

    summary = agent.invoke(f"Summarize this in one sentence:\n\n{TEXT}")["content"]
    logger.info(f"Summary: {summary}")

    return pd.DataFrame([{"text": TEXT.strip(), "summary": summary}])

class

Create an agent bound to a model account and model.Parameters:

model (optional) — model to use, e.g. CLAUDE_45_SONNET. Defaults to your organisation’s default model.
model_account (optional) — model account id. Defaults to your organisation’s AI settings account.
response_format (optional) — a Pydantic model or JSON schema dict for structured output.

method

Invoke the model and return a result dictionary.Accepts either a plain string or a message list:

agent.invoke("Say hi")

agent.invoke({
    "messages": [
        {"role": "system", "content": "You are concise."},
        {"role": "user", "content": "Say hi"},
    ]
})

Message roles are system, user and assistant. Returns a dict with:

content — the raw text answer (empty when response_format is used)
structured_response — the parsed object when response_format is used, otherwise None
model — the model that produced the answer

Raises RuntimeError if the invocation fails.

Structured Output

Pass a Pydantic model as response_format to get a validated object back instead of raw text:

from typing import Literal

import pandas as pd
from pydantic import BaseModel, Field

from datazone import Dataset, Input, Output, transform, Agent

reviews = Dataset(alias="product_reviews")

class ProductReview(BaseModel):
    """Analysis of a product review."""
    rating: int | None = Field(description="The rating of the product", ge=1, le=5)
    sentiment: Literal["positive", "negative"] = Field(description="The sentiment of the review")
    key_points: list[str] = Field(description="Key points of the review, 1-3 words each")

@transform(
    input_mapping={"data": Input(reviews)},
    output_mapping={"result": Output(materialized=True)},
    engine="pandas",
)
def analyze_reviews(data):
    agent = Agent(model="CLAUDE_45_SONNET", response_format=ProductReview)

    rows = []
    for comment in data["comment"]:
        analysis = agent.invoke(f"Analyze this review: '{comment}'")["structured_response"]
        rows.append({
            "comment": comment,
            "rating": analysis.rating,
            "sentiment": analysis.sentiment,
            "key_points": ", ".join(analysis.key_points),
        })

    return pd.DataFrame(rows)

Each invoke is one request, so a row-by-row loop over a large dataset is slow and costly. Filter the input first, or batch several records into a single prompt and ask for a list back.

Transform Selection

When executing a pipeline, you can selectively run specific transforms using the transform_selection parameter in the “Run with Config” modal. This allows you to execute only the transforms you need, along with their dependencies if required.

Selection Patterns

Pattern	Description
`some_transform`	Select the transform only
`*some_transform`	Select transform and all ancestors (upstream dependencies)
`some_transform*`	Select transform and all descendants (downstream dependencies)
`some_transform`	Select transform with all ancestors and descendants
`+some_transform`	Select transform and its direct parents
`some_transform+`	Select transform and direct children
`some_transform++`	Select transform and 2 levels of children
`some_transform+++`	Select transform and 3 levels of children

Use transform selection to optimize execution time by running only the necessary parts of your pipeline during development and testing.

Usage Examples

For the pipeline example above with prepare, build_project, build_report, and notify_email:

build_project - Runs only the build_project transform
*build_project - Runs prepare and build_project (transform with all ancestors)
prepare* - Runs prepare, build_project, and build_report (transform with all descendants)
*notify_email* - Runs the entire pipeline (transform with all ancestors and descendants)

Get Started

Reference

Deployment

Tutorial

Complex Pipeline Example

Data Flow Management

Invoking LLM Models

Structured Output

Transform Selection

Selection Patterns

Usage Examples

​Complex Pipeline Example

​Data Flow Management

​Invoking LLM Models

​Structured Output

​Transform Selection

​Selection Patterns

​Usage Examples

Complex Pipeline Example

Data Flow Management

Invoking LLM Models

Structured Output

Transform Selection

Selection Patterns

Usage Examples