View Source Batch Worker

🌟 This worker is available through Oban.Pro

A Batch worker links the execution of many jobs as a group and runs optional callbacks after jobs are processed. This allows your application to coordinate the execution of tens, hundreds or thousands of jobs in parallel. It is built as an ergonomic abstraction over the top of the standard Oban.Worker.

Usage

While Batch workers are built within your application, they all rely on the BatchManager plugin from Oban.Pro. The BatchManager is responsible for tracking the execution of jobs within a batch and reliably enqueuing callback jobs.

Start by adding BatchManager to your list of Oban plugins in config.exs:

config :my_app, Oban, plugins: [Oban.Pro.Plugins.BatchManager]

With the BatchManager plugin set to run we can create our batch workers. Let's define a worker that delivers daily emails in a large batch:

defmodule MyApp.EmailBatch do
  use Oban.Pro.Workers.Batch, queue: :mailers

  @impl true
  def process(%Job{args: %{"email" => email}}) do
    MyApp.Mailer.daily_update(email)
  end
end

Note that we define a process/1 callback instead of perform/1 because perform/1 is used behind the scenes to coordinate regular execution and callbacks within the same worker. The process/1 function receives an Oban.Job struct, just like perform/1 would and it should return the same accepted values, i.e. :ok, {:ok, value}, {:error, error}.

The process/1 function above only looks for an "email" key in the job args, but a "batch_id" is also available in meta. We'll modify the function to extract the batch_id as well:

def process(%Job{args: %{"email" => email}, meta: %{"batch_id" => batch_id}}) do
  {:ok, reply} = MyApp.Mailer.daily_update(email)

  track_delivery(batch_id, reply)

  :ok
end

Now the hypothetical track_delivery/2 function will store the delivery details for retrieval later, possibly by one of our handler callbacks.

Typespecs

📚 In order to bridge the gap between module level docs and a guide, each section includes a typespec for the corresponding function. The snippet below defines the types listed in each section.

@type args_or_jobs :: [Job.t() | Job.args()]

@type batch_opts ::
        Job.option()
        | {:batch_id, String.t()}
        | {:batch_callback_args, map()}
        | {:batch_callback_worker, module()}

Inserting Batches

@callback new_batch(args_or_jobs(), [batch_opts()]) :: [Changeset.t()]

Create batches with new_batch/1,2 by passing a list of args and options, or a list of heterogeneous jobs. A list of args and options is transformed into a list of jobs for that batch module. For example, this will build and insert two EmailBatch jobs:

[%{email: "foo@example.com"}, %{email: "bar@example.com"}]
|> MyApp.EmailBatch.new_batch()
|> Oban.insert_all()

To schedule a batch in the future, or override default options, you pass a list of options:

[%{email: "foo@example.com"}, %{email: "bar@example.com"}]
|> MyApp.EmailBatch.new_batch(schedule_in: 60, priority: 1, max_attempts: 3)
|> Oban.insert_all()

The new_batch/1,2 function automatically injects a unique batch_id into each job's meta. A Batch worker is a regular Oban.Worker under the hood, which means you can use new/2 to insert jobs as well, provided you use a deterministic batch id.

