By Dherik Barison · May 2025 · 20 min read

What are Custom Functions?

Custom Functions let developers run Python code inside a data pipeline to process documents. With them you can filter out specific records, convert or standardize data formats, add new information to documents, and run advanced calculations. The code executes as part of the pipeline, so your logic sits alongside the platform's managed components rather than in a separate service you have to operate.

Why use Custom Functions?

The platform ships with pre-built components, but Custom Functions cover the cases that need business logic unique to your team. Common reasons to reach for them include:

  • Cleaning data with rules specific to your business
  • Filtering based on nested conditions the default components do not express
  • Transforming formats beyond the default mappings
  • Enriching data through algorithms you write yourself

Getting started: function structure

There are two supported ways to write a Custom Function. You can process a batch of documents at once, or process one document at a time.

Batch processing

def process_batch(documents: list) -> list:
    # Your custom logic here
    return filtered_documents

Individual document processing

def process_document(document):
    # Your custom logic here
    return changed_document

Practical example: Flesch reading ease

The guide walks through calculating a text readability score with the Flesch-Kincaid method. The calculation counts sentences, words, and syllables to produce a readability score for a piece of text. That score has practical uses across content workflows:

  • Analyzing customer support responses for clarity
  • Evaluating marketing materials before they ship
  • Monitoring social media posts for accessibility

Testing capabilities

Datastreamer includes a built-in testing environment so you can validate a function before it runs in production. It provides:

  • A Python code editor
  • A sample JSON document as input
  • Output showing both successfully processed and failed documents
  • Console output for debugging
  • Execution time metrics

Best practices

Performance

  • Use efficient data structures
  • Avoid unnecessary iterations
  • Monitor memory usage
  • Use built-in Python functions where they fit
  • Run local load testing before deployment

Error handling

Wrap processing in try-except blocks so a single failure does not stop the rest of a batch. Use safe access methods like .get() with default values when reading from dictionaries, which prevents crashes when a field is missing.

Documentation

Document your functions clearly. Good documentation gives teammates context, makes maintenance easier, supports collaboration, and guides testing and validation over the life of the function.

Advanced use cases

External data enrichment

A function can call third-party APIs, such as a geocoding service, to append relevant metadata to each document as it passes through the pipeline.

Document normalization

Custom Functions can standardize inconsistent formats from multiple sources into one unified structure, which simplifies everything downstream.

Common challenges and solutions

ChallengeSolution
Large documents Extract the key information you need, and optionally remove large fields to reduce memory usage.
Multiple document schemas Route documents to the right handler based on their structure. For complex cases, consider the Unify Transformer component.

A key recommendation

Back up your Python functions. Deleted components cannot be recovered by Datastreamer, so keep your own copy of the code.

Custom Functions balance the control of custom code with the reliability of a managed pipeline platform. You write the logic that is specific to your product, and Datastreamer runs it inside the same pipeline that handles ingestion, transformation, enrichment, and delivery.