Add to Cursor Add to Claude Copy for LLM View as MD

Batch Tasks

⚠️

Batch tasks are in beta and may change in future releases.

For tasks that do not need to be executed immediately, you can use batch tasks to automatically accumulate tasks and execute them in a single batch. The input to a batch task is a map that associates the ID of the originating task with it’s input. The output must be the same shape, using the same keys, to send outputs back to the original call site (unless output broadcasting is used, see Broadcasting outputs). For the caller of a batch task, the batched execution is hidden. It is called, and receives output, the same way as a normal task. Though each batch will be executed in one handler function, the call site only receives output corresponding to its input, not the output for the entire batch.

Batch flushing behavior

Batches will accumulate tasks continuously, preventing them from executing until one of the following conditions is true:

The max size for the batch is reached.
The maximum interval has elapsed since the last batch flush.
The size of the payloads for the buffered tasks has exceeded the 4Mb gRPC message limit.

In the following example, the batch task will be executed once 3 tasks have accumulated, or every 200ms, whichever comes first.

examples/python/batch_assign/worker.py

@hatchet.batch_task(
    batch_max_size=3,
    batch_max_interval=timedelta(milliseconds=200),
    input_validator=SimpleInput,
)
async def batch_simple(
    tasks: dict[BatchMemberId, SimpleInput], context: Context
) -> dict[BatchMemberId, SimpleOutput]:
    return {
        id: SimpleOutput(transformed_message=inp.message.upper())
        for id, inp in tasks.items()
    }

examples/typescript/batch_assign/workflow.ts

export const batchSimple = hatchet.batchTask({
  name: 'batch-simple',
  batch: { maxSize: 3, maxInterval: 200 },
  fn: async (tasks: Record<string, SimpleInput>) => {
    const out: Record<string, SimpleOutput> = {};
    Object.entries(tasks).forEach(([id, input]) => {
      out[id] = { transformed_message: input.message.toUpperCase() };
    });
    return out;
  },
});

examples/go/batch_assign/main.go

batchSimple := client.NewStandaloneBatchTask("batch-simple",
	func(ctx hatchet.Context, tasks map[string]SimpleInput) (map[string]SimpleOutput, error) {
		out := make(map[string]SimpleOutput, len(tasks))
		for id, inp := range tasks {
			out[id] = SimpleOutput{TransformedMessage: strings.ToUpper(inp.Message)}
		}
		return out, nil
	},
	hatchet.BatchConfig{
		MaxSize:     3,
		MaxInterval: durationPtr(200 * time.Millisecond),
	},
)

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/batch_assign/worker.rb

BATCH_SIMPLE = HATCHET.batch_task(
  name: 'ruby-e2e-batch-simple',
  batch: Hatchet::BatchTaskConfig.new(max_size: 3, max_interval_ms: 200)
) do |inputs, _ctx|
  inputs.transform_values { |input| { 'transformed_message' => input['message'].upcase } }
end

Batch keys

To allow for accumulation of multiple batches simultaneously using the same workflow, batch keys can be used. For example, if you had multiple tenants, and wanted to ensure that each batch execution did not mix inputs from multiple tenants, batch keys can be used to partition the inputs. In the following example, the batches will flush independently based on the group value.

examples/python/batch_assign/worker.py

@hatchet.batch_task(
    batch_max_size=2,
    batch_max_interval=timedelta(milliseconds=200),
    batch_group_key="input.group",
    input_validator=KeyedInput,
)
async def batch_keyed(
    tasks: dict[BatchMemberId, KeyedInput], context: Context
) -> dict[BatchMemberId, KeyedOutput]:
    unique_keys = len({inp.group for _, inp in tasks.items()})
    return {
        id: KeyedOutput(
            batch_key=inp.group,
            batch_size=len(tasks),
            unique_keys=unique_keys,
            uppercase=inp.message.upper(),
        )
        for id, inp in tasks.items()
    }

examples/typescript/batch_assign/workflow.ts

export const batchKeyed = hatchet.batchTask({
  name: 'batch-keyed',
  batch: { maxSize: 2, maxInterval: 200, groupKey: 'input.group' },
  fn: async (tasks: Record<string, KeyedInput>) => {
    const uniqueKeys = new Set(Object.values(tasks).map((i) => i.group)).size;
    const batchSize = Object.keys(tasks).length;
    const out: Record<string, KeyedOutput> = {};
    Object.entries(tasks).forEach(([id, input]) => {
      out[id] = {
        batch_key: input.group,
        batch_size: batchSize,
        unique_keys: uniqueKeys,
        uppercase: input.message.toUpperCase(),
      };
    });
    return out;
  },
});

