Class: Hatchet::BatchTaskConfig

Inherits:
Object
  • Object
show all
Defined in:
lib/hatchet/batch.rb,
sig/hatchet/batch.rbs

Overview

Configures a task as a batch task: concurrent runs are buffered and dispatched together as a single execution once max_size is reached, max_interval_ms elapses, or (if group_key is set) once group_max_runs concurrent batches per group are exceeded.

When batch is set on a task, the task's block receives a single Hash argument mapping each buffered run's task-run external id to that run's input, instead of a single input value. Unless broadcast_output is true, the block must return a Hash with the exact same key set, mapping each id to that run's output.

Examples:

Hatchet::BatchTaskConfig.new(max_size: 3, max_interval_ms: 200, group_key: "input.group")

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_size:, max_interval_ms: nil, group_key: nil, group_max_runs: nil, broadcast_output: false) ⇒ BatchTaskConfig

Returns a new instance of BatchTaskConfig.

Parameters:

  • max_size (Integer) —

    Maximum items per batch. Must be positive.

  • max_interval_ms (Integer, nil) (defaults to: nil) —

    Time before batch flushes, in milliseconds. Must be positive when provided.

  • group_key (String, nil) (defaults to: nil) —

    CEL expression to partition batches, e.g. "input.group"

  • group_max_runs (Integer, nil) (defaults to: nil) —

    Concurrent batches per group. Must be positive when provided.

  • broadcast_output (Boolean) (defaults to: false) —

    Whether the block's return value is broadcast to every batch member

  • max_size: (Integer)
  • max_interval_ms: (Integer, nil) (defaults to: nil)
  • group_key: (String, nil) (defaults to: nil)
  • group_max_runs: (Integer, nil) (defaults to: nil)
  • broadcast_output: (Boolean) (defaults to: false)

Raises:

  • (ArgumentError) 


36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/hatchet/batch.rb', line 36

def initialize(max_size:, max_interval_ms: nil, group_key: nil, group_max_runs: nil, broadcast_output: false)
  raise ArgumentError, "max_size must be positive" unless max_size.positive?
  raise ArgumentError, "max_interval_ms must be positive when provided" if !max_interval_ms.nil? && !max_interval_ms.positive?
  raise ArgumentError, "group_max_runs must be positive when provided" if !group_max_runs.nil? && !group_max_runs.positive?

  @max_size = max_size
  @max_interval_ms = max_interval_ms
  @group_key = group_key
  @group_max_runs = group_max_runs
  @broadcast_output = broadcast_output
end

Instance Attribute Details

#broadcast_output ⇒ Boolean (readonly)

Returns When true, the block returns a single value broadcast to every member of the batch.

Returns:

  • (Boolean) —

    When true, the block returns a single value broadcast to every member of the batch



29
30
31
 
# File 'lib/hatchet/batch.rb', line 29

def broadcast_output
  @broadcast_output
end

#group_key ⇒ String? (readonly)

Returns CEL expression evaluated against each item's input to partition items into independent batches.

Returns:

  • (String, nil) —

    CEL expression evaluated against each item's input to partition items into independent batches



23
24
25
 
# File 'lib/hatchet/batch.rb', line 23

def group_key
  @group_key
end

#group_max_runs ⇒ Integer? (readonly)

Returns Maximum number of concurrent batches per group.

Returns:

  • (Integer, nil) —

    Maximum number of concurrent batches per group



26
27
28
 
# File 'lib/hatchet/batch.rb', line 26

def group_max_runs
  @group_max_runs
end

#max_interval_ms ⇒ Integer? (readonly)

Returns Maximum time to wait before flushing a partially-filled batch, in milliseconds.

Returns:

  • (Integer, nil) —

    Maximum time to wait before flushing a partially-filled batch, in milliseconds



20
21
22
 
# File 'lib/hatchet/batch.rb', line 20

def max_interval_ms
  @max_interval_ms
end

#max_size ⇒ Integer (readonly)

Returns Maximum number of items buffered before the batch is flushed.

Returns:

  • (Integer) —

    Maximum number of items buffered before the batch is flushed



17
18
19
 
# File 'lib/hatchet/batch.rb', line 17

def max_size
  @max_size
end

Instance Method Details

#to_proto ⇒ V1::TaskBatchConfig

Returns:



49
50
51
52
53
54
55
56
 
# File 'lib/hatchet/batch.rb', line 49

def to_proto
  args = { batch_max_size: @max_size, broadcast_output: @broadcast_output }
  args[:batch_max_interval_ms] = @max_interval_ms if @max_interval_ms
  args[:batch_group_key] = @group_key if @group_key
  args[:batch_group_max_runs] = @group_max_runs if @group_max_runs

  ::V1::TaskBatchConfig.new(**args)
end