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.

Asking a Decision Model

For classification, routing and scoring, ask a decision model with agent.decide. It answers typed questions with probabilities instead of text, so there is nothing to parse, and it is fast enough to call once per row.

method

Ask a decision model typed questions about a state.Parameters:

state — what the questions are about: a string, a JSON object, or a list.
questions — a dict of question name → question with type (noul, choice or score), instructions and criteria — see Question Types.

The agent’s model is required and must be an enabled model of kind Decision on the model account.Returns a dict with answers (keyed by question name) and model. Raises RuntimeError if the call fails.

This transform triages a list of support tickets and writes one row per ticket:

import pandas as pd

from datazone import Agent, Output, transform

MODEL_ACCOUNT_ID = "6adf656a81ea9b797f0be8f4"

TICKETS = [
    "Stripe integration has been failing for 3 days and we are losing sales. Please help ASAP!",
    "I was charged twice for my October invoice, can you refund one of them?",
    "How much does the enterprise plan cost for 50 seats?",
    "Hello, I cannot access the website, it says permission denied.",
    "Just wanted to say thanks, the new dashboard looks great.",
]

QUESTIONS = {
    "urgent": {"type": "noul", "instructions": "Does this message need an urgent response?"},
    "team": {
        "type": "choice",
        "instructions": "Which team should handle this request?",
        "criteria": {
            "billing": "Payments, invoices, refunds",
            "technical": "Bugs, access problems and integration failures",
            "sales": "Pricing and purchasing questions",
            "none": "No action needed",
        },
    },
    "frustration": {
        "type": "score",
        "instructions": "How frustrated does the customer appear?",
        "criteria": ["Calm", "Concerned but civil", "Very angry"],
    },
}


@transform(engine="pandas", output_mapping={"decision_triaged_tickets": Output(materialized=True)})
def triage_tickets():
    agent = Agent(model_account=MODEL_ACCOUNT_ID, model="jev-latest")

    rows = []
    for ticket in TICKETS:
        answers = agent.decide(state=ticket, questions=QUESTIONS)["answers"]
        rows.append(
            {
                "ticket": ticket,
                "urgency": answers["urgent"]["noul"],
                "team": answers["team"]["choice"],
                "team_confidence": answers["team"]["confidence"],
                "frustration": answers["frustration"]["score"],
            },
        )

    return pd.DataFrame(rows)

The resulting decision_triaged_tickets dataset:

ticket	urgency	team	team_confidence	frustration
Stripe integration has been failing for 3 days…	0.95	technical	0.89	1.17
I was charged twice for my October invoice…	0.40	billing	1.0	0.34
How much does the enterprise plan cost for 50 seats?	0.17	sales	1.0	0.0
Hello, I cannot access the website…	0.38	technical	1.0	0.42
Just wanted to say thanks, the new dashboard looks great.	0.07	none	1.0	0.0

To triage a dataset instead of a fixed list, take it as an input and loop over its rows. Pass each row as a JSON object — agent.decide(state=json.loads(row.to_json()), questions=QUESTIONS) — so the model sees each column by name. to_json also turns timestamps and numpy numbers into plain JSON values.

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

Asking a Decision Model

Transform Selection

Selection Patterns

Usage Examples

​Complex Pipeline Example

​Data Flow Management

​Invoking LLM Models

​Structured Output

​Asking a Decision Model

​Transform Selection

​Selection Patterns

​Usage Examples

Complex Pipeline Example

Data Flow Management

Invoking LLM Models

Structured Output

Asking a Decision Model

Transform Selection

Selection Patterns

Usage Examples