examples/go/batch_assign/main.go

batchKeyed := client.NewStandaloneBatchTask("batch-keyed",
	func(ctx hatchet.Context, tasks map[string]KeyedInput) (map[string]KeyedOutput, error) {
		uniqueGroups := make(map[string]struct{})
		for _, inp := range tasks {
			uniqueGroups[inp.Group] = struct{}{}
		}

		out := make(map[string]KeyedOutput, len(tasks))
		for id, inp := range tasks {
			out[id] = KeyedOutput{
				BatchKey:   inp.Group,
				BatchSize:  len(tasks),
				UniqueKeys: len(uniqueGroups),
				Uppercase:  strings.ToUpper(inp.Message),
			}
		}
		return out, nil
	},
	hatchet.BatchConfig{
		MaxSize:     2,
		MaxInterval: durationPtr(200 * time.Millisecond),
		GroupKey:    stringPtr("input.group"),
	},
)

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/batch_assign/worker.rb

BATCH_KEYED = HATCHET.batch_task(
  name: 'ruby-e2e-batch-keyed',
  batch: Hatchet::BatchTaskConfig.new(max_size: 2, max_interval_ms: 200, group_key: 'input.group')
) do |inputs, _ctx|
  unique_keys = inputs.values.map { |i| i['group'] }.uniq.length
  batch_size = inputs.length

  inputs.transform_values do |input|
    {
      'batch_key' => input['group'],
      'batch_size' => batch_size,
      'unique_keys' => unique_keys,
      'uppercase' => input['message'].upcase,
    }
  end
end

Broadcasting outputs

In the default case, batch outputs are mapped 1-1 back to the callers. So if you called a batch task with input a and b at separate call sites, you would receive back outputs a' and b' respectively, despite the fact they would both be executed at the same time, in the same handler. However, if you wish to return the same input back to all callsites, you can use output broadcasting. In the following example, every task will receive the same output regardless of call site. Note, however, that this will only return the same output to every task buffered into the same batch. See the batch flushing rules above for more details.

examples/python/batch_assign/worker.py

@hatchet.batch_task(
    batch_max_size=10,
    batch_max_interval=timedelta(seconds=2),
    input_validator=SimpleInput,
    broadcast_output=True,
)
async def batch_broadcast(
    tasks: dict[BatchMemberId, SimpleInput], context: Context
) -> BroadcastOutput:
    return BroadcastOutput(sum=sum(len(i.message) for _, i in tasks.items()))

examples/typescript/batch_assign/workflow.ts

export const batchBroadcast = hatchet.batchTask({
  name: 'batch-broadcast',
  batch: { maxSize: 10, maxInterval: 2_000, broadcastOutput: true },
  fn: async (tasks: Record<string, SimpleInput>): Promise<BroadcastOutput> => {
    const sum = Object.values(tasks).reduce((acc, i) => acc + i.message.length, 0);
    return { sum };
  },
});

examples/go/batch_assign/main.go

batchBroadcast := client.NewStandaloneBatchTask("batch-broadcast",
	func(ctx hatchet.Context, tasks map[string]SimpleInput) (BroadcastSumOutput, error) {
		sum := 0
		for _, inp := range tasks {
			sum += len(inp.Message)
		}
		return BroadcastSumOutput{Sum: sum}, nil
	},
	hatchet.BatchConfig{
		MaxSize:         10,
		MaxInterval:     durationPtr(2 * time.Second),
		BroadcastOutput: true,
	},
)

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/batch_assign/worker.rb

BATCH_BROADCAST = HATCHET.batch_task(
  name: 'ruby-e2e-batch-broadcast',
  batch: Hatchet::BatchTaskConfig.new(max_size: 10, max_interval_ms: 2_000, broadcast_output: true)
) do |inputs, _ctx|
  { 'sum' => inputs.values.sum { |i| i['message'].length } }
end

Idempotency Durable Tasks

We use cookies

Batch Tasks

Batch flushing behavior

Batch keys

Broadcasting outputs