Gain insights into data processing and concurrency in Elixir. Explore GenServer, GenStage, Flow, and Broadway to enhance performance, set up pipelines, and handle increased workloads efficiently.

Concurrent Data Processing in Elxir by The Pragmatic Programmer-01.png

myDokcer.tar.gz

ch1_1

ch_3

ch_3.1

ch1_1-task supervisor

ch_4.1

ch_5.1 RabbitMQ

Ch_5.2 RabbitMQ

Most engineers don’t think that concurrency is distinct from programming. Data processing includes transforming information in some way, like reporting, data aggregation, or analytics. On the other hand, concurrency involves significantly improving performance.

This course is designed to help you navigate the world of concurrency tools available in the Elixir ecosystem. You’ll explore popular modules and libraries like GenServer, GenStage, Flow, and Broadway. You’ll be able to ingest data, set up pipelines, and run concurrent processes with Elixir.

By the end of the course, you’ll have a wide range of new techniques to simplify your product, improve code performance, and make your application more resilient to increased workloads.

Concurrent Data Processing in Elixir

# Blueprint of our job processing system

We used the `Task` module in the previous chapter to start a new process when sending each email. We’ll adopt the same approach for our job-processing system. Rather than having a single `GenServer` that does all the work, we start a `GenServer` process for each job. This helps us leverage concurrency and improve performance.

We also don’t want to lose any features that we have introduced so far in this chapter. We want to be able to retry failed jobs and configure the process to suit our needs. Finally, we’ll learn a new type of supervisor to help us manage these processes and provide fault tolerance. This figure gives us a high-level overview of what we’re going to build:

# Set up the jobber project

To begin, let’s scaffold a new Elixir project with a supervision tree. We'll call it `jobber`, which is not the most creative name, but it will do the job (pun intended):

```shell
mix new jobber --sup
```
Then, we change our current directory to `jobber`.

After this, we create our GenServer process, which will do the heavy lifting. We create the `job.ex` file in `lib/jobber` with the following content:

```Elixir
defmodule Jobber.Job do
   use GenServer
   require Logger
end
```

Like we did before, we bring in the GenServer behaviour and implementation. We also use Elixir’s **`Logger`** module for log messages.


# Initialize the job process

Now, let’s implement the `init/1` callback and the initial process configuration.

We use a `struct` to keep track of the process state. **Structs** are just like maps, but they provide additional features, including default values for the struct’s attributes. Each job will have several fields, such as a short ID to identify the process uniquely, the work function itself, and the already familiar `retries` and `max_retries`. We also keep track of each job’s status, which will be either `new`, `errored`, `failed`, or `done`.

We add the following struct definition to `jobber/lib/jobber/job.ex`:

```Elixir
defstruct [:work, :id, :max_retries, retries: 0, status: "new"]
```
Then, implement the `init/1` callback:

```Elixir
def init(args) do
   work = Keyword.fetch!(args, :work)
   id = Keyword.get(args, :id, random_job_id()) 
   max_retries = Keyword.get(args, :max_retries, 3)
   state = %Jobber.Job{id: id, work: work, max_retries: max_retries}
   {:ok, state, {:continue, :run}} 
end
```

The code is similar to SendServer. We use the **`Keyword.fetch!/2`** function here to get the work argument. This function throws an error if we forget to provide this parameter by accident. We also generate a default ID in case an ID is not available. There are different ways to generate a random identifier in Elixir. We use Erlang’s `:crypto` module and define `random_job_id/0` like this:

```Elixir
defp random_job_id() do
   :crypto.strong_rand_bytes(5) |> Base.url_encode64(padding: false)
end
```

This ensures that the ID is unique and short at the same time, making it easy to work with. We must also add `:crypto` to the `extra_applications` list in `mix.exs`:

```Elixir
def application do
  [
     extra_applications: [:logger, :crypto],
     mod: {Jobber.Application, []}
  ]
end
```

Although the `:crypto` module is available to us, it isn’t a part of the Erlang standard library, so it must be added to `extra_applications` when used.

---


Make a new mix project to explain building a job processing system.

Getting Started

Easy Concurrency with the Task Module

Long-Running Processes Using GenServer

Data Processing Pipelines with GenStage

Process Collections with Flow

Data Ingestion Pipelines with Broadway

Concluding the Course

The blueprint of our job-processing system

Blueprint of our job processing system