Broadway is a concurrent, multi-stage tool for building data ingestion and data processing pipelines.

It allows developers to consume data efficiently from different sources, such as Amazon SQS, RabbitMQ and others.

Built-in features

• Back-pressure - by relying on GenStage, we only get the amount of events necessary from upstream sources, never flooding the pipeline.

• Automatic acknowledgements - Broadway automatically acknowledges messages at the end of the pipeline or in case of errors.

• Batching - Broadway provides built-in batching, allowing you to group messages either by size and/or by time. This is important in systems such as Amazon SQS, where batching is the most efficient way to consume messages, both in terms of time and cost.

• Fault tolerance with minimal data loss - Broadway pipelines are carefully designed to minimize data loss. Producers are isolated from the rest of the pipeline and automatically resubscribed to in case of failures. On the other hand, user callbacks are stateless, allowing us to handle any errors locally. Finally, in face of any unforeseen bug, we restart only downstream components, avoiding data loss.

• Graceful shutdown - Broadway integrates with the VM to provide graceful shutdown. By starting Broadway as part of your supervision tree, it will guarantee all events are flushed once the VM shuts down.

• Built-in testing - Broadway ships with a built-in test API, making it easy to push test messages through the pipeline and making sure the event was properly processed.

• Partitioning - Broadway allows developers to batch messages based on dynamic partitions. For example, if your pipeline needs to build batches based on the user_id, email address, etc, it can be done by calling Broadway.Message.put_batch_key/2.

• Rate-limiting (TODO)

• Statistics/Metrics (TODO)

• Back-off (TODO)

The Broadway Behaviour

In order to use Broadway, you need to:

1. Define your pipeline configuration
2. Define a module implementing the Broadway behaviour

Example

Like any other process-based behaviour, you can start your Broadway process by defining a module that invokes use Broadway and has a start_link function:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(MyBroadway,
      name: MyBroadwayExample,
      producers: [
        default: [
          module: {Counter, []},
          stages: 1
        ]
      ],
      processors: [
        default: [stages: 2]
      ]
    )
  end

  ...callbacks...
end

Then add your Broadway pipeline to your supervision tree (usually in lib/my_app/application.ex):

children = [
  {MyBroadway, []}
]

Supervisor.start_link(children, strategy: :one_for_one)

The configuration above defines a pipeline with:

• 1 producer
• 2 processors

Here is how this pipeline would be represented:

                     [producer_1]
                         / \
                        /   \
                       /     \
                      /       \
             [processor_1] [processor_2]   <- process each message

After the pipeline is defined, you need to implement handle_message/3, which will be invoked by processors for each message.

handle_message/3 receives every message as a Broadway.Message struct and it must return an updated message.

Batching

Depending on the scenario, you may want to group processed messages as batches before publishing your data. This is common and especially important when working with services like AWS S3 and SQS that provide specific API for sending and retrieving batches. This can drastically increase throughput and consequently improve the overall performance of your pipeline.

In order to create batches you need to define the batchers option in the configuration:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(MyBroadway,
      name: MyBroadwayExample,
      producers: [
        default: [
          module: {Counter, []},
          stages: 1
        ]
      ],
      processors: [
        default: [stages: 2]
      ],
      batchers: [
        sqs: [stages: 2, batch_size: 10],
        s3: [stages: 1, batch_size: 10]
      ]
    )
  end

  ...callbacks...
end

The configuration above defines a pipeline with:

• 1 producer
• 2 processors
• 1 batcher named :sqs with 2 consumers
• 1 batcher named :s3 with 1 consumer

Here is how this pipeline would be represented:

                     [producer_1]
                         / \
                        /   \
                       /     \
                      /       \
             [processor_1] [processor_2]   <- process each message
                      /\     /\
                     /  \   /  \
                    /    \ /    \
                   /      x      \
                  /      / \      \
                 /      /   \      \
                /      /     \      \
           [batcher_sqs]    [batcher_s3]
                /\                  \
               /  \                  \
              /    \                  \
             /      \                  \
 [consumer_sqs_1] [consumer_sqs_2]  [consumer_s3_1] <- process each batch

