Transform

​

Basic Transform Example

from datazone import transform

@transform(name="say_hello", description="Prints 'Hello, World!'")
def say_hello():
    print("Hello, World!")

​

Transform Engines

The transform decorator supports two computation engines: PySpark and Pandas. You can specify the engine using the engine parameter.

from datazone import transform

@transform(
    input_mapping={"user_data": Input(Dataset(id="66280b3f49ae018b4c0c904a"))},
    engine="pandas"
)
def filter_by_age(user_data):
    adult_users = user_data[user_data.age >= 30]
    return adult_users

​

Engine Characteristics

• PySpark (default):
  • Distributed processing capabilities
  • Better for large-scale data processing
  • Supports all materialization features
• Pandas:
  • Better for smaller datasets
  • More intuitive Python-native syntax
  • Ideal for local development and testing

​

Materialization

Materialization is the process of storing the output of a transform function for reuse. This can be useful when a transform function is computationally expensive and its output is used multiple times in the pipeline.

from datazone import transform

@transform(name="fetch_data", materialized=True)
def fetch_data(context):
    data = [("Alice", 9), ("Bob", 6), ("Charlie", 3), ("Maria", 7)]
    columns = ["Name", "Score"]

    spark = context.resources["pyspark"].spark
    return spark.createDataFrame(data, columns)

After running the pipeline, the output of the fetch_data function will be stored in Datazone as a dataset. You can check the dataset alias in the Datazone UI or use the datazone dataset list command to list all datasets.

​

Input Mapping

You can define input mappings to specify the data sources and dependencies for your transform functions. Input mappings enable you to:

• Chain multiple transform functions
• Create directed acyclic graphs (DAGs)
• Connect to different data sources
• Apply data transformations sequentially

Here are the common input mapping patterns:

from datazone import transform

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


@transform(
    input_mapping={
        "clean_data": Input(clean_data),
        "another_data": Input(Dataset(alias="another_data")),
    }
)
def aggregate_data(clean_data, another_data):
    return (
        clean_data.join(another_data, "column")
        .groupBy("column")
        .agg({"column": "sum"})
        .select("column", "sum(column)")
    )

• Input mappings should be defined as a dictionary where the key is the input parameter name and the value is an instance of the Input class.
• The Input class accepts a Dataset or another transform function as an argument.

​

Output Mapping

Output mapping defines how the output of a transform function should be stored or mapped. You can specify the output mapping using the output_mapping parameter.

from datazone import transform, Output, Dataset

@transform(output_mapping={dataset: Output(dataset=Dataset(alias="clean_data"))})
def clean_data(data):
    return data.filter(data["column"] > 0)

• Output mapping should be defined as a dictionary where the key is the output parameter name and the value is an instance of the Output class.

​

Multiple Outputs

@transform(
    input_mapping={"orders": Input(Dataset(alias="orders"))},
    output_mapping={
        "daily_sales": Output(materialized=True),
        "monthly_sales": Output(materialized=True),
        "yearly_sales": Output(
            partition_by=["year"],
            materialized=True
        )
    }
)
def process_sales(orders):
    daily = orders.groupBy("date").agg(...)
    monthly = orders.groupBy("year", "month").agg(...)
    yearly = orders.groupBy("year").agg(...)

    return daily, monthly, yearly

​

Partitioning

Partitioning helps organize and optimize your datasets in our data platform. When you create a transform function, you can specify partition columns using the partition_by parameter.

@transform(
    partition_by=["date", "country"],
    materialized=True
)
def sales_by_country(data):
    return data.filter(...)

​

Why Partition?

• Improve query performance when filtering by partition columns
• Efficiently manage large datasets
• Enable data retention policies by date partitions

​

Common Partition Strategies

# Daily partitioning for time-series data
@transform(partition_by=["date"])

# Geographic partitioning
@transform(partition_by=["country", "region"])

# Multiple partition columns
@transform(partition_by=["date", "customer_type"])

​

Best Practices

• Use date partitioning for time-series data
• Avoid partitioning by columns with high cardinality
• Consider your query patterns when choosing partition columns