> ## Documentation Index
> Fetch the complete documentation index at: https://docs.datazone.co/llms.txt
> Use this file to discover all available pages before exploring further.

# Pipeline

> Define data processing steps using transforms and dependencies to build workflows

```python theme={null}
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

```python theme={null}
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...")
```

```mermaid theme={null}
graph LR;
    prepare-->build_project;
    prepare-->build_report;
    build_project-->notify_email;
    build_report-->notify_email;
```

### 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

<Info>Data is handled as PySpark DataFrames both for input and output operations.</Info>

```python theme={null}
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,

1. `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.
2. After cleaning the data, the `clean_data` function returns the cleaned data PySpark DataFrame as lazy evaluation.
3. The `aggregate_data` function takes the cleaned data as input and aggregates it and returns the aggregated data.
4. Since the `materialized` parameter is set to `True`, the `aggregate_data` function will be materialized and create a new dataset in Datazone.

<Note>Check the [Transform](/reference/development/transform) section for more information on how to define a transform decorator.</Note>

### Invoking LLM Models

Transforms can call an LLM through one of your [Model Accounts](/reference/development/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.

```python theme={null}
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}])
```

<ParamField path="Agent" type="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.
</ParamField>

<ParamField path="agent.invoke" type="method">
  Invoke the model and return a result dictionary.

  Accepts either a plain string or a message list:

  ```python theme={null}
  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.
</ParamField>

#### Structured Output

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

```python theme={null}
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)
```

<Warning>
  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.
</Warning>

### 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 |

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

#### 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)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.