Consumers

The Consumer pattern enables building event-driven architectures. You define your business logic in a function and attach where it should read data from and where it should write data to.

A consumer is composed of:

• A source primitive
• A sink primitive
• Your processing logic

Consumers can easily be created with the consumer decorator method of your flow.

@app.consumer(source=Primitive(...), sink=Primitive(...))
def my_consumer(...):
    ...

Or you can use the @consumer decorator and attach it with the add_consumer method of your flow.

from buildflow import Flow, consumer
@consumer(source=Primitive(...), sink=Primitive(...))
def my_consumer(...):
    ...

app = Flow()
flow.add_consumer(my_consumer)

​

Input and Output

The input of your consumer will be one element that is read from your source primitive. For instance if your source type is a queue (like Kafka, GCP Pub/Sub, or AWS SQS) then your processor will receive one message at a time that is read from the queue.

The element you return from your consumer will be written to your sink primitive. For instance if your sink type is an OLAP datastore (like GCP BigQuery or Snowflake) one row will be written for each element you return from your processor, or if you sink is a queue (like Kafka, GCP Pub/Sub, or AWS SQS) then one message will be published for each element you return from your processor.

For more details on what these types should / can be for your source and sink see our source and sink types documentation.

​

Consumer Options

Consumers can be individually configured for different resource requirements.

The following options are available:

​

Example

Decorator example:

@app.consumer(
    source=...,
    sink=...,
    num_cpus=2,
    num_concurrent_tasks=2,
    log_level=logging.DEBUG,
    num_replicas=10,
)
def my_conusmer():
    ...