Broadway Documentation

Broadway is a library for building concurrent and multi-stage data ingestion and data processing pipelines in Elixir. It leverages the Erlang VM to provide a robust pipeline architecture with the following built-in capabilities:

• Back-pressure: Manages flow control to prevent overwhelming downstream stages.
• Automatic acknowledgements: Handles message ACKs at the end of the pipeline.
• Batching: Groups messages for efficient processing.
• Fault tolerance: Ensures pipeline resilience.
• Graceful shutdown: Allows the pipeline to finish processing before stopping.
• Built-in testing: Provides tools for verifying pipeline logic.
• Custom failure handling: Allows defining specific logic for failed messages.
• Ordering and partitioning: Supports maintaining message order and distributing work.
• Rate-limiting: Controls the speed of message processing.
• Metrics: Provides visibility into pipeline performance.

You can optimize your pipeline by adjusting concurrency and batching settings for each layer (producers, processors, and batchers). Key options include:

• concurrency: Controls the number of processes in a specific layer.
• max_demand: Limits the number of messages requested from the producer.
• batch_size: The number of messages to collect before triggering a batch.
• batch_timeout: The maximum time to wait before processing a partial batch.

It is recommended to include at least a default batcher in SQS pipelines to control the frequency and size of acknowledgments to Amazon SQS, which improves cost and time efficiency.

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        # ... producer config ...
        concurrency: 10,
      ],
      processors: [
        default: [
          concurrency: 100,
          max_demand: 1,
        ]
      ],
      batchers: [
        default: [
          batch_size: 10,
          concurrency: 10,
        ]
      ]
    )
  end
end

Define a Broadway pipeline by implementing start_link/1, handle_message/3, and handle_batch/4. Use BroadwayCloudPubSub.Producer to consume from a specific subscription.

It is highly recommended to include a batcher in Cloud Pub/Sub pipelines to control the frequency and size of message acknowledgements, which improves cost and time efficiency.

defmodule MyBroadway do
  use Broadway

  alias Broadway.Message

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          BroadwayCloudPubSub.Producer,
          subscription: "projects/test-pubsub/subscriptions/test-subscription"
        }
      ],
      processors: [
        default: []
      ],
      batchers: [
        default: [
          batch_size: 10,
          batch_timeout: 2_000
        ]
      ]
    )
  end

  def handle_message(_, %Message{data: data} = message, _) do
    message
    |> Message.update_data(fn data -> String.upcase(data) end)
  end

  def handle_batch(_, messages, _, _) do
    list = messages |> Enum.map(fn e -> e.data end)
    IO.inspect(list, label: "Got batch of finished jobs from processors, sending ACKs to Pub/Sub as a batch.")
    messages
  end
end

To use Broadway with Cloud Pub/Sub, you must have a Google Cloud project, a topic, a subscription, and service account credentials. You can use the gcloud CLI to set this up.

1. Install and authenticate gcloud:

   brew install --cask google-cloud-sdk
   gcloud auth login

2. Create project, topic, and subscription:

   gcloud projects create test-pubsub
   gcloud pubsub topics create test-topic --project test-pubsub
   gcloud pubsub subscriptions create test-subscription --project test-pubsub --topic test-topic

3. Create a service account and assign roles:

   gcloud iam service-accounts create test-account --project test-pubsub
   gcloud projects add-iam-policy-binding test-pubsub \
       --member serviceAccount:test-account@test-pubsub.iam.gserviceaccount.com \
       --role roles/editor

4. Generate credentials file:

   gcloud iam service-accounts keys create credentials.json --iam-account=test-account@test-pubsub.iam.gserviceaccount.com

5. Enable Pub/Sub API:

   gcloud services enable pubsub --project test-pubsub