Creating a heterogeneous batch is similar, though you may want to provide an explicit worker for [callbacks][#handler-callbacks]:

mail_jobs = Enum.map(mail_args, &MyApp.MailWorker.new/1)
push_jobs = Enum.map(push_args, &MyApp.PushWorker.new/1)

MyApp.BatchWorker.new_batch(mail_jobs ++ push_jobs)

Generating Batch IDs

@callback gen_id() :: String.t()

By default a batch_id is generated as a version 4 random UUID. UUIDs are more than sufficient to ensure that batches are unique between workers and nodes for any period. However, if you require control, you can override batch_id generation at the worker level or pass a value directly to the new_batch/2 function.

To override the batch_id for a particular worker you override the gen_id callback:

defmodule MyApp.BatchWorker do
  use Oban.Pro.Workers.Batch

  # Generate a 24 character long random string instead
  @impl Batch
  def gen_id do
    24
    |> :crypto.strong_rand_bytes()
    |> Base.encode64()
  end
end

The gen_id/0 callback is suited for random/non-deterministic id generation. If you'd prefer to use a deterministic id instead you can pass the batch_id in as an option to new_batch/2:

MyApp.BatchWorker.new_batch(list_of_args, batch_id: "custom-batch-id")

Using this technique you can verify the batch_id in tests or append to the batch manually after it was originally created. For example, you can add to a batch that is scheduled for the future:

batch_id = "daily-batch-#{Date.utc_today()}"
midnight =
  Date.utc_today()
  |> NaiveDateTime.new(~T[11:59:59])
  |> elem(1)
  |> DateTime.from_naive!("Etc/UTC")

# Create the initial batch
initial_args
|> MyApp.BatchWorker.new_batch(batch_id: batch_id, schedule_at: midnight)
|> Oban.insert_all()

# Add items to the batch later in the day
%{batch_id: batch_id, other_arg: "other"}
|> MyApp.BatchWorker.new(schedule_at: midnight)
|> Oban.insert()

When batch jobs execute at midnight they'll all be tracked together.

Handler Callbacks

@callback handle_attempted(job :: Job.t()) :: :ok
@callback handle_completed(job :: Job.t()) :: :ok
@callback handle_discarded(job :: Job.t()) :: :ok
@callback handle_exhausted(job :: Job.t()) :: :ok

After jobs in the batch are processed the BatchManager may insert callback jobs for a callback worker. There are four optional batch handler callbacks that a worker may define:

handle_attempted — called after all jobs in the batch were attempted at least once, regardless of whether they succeeded or not.
handle_completed — called after all jobs in the batch have a completed state. This handler may never be called if one or more jobs keep failing or any are discarded.
handle_discarded — called after any jobs in the batch have a discarded state.
handle_exhausted — called after all jobs in the batch have either a completed or discarded state.

Each handler callback receives an Oban.Job struct with the batch_id in meta and should return :ok. The callbacks are executed as separate isolated jobs, so they may be retried or discarded like any other job.

Here we'll implement each of the optional handler callbacks and have them print out the batch status along with the batch_id:

defmodule MyApp.BatchWorker do
  use Oban.Pro.Workers.Batch

  @impl Batch
  def handle_attempted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.puts("Attempted #{batch_id}")
  end

  @impl Batch
  def handle_completed(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.puts("Completed #{batch_id}")
  end

  @impl Batch
  def handle_discarded(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.puts("Discarded #{batch_id}")
  end

  @impl Batch
  def handle_exhausted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.puts("Exhausted #{batch_id}")
  end
end

Forwarding Callback Args

By default, callback jobs have an empty args map. With the :batch_callback_args option to new_batch/2, you can pass custom args through to each callback. For example, here we're passing a webhook URLs in the args for use callbacks:

MyBatch.new_batch(jobs, batch_callback_args: %{webhook: "https://web.hook"})

Any JSON encodable map may be passed to callbacks, but note that the complete map is stored in each batch job's meta.

Alternate Callback Workers

For some batches, notably those with heterogeneous jobs, it's handy to specify a different worker for callbacks. That is easily accomplished by passing the :batch_callback_worker option to new_batch/2:

MyBatch.new(jobs, batch_callback_worker: MyCallbackWorker)

The callback worker must be an Oban.Worker that defines one or more of the batch callback handlers.

Streaming Batch Jobs

@callback stream_batch_jobs(Job.t(), Keyword.t()) :: Enum.t()

For map/reduce style workflows, or to pull more context from batch jobs, it's possible to load all jobs from the batch with stream_batch_jobs/1,2. The function takes a single batch job and returns a stream of all non-callback jobs in the batch, which you can then operate on with Enum or Stream functions.

As an example, imagine you have a batch that ran for a few thousand accounts and you'd like to notify admins that the batch is complete.

defmodule MyApp.BatchWorker do
  use Oban.Pro.Workers.Batch

  @impl Batch
  def handle_completed(%Job{} = job) do
    {:ok, account_ids} =
      MyApp.Repo.transaction(fn ->
        job
        |> stream_batch_jobs()
        |> Enum.map(& &1.args["account_id"])
      end)

    account_ids
    |> MyApp.Accounts.all()
    |> MyApp.Mailer.notify_admins_about_batch()
  end

Streaming is provided by Ecto's Repo.stream, and it must take place within a transaction. While it may be overkill for small batches, for batches with tens or hundreds of thousands of jobs, it will prevent massive memory spikes or the database grinding to a halt!

Inserting Large Batches

PostgreSQL's binary protocol has a limit of 65,535 parameters that may be sent in a single call. That presents an upper limit on the number of rows that may be inserted at one time, and therefor the number of jobs that may be inserted in a batch all at once.

There is a simple workaround that will allow you to create arbitrarily large batches by inserting them in chunks.

reducer = fn {changesets, index}, multi ->
  Oban.insert_all(multi, "batch_#{index}", changesets)
end

list_of_args
|> MyApp.BatchWorker.new_batch()
|> Enum.chunk_every(5_000)
|> Enum.with_index()
|> Enum.reduce(Ecto.Multi.new(), reducer)
|> MyApp.Repo.transaction()

A few words of explanation:

This chunks every 5,000 jobs, you can use a smaller number but anything under 65,000 should work.
The Enum.with_index/1 call is necessary to provide a unique name for each multi, without that you'll get an error about the batch names conflicting.
The jobs are all inserted together within a transaction and won't start executing until they all go in.

Implementation Notes

Callback jobs are only enqueued if your worker defines the corresponding callback, e.g. a worker that only defines handle_attempted/1 will only have a callback for that event.
Callback jobs are unique, with an infinite period.
The BatchManager uses debouncing to minimize queries and reduce overall load on your database.

← Previous Page Reprioritizer Plugin

Next Page → Chunk Worker

Settings View Source Batch Worker