Additionally, you'll need to define the handle_batch/4 callback, which will be invoked by consumers for each batch. You can then invoke Broadway.Message.put_batcher/2 inside handle_message/3 to control to which batcher the message should go to.

The batcher will receive the processed messages and create batches specified by the batch_size and batch_timeout configuration. The goal is to create a batch with at most batch_size entries within batch_timeout milliseconds. Each message goes into a particular batch, controlled by calling Broadway.Message.put_batch_key/2 in handle_message/3. Once a batch is created, it is sent to a separate process that will call handle_batch/4, passing the batcher, the batch itself (i.e. a list of messages), a Broadway.BatchInfo struct and the Broadway context.

For example, imagine your producer generates integers as data. You want to route the odd integers to SQS and the even ones to S3. Your pipeline would look like this:

defmodule MyBroadway do
  use Broadway
  import Integer

  alias Broadway.Message

  ...start_link...

  @impl true
  def handle_message(_, %Message{data: data} = message, _) when is_odd(data) do
    message
    |> Message.update_data(&process_data/1)
    |> Message.put_batcher(:sqs)
  end

  def handle_message(_, %Message{data: data} = message, _) when is_even(data) do
    message
    |> Message.update_data(&process_data/1)
    |> Message.put_batcher(:s3)
  end

  defp process_data(data) do
    # Do some calculations, generate a JSON representation, etc.
  end

  @impl true
  def handle_batch(:sqs, messages, _batch_info, _context) do
    # Send batch of messages to SQS
  end

  def handle_batch(:s3, messages, _batch_info, _context) do
    # Send batch of messages to S3
  end
end

See the callbacks documentation for more information on the arguments given to each callback and their expected return types.

Now you are ready to get started. See the start_link/2 function for a complete reference on the arguments and options allowed.

Also makes sure to check out GUIDES in the documentation sidebar for more examples, how tos and more.

Acknowledgements and failures

At the end of the pipeline, messages are automatically acknowledged.

If there are no batchers, the acknowledgement will be done by processors. The number of messages acknowledged, assuming the pipeline is running at full scale, will be max_demand - min_demand. Since the default values are 10 and 5 respectively, we will be acknowledging in groups of 5.

If there are batchers, the acknowledgement is done by the batchers, using the batch_size.

In case of failures, Broadway does its best to keep the failures contained and avoid losing messages. The failed message or batch is acknowledged as failed immediately. For every failure, a log report is also emitted.

Note however, that Broadway does not provide any sort of retries out of the box. This is left completely as a responsibility of the producer. For instance, if you are using Amazon SQS, the default behaviour is to retry unacknowledged messages after a user-defined timeout. If you don't want unacknowledged messages to be retried, is your responsibility to configure a dead-letter queue as target for those messages.

Testing

Testing Broadway pipelines can be done with test_messages/2. With test_messages/2, you can push some sample data into the pipeline and receive a process message when the pipeline acknowledges the data you have pushed has been processed. This is very useful as a synchronization mechanism. Because many pipelines end-up working with side-effects, you can use the test message acknowledgment to guarantee the message has been processed and therefore side-effects should be visible.

For example, if you have a pipeline named MyApp.Broadway that writes to the database on every message, you could test it as:

# Push 3 messages with the data field set to 1, 2, and 3 respectively
ref = Broadway.test_messages(MyApp.Broadway, [1, 2, 3])

# Assert that the messages have been consumed
assert_receive {:ack, ^ref, [_, _, _] = _successful, failed}

# Now assert the database side-effects
...

Keep in mind that multiple acknowledgement messages may be sent. For example, if the batcher in the example above has size of 2, then two batches would be created and therefore two ack messages would be sent. Similarly, if any of the messages fail when processed, an acknowledgement of their failure may be sent early on. On the positive side, if you always push just a single test message, then there is always one acknowledgment.