​

What are User-Defined Functions?

User-Defined Functions (UDFs) in Pixeltable allow you to extend the platform with custom Python code. They bridge the gap between Pixeltable’s built-in operations and your specific data processing needs, enabling you to create reusable components for transformations, analysis, and AI workflows.

Pixeltable UDFs offer several key advantages:

• Reusability: Define a function once and use it across multiple tables and operations
• Type Safety: Strong typing ensures data compatibility throughout your pipelines
• Performance: Batch processing and caching capabilities optimize execution
• Integration: Seamlessly combine custom code with Pixeltable’s query system
• Flexibility: Process any data type including text, images, videos, and embeddings

UDFs can be as simple as a basic transformation or as complex as a multi-stage ML pipeline. Pixeltable offers three types of custom functions to handle different scenarios:

import pixeltable as pxt

# Basic UDF for text transformation
@pxt.udf
def clean_text(text: str) -> str:
    """Clean and normalize text data."""
    return text.lower().strip()

# Use in a computed column
documents = pxt.get_table('my_documents')
documents.add_computed_column(
    clean_content=clean_text(documents.content)
)

​

User-Defined Functions in Pixeltable

This guide covers three types of custom functions in Pixeltable:

1. Basic User-Defined Functions (UDFs)
2. Tables as UDFs
3. User-Defined Aggregates (UDAs)

​

1. Basic User-Defined Functions (UDFs)

​

Overview

UDFs allow you to:

• Write custom Python functions for data processing
• Integrate them into computed columns and queries
• Optimize performance through batching
• Create reusable components for your data workflow

​

Creating Basic UDFs

@pxt.udf
def add_tax(price: float, rate: float = 0.1) -> float:
    return price * (1 + rate)

# Use in computed column
table.add_computed_column(
    price_with_tax=add_tax(table.price)
)

​

UDF Types

# Defined directly in your code
@pxt.udf
def extract_year(date_str: str) -> int:
    return int(date_str.split('-')[0])
    
# Used immediately
table.add_computed_column(
    year=extract_year(table.date)
)

​

Supported Types

@pxt.udf
def process_data(
    text: str,           # String data
    count: int,          # Integer numbers
    score: float,        # Floating point
    active: bool,        # Boolean
    items: list[str],    # Generic lists
    meta: dict[str,any]  # Dictionaries
) -> str:
    return "Processed"

@pxt.udf
def process_media(
    img: PIL.Image.Image,    # Images
    embeddings: pxt.Array,   # Numerical arrays
    config: pxt.Json,        # JSON data
    doc: pxt.Document        # Documents
) -> str:
    return "Processed"

​

Performance Optimization

@pxt.udf(batch_size=16)
def embed_texts(
    texts: Batch[str]
) -> Batch[pxt.Array]:
    # Process multiple texts at once
    return model.encode(texts)

@pxt.udf
def expensive_operation(text: str) -> str:
    # Cache model instance
    if not hasattr(expensive_operation, 'model'):
        expensive_operation.model = load_model()
    return expensive_operation.model(text)

​

Best Practices for Basic UDFs

@pxt.udf
def validate_score(score: float) -> float:
    if not 0 <= score <= 100:
        raise ValueError("Score must be between 0 and 100")
    return score

@pxt.udf(batch_size=32)
def process_chunk(items: Batch[str]) -> Batch[str]:
    if not hasattr(process_chunk, 'model'):
        process_chunk.model = load_expensive_model()
    return process_chunk.model.process_batch(items)

@pxt.udf
def normalize_text(
    text: str,
    lowercase: bool = True,
    remove_punctuation: bool = True
) -> str:
    """Normalize text by optionally lowercasing and removing punctuation."""
    if lowercase:
        text = text.lower()
    if remove_punctuation:
        text = text.translate(str.maketrans("", "", string.punctuation))
    return text

​

2. Tables as UDFs

​

Overview

Tables as UDFs allow you to:

• Convert entire tables into reusable functions
• Create modular and complex data processing workflows
• Encapsulate multi-step operations
• Share workflows between different tables and applications

​

Creating Table UDFs

# Create a table with your workflow
finance_agent = pxt.create_table('directory.financial_analyst', 
                                {'prompt': pxt.String})
# Add computed columns for processing
finance_agent.add_computed_column(/* ... */)

# Convert table to UDF by specifying return column
finance_agent_udf = pxt.udf(finance_agent, 
                           return_value=finance_agent.answer)

# Use like any other UDF
result_table.add_computed_column(
    result=finance_agent_udf(result_table.prompt)
)

​

Flow Diagram

​

Key Benefits of Table UDFs

​

Advanced Techniques

# Create a chain of specialized agents
research_agent = pxt.udf(research_table, return_value=research_table.findings)
analysis_agent = pxt.udf(analysis_table, return_value=analysis_table.insights)
report_agent = pxt.udf(report_table, return_value=report_table.document)

# Use them in sequence
workflow.add_computed_column(research=research_agent(workflow.query))
workflow.add_computed_column(analysis=analysis_agent(workflow.research))
workflow.add_computed_column(report=report_agent(workflow.analysis))

# Define specialized agents for different tasks
stock_agent = pxt.udf(stock_table, return_value=stock_table.analysis)
news_agent = pxt.udf(news_table, return_value=news_table.summary)
sentiment_agent = pxt.udf(sentiment_table, return_value=sentiment_table.score)

# Process in parallel
portfolio.add_computed_column(stock_data=stock_agent(portfolio.ticker))
portfolio.add_computed_column(news_data=news_agent(portfolio.ticker))
portfolio.add_computed_column(sentiment=sentiment_agent(portfolio.ticker))

# Combine results
portfolio.add_computed_column(report=combine_insights(
    portfolio.stock_data, 
    portfolio.news_data, 
    portfolio.sentiment
))

​

3. User-Defined Aggregates (UDAs)

​

Overview

UDAs enable you to:

• Create custom aggregation functions
• Process multiple rows into a single result
• Use them in group_by operations
• Build reusable aggregation logic

​

Creating UDAs

@pxt.uda
class sum_of_squares(pxt.Aggregator):
    def __init__(self):
        self.cur_sum = 0
        
    def update(self, val: int) -> None:
        self.cur_sum += val * val
        
    def value(self) -> int:
        return self.cur_sum

​

UDA Components

1. Initialization (__init__)

   • Sets up initial state
   • Defines parameters
   • Called once at start

2. Update Method (update)

   • Processes each input row
   • Updates internal state
   • Must handle all value types

3. Value Method (value)

   • Returns final result
   • Called after all updates
   • Performs final calculations

​

Using UDAs

# Basic usage
table.select(sum_of_squares(table.value)).collect()

# With grouping
table.group_by(table.category).select(
    table.category,
    sum_of_squares(table.value)
).collect()

​

Best Practices for UDAs

• Manage state carefully
• Handle edge cases and errors
• Optimize for performance
• Use appropriate type hints
• Document expected behavior

​

Additional Resources