GenStage and Flow by @josevalim at ElixirConf

Transcript

1. @elixirlang / elixir-lang.org
2. GenStage & Flow github.com/elixir-lang/gen_stage
3. Prelude: From eager,
 to lazy,
 to concurrent,
 to distributed
4. Example problem: word counting
5. “roses are red\n violets are blue\n" %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1}
6. Eager
7. File.read!("source") "roses are red\n violets are blue\n"
8. File.read!("source") |> String.split("\n") ["roses are red", "violets are blue"]
9. File.read!("source") |> String.split("\n") |> Enum.flat_map(&String.split/1) ["roses", “are", “red", "violets", "are", "blue"]
10. File.read!("source") |> String.split("\n") |> Enum.flat_map(&String.split/1) |> Enum.reduce(%{}, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1}
11. Eager • Simple • Efficient for small collections • Inefficient for large collections with multiple passes
12. File.read!(“really large file") |> String.split("\n") |> Enum.flat_map(&String.split/1)
13. Lazy
14. File.stream!("source", :line) #Stream<...>
15. File.stream!("source", :line) |> Stream.flat_map(&String.split/1) #Stream<...>
16. File.stream!("source", :line) |> Stream.flat_map(&String.split/1) |> Enum.reduce(%{}, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1}
17. Lazy • Folds computations, goes item by item • Less memory usage at the cost of computation • Allows us to work with large or infinite collections
18. Concurrent
19. #Stream<...> File.stream!("source", :line)
20. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) #Flow<...>
21. %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1} File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) |> Enum.into(%{})
22. Flow • We give up ordering and process locality for concurrency • Tools for working with bounded and unbounded data
23. Flow • It is not magic! There is an overhead when data flows through processes • Requires volume and/or cpu/io-bound work to see benefits
24. Flow stats • ~1200 lines of code (LOC) • ~1300 lines of documentation (LOD)
25. Topics • 1200LOC: How is flow implemented? • 1300LOD: How to reason about flows?
26. GenStage
27. GenStage • It is a new behaviour • Exchanges data between stages transparently with back-pressure • Breaks into producers, consumers and producer_consumers
28. GenStage producer consumer producer producer consumer producer consumer consumer
29. GenStage: Demand-driven B A Producer Consumer 1. consumer subscribes to producer 2. consumer sends demand 3. producer sends events Asks 10 Sends max 10 Subscribes
30. B C A Asks 10 Asks 10 Sends max 10 Sends max 10 GenStage: Demand-driven
31. • It is a message contract • It pushes back-pressure to the boundary • GenStage is one impl of this contract GenStage: Demand-driven
32. GenStage example
33. B A Printer (Consumer) Counter (Producer)
34. defmodule Producer do use GenStage def init(counter) do {:producer, counter} end def handle_demand(demand, counter) when demand > 0 do events = Enum.to_list(counter..counter+demand-1) {:noreply, events, counter + demand} end end
35. state demand handle_demand 0 10 {:noreply, [0, 1, …, 9], 10} 10 5 {:noreply, [10, 11, 12, 13, 14], 15} 15 5 {:noreply, [15, 16, 17, 18, 19], 20}
36. defmodule Consumer do use GenStage def init(:ok) do {:consumer, :the_state_does_not_matter} end def handle_events(events, _from, state) do Process.sleep(1000) IO.inspect(events) {:noreply, [], state} end end
37. {:ok, counter} = GenStage.start_link(Producer, 0) {:ok, printer} = GenStage.start_link(Consumer, :ok) GenStage.sync_subscribe(printer, to: counter) (wait 1 second) [0, 1, 2, ..., 499] (500 events) (wait 1 second) [500, 501, 502, ..., 999] (500 events)
38. Subscribe options • max_demand: the maximum amount of events to ask (default 1000) • min_demand: when reached, ask for more events (default 500)
39. max_demand: 10, min_demand: 5 1. the consumer asks for 10 items 2. the consumer receives 10 items 3. the consumer processes 5 of 10 items 4. the consumer asks for more 5 5. the consumer processes the remaining 5
40. max_demand: 10, min_demand: 0 1. the consumer asks for 10 items 2. the consumer receives 10 items 3. the consumer processes 10 items 4. the consumer asks for more 10 5. the consumer waits
41. GenStage dispatchers
42. Dispatchers • Per producer • Effectively receive the demand and send data • Allow a producer to dispatch to multiple consumers at once
43. prod 1 2,5 3 4 1,2,3,4,5 DemandDispatcher
44. prod 1,2,3 1,2,3 1,2,3 1,2,3 1,2,3 BroadcastDispatcher
45. prod 1,5 2,6 3 4 1,2,3,4,5,6 PartitionDispatcher rem(event, 4)
46. Validating GenStage
47. GenStage goals • Support generic producers • Replace GenEvent • Introduce DynamicSupervisor
48. http://bit.ly/genstage
49. Topics • 1200LOC: How is flow implemented? • Its core is a 80LOC stage • 1300LOD: How to reason about flows?
50. Flow
51. “roses are red\n violets are blue\n" %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1}
52. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) #Flow<...>
53. %{“are” => 2, “blue” => 1, “red” => 1, “roses" => 1, “violets" => 1} File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) |> Enum.into(%{})
54. File.stream!("source", :line) |> Flow.from_enumerable() Producer
55. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) Producer Stage 1 Stage 2 Stage 3 Stage 4 DemandDispatcher
56. "blue" Producer Stage 1 Stage 2 Stage 3 Stage 4 "roses are red" "roses" "are" "red" "violets are blue" "violets" "are"
57. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) Producer Stage 1 Stage 2 Stage 3 Stage 4
58. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) Producer Stage 1 Stage 2 Stage 3 Stage 4
59. Producer Stage 1 Stage 2 Stage 3 Stage 4 "roses are red" %{“are” => 1, “red” => 1, “roses" => 1} "violets are blue" %{“are” => 1, “blue” => 1, “violets" => 1}
60. Producer Stage 1 Stage 2 Stage 3 Stage 4 %{“are” => 1, “red” => 1, “roses" => 1} %{“are” => 1, “blue” => 1, “violets" => 1}
61. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) Producer Stage 1 Stage 2 Stage 3 Stage 4 Stage A Stage B Stage C Stage D
62. %{“are” => 1, “red” => 1} Stage C Stage A Stage B Stage D Stage 1 Stage 4 Producer "roses are red" %{“roses” => 1} Stage 2 Stage 3
63. %{“are” => 2, “red” => 1} Stage C Stage A Stage B Stage D Stage 1 Stage 4 Producer "roses are red" %{“roses” => 1} "violets are blue" Stage 2 Stage 3
64. Mapper 4 Reducer C Producer Reducer A Reducer B Reducer D Mapper 1 Mapper 2 Mapper 3 DemandDispatcher PartitionDispatcher
65. File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(fn -> %{} end, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) |> Enum.into(%{}) • reduce/3 collects all data into maps • when it is done, the maps are streamed • into/2 collects the state into a map
66. Windows and triggers • If reduce/3 runs until all data is processed… what happens on unbounded data? • See Flow.Window documentation.
67. Postlude: From eager,
 to lazy,
 to concurrent,
 to distributed
