{"id":8798,"date":"2019-02-25T14:03:31","date_gmt":"2019-02-25T17:03:31","guid":{"rendered":"http:\/\/blog.plataformatec.com.br\/?p=8798"},"modified":"2019-02-26T10:13:47","modified_gmt":"2019-02-26T13:13:47","slug":"announcing-broadway","status":"publish","type":"post","link":"http:\/\/blog.plataformatec.com.br\/2019\/02\/announcing-broadway\/","title":{"rendered":"Announcing Broadway"},"content":{"rendered":"<p>In December 2018, we published the <a href=\"http:\/\/blog.plataformatec.com.br\/2018\/12\/plataformatec-and-elixir-2018-in-review\/\" target=\"_blank\" rel=\"noopener\">Plataformatec and Elixir: 2018 in review<\/a> article, in which we shared some of our achievements throughout the year, as well as some of the goals we planned for 2019. One of these goals was to engage the R&amp;D team to develop a new open source tool that could streamline data processing pipelines with Elixir.<\/p>\n<p>Today, we are glad to announce the first official release of this tool: <a href=\"https:\/\/github.com\/plataformatec\/broadway\" target=\"_blank\" rel=\"noopener\">Broadway v0.1<\/a>. Broadway was mainly designed to help developers build concurrent, multi-stage data ingestion and data processing pipelines. It allows developers to consume data efficiently from different sources, such as Amazon SQS, RabbitMQ, and others.<\/p>\n<h2>Motivation<\/h2>\n<p>We have worked with many companies building data processing pipelines and we have noticed that they were often reimplementing the same features and also running into common pitfalls when assembling complex GenStage topologies. The goal of Broadway is to significantly cut down the development time to assemble those pipelines, while providing many features and avoiding common pitfalls.<\/p>\n<h2>Features<\/h2>\n<p>Broadway comes with a handful of features that take the burden of defining concurrent GenStage topologies and provide a simple configuration API that automatically defines concurrent producers, concurrent processing, batch handling, and more, leading to both time and cost efficient ingestion and processing of data. Some of those features include:<\/p>\n<ul>\n<li>Back-pressure<\/li>\n<li>Automatic acknowledgements<\/li>\n<li>Batching<\/li>\n<li>Automatic restarts in case of failures<\/li>\n<li>Graceful shutdown<\/li>\n<li>Built-in testing<\/li>\n<li>Partitioning<\/li>\n<\/ul>\n<p>Other features are already on the roadmap, such as:<\/p>\n<ul>\n<li>Rate-limiting<\/li>\n<li>Statistics \/ Metrics<\/li>\n<li>Back-off<\/li>\n<\/ul>\n<h2>How does it work?<\/h2>\n<p>Similarly to other process-based behaviours, we can create a Broadway-based data pipeline by defining a module like this:<\/p>\n<pre><code class=\"elixir\">defmodule MyBroadway do\n  use Broadway\n\n  alias Broadway.Message\n\n  def start_link(_opts) do\n    Broadway.start_link(__MODULE__,\n      name: __MODULE__,\n      producers: [\n        sqs: [\n          module: {BroadwaySQS.Producer, queue_name: \"my_queue\"}\n        ]\n      ],\n      processors: [\n        default: [stages: 50]\n      ],\n      batchers: [\n        s3_odd: [stages: 2, batch_size: 10],\n        s3_even: [stages: 1, batch_size: 10]\n      ]\n     )\n   end\n\n    ...callbacks...\n\n  end\n<\/code><\/pre>\n<p>The configuration above defines a pipeline with:<\/p>\n<ul>\n<li>1 producer<\/li>\n<li>50 processors<\/li>\n<li>1 batcher named <code>:s3_odd<\/code> with 2 consumers<\/li>\n<li>1 batcher named <code>:s3_even<\/code> with 1 consumer<\/li>\n<\/ul>\n<pre><code class=\"text\"> \n                         [producer_1]\n                             \/ \\\n                            \/   \\\n                           \/     \\\n                          \/       \\\n                 [processor_1] [processor_2] ... [processor_50]  &lt;- process each message\n                          \/\\     \/\\\n                         \/  \\   \/  \\\n                        \/    \\ \/    \\\n                       \/      x      \\ \n                      \/      \/ \\      \\\n                     \/      \/   \\      \\\n                    \/      \/     \\      \\\n              [batcher_s3_odd]  [batcher_s3_even]\n                    \/\\                  \\\n                   \/  \\                  \\\n                  \/    \\                  \\\n                 \/      \\                  \\\n [consumer_s3_odd_1] [consumer_s3_odd_2]  [consumer_s3_even_1] &lt;- process each batch\n<\/code><\/pre>\n<p>In order to process the data provided by the SQS producer, we need to implement two Broadway callbacks: <code>handle_message\/3<\/code>, invoked by processors for each message, and <code>handle_batch\/4<\/code>,&nbsp;invoked by consumers for each batch:<\/p>\n<pre><code class=\"elixir\">\ndefmodule MyBroadway do\n  use Broadway\n\n  alias Broadway.Message\n\n  ...start_link...\n\n  @impl true\n  def handle_message(_, %Message{data: data} = message, _) when is_odd(data) do\n    message\n    |&gt; Message.update_data(&amp;process_data\/1)\n    |&gt; Message.put_batcher(:s3_odd)\n  end\n\n  def handle_message(_, %Message{data: data} = message, _) do\n    message\n    |&gt; Message.update_data(&amp;process_data\/1)\n    |&gt; Message.put_batcher(:s3_even)\n  end\n\n  @impl true\n  def handle_batch(:s3_odd, messages, _batch_info, _context) do\n    # Send batch of messages to S3 \"odd\" bucket\n  end\n\n  def handle_batch(:s3_even, messages, _batch_info, _context) do\n    # Send batch of messages to S3 \"even\" bucket\n  end\n\n  defp process_data(data) do\n    # Do some calculations, generate a JSON representation, etc.\n  end\nend\n<\/code><\/pre>\n<p>At the end of the pipeline, messages are automatically acknowledged by the SQS producer.<\/p>\n<p>Note: You can also use existing GenStage producers as the source of the pipeline. For more information see the <a href=\"https:\/\/hexdocs.pm\/broadway\/custom-producers.html\" target=\"_blank\" rel=\"noopener\">Custom Producers Guide<\/a>.<\/p>\n<h2>What&#8217;s next?<\/h2>\n<p>There&#8217;s a lot more about Broadway. We put a lot of effort in the <a href=\"https:\/\/hexdocs.pm\/broadway\/\" target=\"_blank\" rel=\"noopener\">documentation<\/a>, including <a href=\"https:\/\/hexdocs.pm\/broadway\/architecture.html\" target=\"_blank\" rel=\"noopener\">architectural aspects<\/a> and a full guide on consuming events from <a href=\"https:\/\/hexdocs.pm\/broadway\/amazon-sqs.html\" target=\"_blank\" rel=\"noopener\">Amazon SQS queues<\/a>.<\/p>\n<p>As with any first release, we expect to gather as much feedback as possible from the community so we can incorporate new use cases and improve the API appropriately. You can also contribute to this project in many ways, either by giving the project a try or building your own connector. The <a href=\"https:\/\/github.com\/plataformatec\/broadway_sqs\" target=\"_blank\" rel=\"noopener\">SQS connector<\/a> presented in this post is already available. A RabbitMQ connector is also planned and should be available soon.<\/p>\n<p>We plan to continue pushing the Elixir ecosystem forward! If you would like to build Elixir systems together with our team, <a href=\"http:\/\/plataformatec.com.br\/contact\" target=\"_blank\" rel=\"noopener\">reach out<\/a> and we will be glad to discuss anything Elixir related, from data pipelines to web applications and distributed systems!<\/p>\n<p>Happy coding!<\/p>\n","protected":false},"excerpt":{"rendered":"<p>In December 2018, we published the Plataformatec and Elixir: 2018 in review article, in which we shared some of our achievements throughout the year, as well as some of the goals we planned for 2019. One of these goals was to engage the R&amp;D team to develop a new open source tool that could streamline &#8230; <a class=\"read-more-link\" href=\"http:\/\/blog.plataformatec.com.br\/2019\/02\/announcing-broadway\/\">\u00bb<\/a><\/p>\n","protected":false},"author":77,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"ngg_post_thumbnail":0,"footnotes":""},"categories":[1],"tags":[143],"aioseo_notices":[],"jetpack_sharing_enabled":true,"jetpack_featured_media_url":"","_links":{"self":[{"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/posts\/8798"}],"collection":[{"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/users\/77"}],"replies":[{"embeddable":true,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/comments?post=8798"}],"version-history":[{"count":22,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/posts\/8798\/revisions"}],"predecessor-version":[{"id":8827,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/posts\/8798\/revisions\/8827"}],"wp:attachment":[{"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/media?parent=8798"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/categories?post=8798"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/blog.plataformatec.com.br\/wp-json\/wp\/v2\/tags?post=8798"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}