Python SDK for Numaflow

This is the Python SDK for Numaflow.

This SDK provides the interface for writing different functionalities of Numaflow like UDFs, UDSinks, UDSources and SideInput in Python.

Installation

Install the package using pip.

pip install pynumaflow

Build locally

This project uses Poetry for dependency management and packaging. To build the package locally, run the following command from the root of the project.

make setup

To run unit tests:

make test

To format code style using black and ruff:

make lint

Setup pre-commit hooks:

pre-commit install

Implementing different functionalities

• Implement User Defined Sources
• Implement User Defined Source Transformers
• Implement User Defined Functions
  • Map
  • Reduce
  • Map Stream
  • Batch Map
• Implement User Defined Sinks
• Implement User Defined SideInputs

Server Types

There are different types of gRPC server mechanisms which can be used to serve the UDFs, UDSinks and UDSource. These have different functionalities and are used for different use cases.

Currently we support the following server types:

• Sync Server
• Asyncronous Server
• MultiProcessing Server

Not all of the above are supported for all UDFs, UDSource and UDSinks.

For each of the UDFs, UDSource and UDSinks, there are seperate classes for each of the server types. This helps in keeping the interface simple and easy to use, and the user can start the specific server type based on the use case.

SyncServer

Syncronous Server is the simplest server type. It is a multithreaded threaded server which can be used for simple UDFs and UDSinks. Here the server will invoke the handler function for each message. The messaging is synchronous and the server will wait for the handler to return before processing the next message.

grpc_server = MapServer(handler)

AsyncServer

Asyncronous Server is a multi threaded server which can be used for UDFs which are asyncronous. Here we utilize the asyncronous capabilities of Python to process multiple messages in parallel. The server will invoke the handler function for each message. The messaging is asyncronous and the server will not wait for the handler to return before processing the next message. Thus this server type is useful for UDFs which are asyncronous. The handler function for such a server should be an async function.

grpc_server = MapAsyncServer(handler)

MultiProcessServer

MultiProcess Server is a multi process server which can be used for UDFs which are CPU intensive. Here we utilize the multi process capabilities of Python to process multiple messages in parallel by forking multiple servers in different processes. The server will invoke the handler function for each message. Individually at the server level the messaging is synchronous and the server will wait for the handler to return before processing the next message. But since we have multiple servers running in parallel, the overall messaging also executes in parallel.

This could be an alternative to creating multiple replicas of the same UDF container as here we are using the multi processing capabilities of the system to process multiple messages in parallel but within the same container.

Thus this server type is useful for UDFs which are CPU intensive.

grpc_server = MapMultiProcServer(mapper_instance=handler, server_count=2)

Currently Supported Server Types for each functionality

These are the class names for the server types supported by each of the functionalities.

• UDFs
  • Map
    • MapServer
    • MapAsyncServer
    • MapMultiProcServer
  • Reduce
    • ReduceAsyncServer
  • MapStream
    • MapStreamAsyncServer
  • BatchMap
    • BatchMapAsyncServer
  • Source Transform
    • SourceTransformServer
    • SourceTransformMultiProcServer
• UDSource
  • SourceServer
  • SourceAsyncServer
• UDSink
  • SinkServer
  • SinkAsyncServer
• SideInput
  • SideInputServer

Handler Function and Classes

All the server types take a instance of a handler class or a handler function as an argument. The handler function or class is the function or class which implements the functionality of the UDF, UDSource or UDSink. For ease of use the user can pass either of the two to the server and the server will handle the rest.

The handler for each of the servers has a specific signature which is defined by the server type and the implentation of the handlers should follow the same signature.

For using the class based handlers the user can inherit from the base handler class for each of the functionalities and implement the handler function. The base handler class for each of the functionalities has the same signature as the handler function for the respective server type. The list of base handler classes for each of the functionalities is given below -

• UDFs
  • Map
    • Mapper
  • Reduce
    • Reducer
  • MapStream
    • MapStreamer
  • Source Transform
    • SourceTransformer
  • Batch Map
    • BatchMapper
• UDSource
  • Sourcer
• UDSink
  • Sinker
• SideInput
  • SideInput

More details about the signature of the handler function for each of the server types is given in the documentation of the respective server type.