68. Enum (eager) File.read!("source") |> String.split("\n") |> Enum.flat_map(&String.split/1) |> Enum.reduce(%{}, fn word, map -> Map.update(map, word, 1, & &1 + 1) end)
69. Stream (lazy) File.stream!("source", :line) |> Stream.flat_map(&String.split/1) |> Enum.reduce(%{}, fn word, map -> Map.update(map, word, 1, & &1 + 1) end)
70. Flow (concurrent) File.stream!("source", :line) |> Flow.from_enumerable() |> Flow.flat_map(&String.split/1) |> Flow.partition() |> Flow.reduce(%{}, fn word, map -> Map.update(map, word, 1, & &1 + 1) end) |> Enum.into(%{})
71. Flow features • Provides map and reduce operations, partitions, flow merge, flow join • Configurable batch size (max & min demand) • Data windowing with triggers and watermarks
72. Distributed? • Flow API has feature parity with frameworks like Apache Spark • However, there is no distribution nor execution guarantees
73. CHEN, Y., ALSPAUGH, S., AND KATZ, R. Interactive analytical processing in big data systems: a cross-industry study of MapReduce workloads “small inputs are common in practice: 40–80% of Cloudera customers’ MapReduce jobs and 70% of jobs in a Facebook trace have ≤ 1GB of input”
74. GOG, I., SCHWARZKOPF, M., CROOKS, N., GROSVENOR, M. P., CLEMENT, A., AND HAND, S. Musketeer: all for one, one for all in data processing systems. “For between 40-80% of the jobs submitted to MapReduce systems, you’d be better off just running them on a single machine”
75. Distributed? • Single machine matters - try it! • The gap between concurrent and distributed in Elixir is small • Durability concerns will be tackled next
76. Thank-yous
77. Library authors • Give GenStage a try • Build your own producers • TIP: use @behaviour GenStage if you want to make it an optional dep
78. Inspirations • Akka Streams - back pressure contract • Apache Spark - map reduce API • Apache Beam - windowing model • Microsoft Naiad - stage notifications
79. The Team • The Elixir Core Team • Specially James Fish (fishcakez) • And Eric and Chris for therapy sessions
80. consulting and software engineering Built and designed at
81. consulting and software engineering Elixir coaching Elixir design review Custom development
82. @elixirlang / elixir-lang.org