Data Pipelines

Consumers demand events — producers never overwhelm consumers.

41.1 GenStage — back-pressured producer/consumer stages

Consumers demand events — producers never overwhelm consumers.

defmodule Numbers do
  use GenStage
  def start_link(n), do: GenStage.start_link(__MODULE__, n, name: __MODULE__)
  def init(n), do: {:producer, n}
  def handle_demand(demand, n) do
    events = Enum.to_list(n..(n + demand - 1))
    {:noreply, events, n + demand}
  end
end

defmodule Printer do
  use GenStage
  def start_link(_), do: GenStage.start_link(__MODULE__, :ok)
  def init(:ok), do: {:consumer, :ok, subscribe_to: [{Numbers, max_demand: 10}]}
  def handle_events(events, _from, state) do
    Enum.each(events, &IO.inspect/1)
    {:noreply, [], state}
  end
end

41.2 Flow — parallel collection processing

File.stream!("big.txt")
|> Flow.from_enumerable()
|> Flow.flat_map(&String.split/1)
|> Flow.partition()
|> Flow.reduce(fn -> %{} end, fn w, acc -> Map.update(acc, w, 1, &(&1 + 1)) end)
|> Enum.to_list()

41.3 Broadway — data ingestion pipelines

Consume from SQS, Kafka, RabbitMQ, Google Pub/Sub with batching, acknowledgements, rate limiting, graceful shutdown.

defmodule MyPipeline do
  use Broadway
  def start_link(_) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [module: {BroadwayRabbitMQ.Producer, queue: "events"}, concurrency: 1],
      processors: [default: [concurrency: 10]],
      batchers: [default: [batch_size: 100, batch_timeout: 1000]])
  end
  def handle_message(_, msg, _), do: Broadway.Message.update_data(msg, &Jason.decode!/1)
  def handle_batch(_, msgs, _, _), do: (insert_all(msgs); msgs)
end