Map And Reduce

Overview

Map and Reduce work on ordered lists:

• map runs one Action for each item. Its result is a list of Action results in input order.
• reduce passes each item and an accumulator through one Action, one item at a time. Its result is the final accumulator.

Run the cells from top to bottom.

Install The Package

Mix.install([{:jido_action, "~> 3.0.0-beta.22"}])

Define Item Actions

defmodule FlowCollections.Double do
  use Jido.Action, name: "flow_collections_double"

  @impl true
  def run(%{value: value, index: index}, _context) do
    {:ok, %{value: value * 2, index: index}}
  end
end

defmodule FlowCollections.Sum do
  use Jido.Action, name: "flow_collections_sum"

  @impl true
  def run(%{total: total, value: value}, _context) do
    {:ok, %{total: total + value}}
  end
end

Map Then Reduce

defmodule FlowCollections.Example do
  use Jido.Flow, name: "flow_collections_example"

  flow do
    map "doubled",
      collection: input(:values),
      action: FlowCollections.Double,
      params: %{
        value: item(),
        index: item_index()
      }

    reduce "total" do
      collection result("doubled")
      initial %{total: 0}
      action FlowCollections.Sum

      params %{
        total: accumulator(:total),
        value: item(:value)
      }
    end

    output %{items: result("doubled"), total: result("total")}
  end
end

The side arrow shows that the output also reads the complete Map result. Map can run item calls concurrently; Reduce passes each accumulator to the next call. The values shown use the input [1, 2, 3].

{:ok, %{items: items, total: %{total: 12}} = result} =
  Jido.Exec.run(FlowCollections.Example, %{values: [1, 2, 3]})

[%{value: 2, index: 0}, %{value: 4, index: 1}, %{value: 6, index: 2}] = items

result

result("doubled") reads the complete Map result: an ordered list of FlowCollections.Double results. The Reduce reads that list without consuming it, so the Flow output can return both results.

Local References

An item ID is deterministic for the Flow, the component name, and the item index. Each component uses its own name, so a Reduce over a Map result has different item IDs than the Map. Do not use item IDs as execution IDs.

Rules

• An empty Map returns []. An empty Reduce returns its initial value.
• The Reduce initial value must be a map or a Jido.Action.Output value. A list or scalar compiles, but fails at runtime.
• Each Reduce call returns the next accumulator.
• The Map Action runs once per item, no matter how many components read the Map result.
• Map and Reduce targets must be Action modules, not Flow modules or inline bodies.

Map Error Modes

With the default on_error: :fail_fast, the first item failure fails the Map.

With on_error: :collect_errors, an Action failure becomes a record instead of failing the Map. A reference failure in params, such as a missing item key, still fails the Map. The Map returns one record for each item, in input order:

%{status: :ok, value: action_result}
%{status: :error, error: %{type: type, message: message}}

type is the error module name as a string, such as "Elixir.Jido.Action.Error.ExecutionFailureError". message is the error message. The record does not include error details or stacktraces.

defmodule FlowCollections.Check do
  use Jido.Action, name: "flow_collections_check"

  @impl true
  def run(%{value: value}, _context) when value < 0 do
    {:error, Jido.Action.Error.execution_error("negative value")}
  end

  def run(%{value: value}, _context), do: {:ok, %{value: value}}
end

defmodule FlowCollections.CollectErrors do
  use Jido.Flow, name: "flow_collections_collect_errors"

  flow do
    map "checked",
      collection: input(:values),
      action: FlowCollections.Check,
      params: %{value: item()},
      on_error: :collect_errors

    output %{checked: result("checked")}
  end
end

{:ok, %{checked: checked}} =
  Jido.Exec.run(FlowCollections.CollectErrors, %{values: [1, -2, 3]})

checked

Choose the mode on purpose. Jido does not add a fallback or retry for you.

Concurrency

Map items can run concurrently when max_concurrency is greater than 1:

Jido.Exec.run(
  FlowCollections.Example,
  %{values: [1, 2, 3]},
  %{},
  max_concurrency: 2
)

The result order still matches the input order. Reduce always runs one item at a time. See Execution for runtime options.

Data Form

In a map definition, use kind: :map with collection, action, params, and on_error, or kind: :reduce with collection, initial, action, and params. See Flow Data Definitions.