DEV Community: Geazi Anc

Nesse artigo, iremos desenvolver uma pipeline de dados bem simples em tempo real utilizando o Apache Flink em conjunto com a versão 3 da linguagem de programação Scala, fazendo o uso do Pub/Sub como message broker 🚀

Geazi Anc — Tue, 07 Jan 2025 15:11:42 +0000

Engenharia de Dados com Scala: masterizando o processamento de dados em tempo real com Apache Flink e Google Pub/Sub

Geazi Anc ・ Aug 9 '24

#dataengineering #scala #braziliandevs #flink

New article alert! Data Engineering with Scala: mastering data processing with Apache Flink and Pub/Sub ❤️‍🔥

Geazi Anc — Sat, 04 Jan 2025 00:14:28 +0000

Data Engineering with Scala: Mastering Real-Time Data Processing with Apache Flink and Google Pub/Sub

Geazi Anc ・ Oct 18 '24

#dataengineering #scala #datascience #flink

Análise de dados de tráfego aéreo em tempo real com Spark Structured Streaming e Apache Kafka

Geazi Anc — Mon, 28 Oct 2024 15:50:57 +0000

Atualmente, vivemos em um mundo onde peta bytes de dados são gerados a cada segundo. Como tal, a análise e o processamento desses dados em tempo real torna-se mais do que essencial para uma empresa que busca gerar insights de negócios com mais precisão conforme dados e mais dados são produzidos.

Hoje, vamos desenvolver uma análise de dados em tempo real com base em dados fictícios de um tráfego aéreo utilizando Spark Structured Streaming e Apache Kafka. Caso não saiba o que são essas tecnologias, sugiro a leitura de meu artigo que escrevi introduzindo elas com mais detalhes, assim como outros conceitos que serão abordados no decorrer desse artigo. Então, não esquece de conferir lá 💚.

Uma breve Introdução ao processamento de dados em tempo real com Spark Structured Streaming e Apache Kafka

Geazi Anc ・ Sep 29 '22

#python #pyspark #dataengineering #apachekafka

Você pode conferir o projeto completo em meu GitHub.

Arquitetura

Pois bem, imagine que você, pessoa engenheira de dados, trabalhe em uma empresa aérea chamada de SkyX, onde a cada segundo dados sobre o tráfego aéreo são gerados.

Você foi solicitada para desenvolver uma dashboard que exibe em tempo real dados desses voos, como um rank das cidades mais visitadas no exterior; as cidades onde mais saem pessoas; e as aeronaves que mais transportam pessoas ao redor do mundo.

Esses são os dados que são gerados a cada voo:

aircraft_name: nome da aeronave. Na SkyX, só existem apenas cinco aeronaves disponíveis.
From: cidade de onde a aeronave está partindo. A SkyX só realiza voos entre cinco cidades ao redor do mundo.
To: cidade de destino da aeronave. Como foi dito, a SkyX só realiza voos entre cinco cidades ao redor do mundo.
Passengers: quantidade de passageiros que a aeronave está transportando. Todas as aeronaves da SkyX transportam entre 50 e 100 pessoas a cada voo.

A seguir está a arquitetura básica de nosso projeto:

Produtor: responsável por produzir dados do tráfego aéreo das aeronaves e enviá-los à um tópico do Apache Kafka.
Consumidor: apenas observa os dados que chegam em tempo real ao tópico do Apache Kafka.
Análise de dados: três dashboards que processam e analisam em tempo real os dados que chegam no tópico do Apache Kafka. Análise das cidades que mais recebem turistas; análise das cidades que mais saem pessoas para visitar outras cidades; e análise das aeronaves da SkyX que mais transportam pessoas entre as cidades ao redor do mundo.

Preparando o ambiente de desenvolvimento

Este tutorial assume que você já tenha o PySpark instalado em sua máquina. Caso ainda não tenha, confira as etapas na própria documentação.

Já para o Apache Kafka, vamos utilizar ele por meio de conteinerização via Docker 🎉🐳.

E, por fim, vamos utilizar o Python através de um ambiente virtual.

Apache Kafka por conteinerização via Docker

Sem mais delongas, crie uma pasta chamada skyx e adicione o arquivo docker-compose.yml dentro dela.

$ mkdir skyx
$ cd skyx
$ touch docker-compose.yml

Agora, adicione o seguinte conteúdo dentro do arquivo docker-compose:

version: '3.9'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

Feito! Já podemos subir nosso servidor do Kafka. Para isso, digite o seguinte comando no terminal:

$ docker compose up -d
$ docker compose ps

NAME                                COMMAND                  SERVICE             STATUS              PORTS
skyx-kafka-1       "/etc/confluent/dock…"   kafka               running             9092/tcp, 0.0.0.0:29092->29092/tcp
skyx-zookeeper-1   "/etc/confluent/dock…"   zookeeper           running             2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp

Observação: este tutorial está utilizando a versão 2.0 do Docker Compose. É por este motivo que não há o "-" entre docker e compose ☺.

Agora, precisamos criar um tópico dentro do Kafka que irá armazenar os dados enviados em tempo real pelo produtor. Para isso, vamos acessar o Kafka dentro do contêiner:

$ docker compose exec kafka bash

E enfim criar o tópico, chamado de airtraffic.

$ kafka-topics --create --topic airtraffic --bootstrap-server localhost:29092

Created topic airtraffic.

Criação do ambiente virtual

Para desenvolvermos nosso produtor, ou seja, a aplicação que será responsável por enviar os dados do tráfego aéreo em tempo real para o tópico do Kafka, precisamos fazer o uso da biblioteca kafka-python. O kafka-python é uma biblioteca desenvolvida pela comunidade que nos permite desenvolver produtores e consumidores que se integram com o Apache Kafka.

Primeiro, vamos criar um arquivo chamado requirements.txt e adicionar a seguinte dependência dentro dele:

kafka-python

Segundo, vamos criar um ambiente virtual e instalar as dependências no arquivo requirements.txt:

$ python -m venv venv
$ venv\scripts\activate
$ pip install -r requirements.txt

Feito! Agora sim nosso ambiente já está pronto para o desenvolvimento 🚀.

Desenvolvimento do produtor

Agora vamos criar nosso produtor. Como foi dito, o produtor será responsável por enviar os dados do tráfego aéreo para o tópico recém criado do Kafka.

Como também foi dito na arquitetura, a SkyX realiza voos apenas entre cinco cidades ao redor do mundo, e tem apenas cinco aeronaves disponíveis 😹. Vale ressaltar que cada aeronave transporta entre 50 e 100 pessoas.

Observe que os dados são gerados de forma aleatória e enviados ao tópico no formato json em um intervalo de tempo entre 1 e 6 segundos 😉.

Vamos lá! Crie um subdiretório chamado src e outro subdiretório chamado kafka. Dentro do diretório kafka, crie um arquivo chamado airtraffic_producer.py e adicione o seguinte código dentro dele:

import random
from json import dumps
from time import sleep
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers="localhost:29092",
    value_serializer=lambda x: dumps(x).encode("utf-8")
)

while True:
    cities = [
        "São Paulo, Brazil",
        "Tokyo, Japan",
        "Berlin, Germany",
        "Rome, Italy",
        "Seoul, South Korea"
    ]

    aircraft_names = [
        "Convair B-36 Peacemaker",
        "Lockheed C-5 Galaxy",
        "Northrop B-2 Spirit",
        "Boeing B-52 Stratofortress",
        "McDonnell XF-85 Goblin"
    ]

    aircraft = {
        "aircraft_name": random.choice(aircraft_names),
        "from": random.choice(cities),
        "to": random.choice(cities),
        "passengers": random.randint(50, 101)
    }

    future = producer.send("airtraffic", value=aircraft)
    print(future.get(timeout=60))

    sleep(random.randint(1, 6))

Feito! Desenvolvemos nosso produtor. Execute-o e deixe rodando por um tempo.

$ python airtraffic_producer.py

Desenvolvimento do consumidor

Agora vamos desenvolver nosso consumidor. Essa será uma aplicação bem simples. Ela irá apenas exibir no terminal em tempo real os dados que chegam no tópico do kafka.

Ainda dentro do diretório kafka, crie um arquivo chamado airtraffic_consumer.py e adicione o seguinte código dentro dele:

from json import loads
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "airtraffic",
    bootstrap_servers="localhost:29092",
    value_deserializer=lambda x: loads(x.decode("utf-8"))
)

for msg in consumer:
    print(msg.value)

Viu só, eu te disse que era bem simples. Execute-o e observe os dados que serão exibidos em tempo real conforme o produtor envia os dados ao tópico.

$ python airtraffic_consumer.py

Análise de dados: cidades que mais recebem turistas

Agora começamos com nossa análise de dados. Nesse momento, vamos desenvolver uma dashboard, uma aplicação, que irá exibir em tempo real um rank das cidades que mais recebem turistas. Ou seja, iremos agrupar os dados pela coluna to e fazer uma somatória com base na coluna passengers. Bem simples!

Para isso, dentro do diretório src, crie um subdiretório chamado dashboards e crie um arquivo chamado tourists_analysis.py. Em seguida, adicione o seguinte código dentro dele:

import json
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("Tourists Analysis")
         .getOrCreate()
         )

df1 = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "localhost:29092")
       .option("subscribe", "airtraffic")
       .option("startingOffsets", "earliest")
       .load()
       )

df2 = df1.selectExpr("CAST(value AS STRING)")

aircraft = {
    "aircraft_name": "",
    "from": "",
    "to": "",
    "passengers": 0
}

schema = F.schema_of_json(F.lit(json.dumps(aircraft)))

airtraffic = (df2.select(F.from_json(df2.value, schema).alias("jsondata"))
              .select("jsondata.*")
              )

tourists = (airtraffic.groupBy("to")
            .agg({"passengers": "sum"})
            .withColumnRenamed("sum(passengers)", "tourists")
            .withColumnRenamed("to", "city")
            .orderBy("tourists", ascending=False)
            )

(tourists.writeStream
 .format("console")
 .outputMode("complete")
 .start()
 .awaitTermination()
 )

E já podemos executar nosso arquivo através do spark-submit. Mas calma lá! Quando estamos integrando o PySpark com o Kafka, devemos executar o spark-submit de modo diferente. É necessário que informemos o pacote do Apache Kafka e a versão atual do Apache Spark através do parâmetro --packages.

Caso seja a primeira vez que esteja integrando o Apache Spark com o Apache Kafka, talvez a execução do spark-submit demore um pouco. Isso ocorre porque ele precisa fazer o download dos pacotes necessários.

Certifique-se que o produtor ainda esteja rodando para que possamos ver a análise dos dados em tempo real. Dentro do diretório dashboards, execute o seguinte comando:

$ spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 tourists_analysis.py

+------------------+--------+
|              city|tourists|
+------------------+--------+
|       Rome, Italy|    2628|
|      Tokyo, Japan|    2467|
|   Berlin, Germany|    2204|
|Seoul, South Korea|    1823|
| São Paulo, Brazil|    1719|
+------------------+--------+

Análise de dados: cidades onde mais saem pessoas

Essa análise é bem semelhante a anterior. Porém, ao invés de analisarmos em tempo real as cidades que mais recebem turistas, vamos analisar as cidades onde mais saem pessoas. Para isso, crie um arquivo chamado leavers_analysis.py e adicione o seguinte código dentro dele:

import json
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("Leavers Analysis")
         .getOrCreate()
         )

df1 = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "localhost:29092")
       .option("subscribe", "airtraffic")
       .option("startingOffsets", "earliest")
       .load()
       )

df2 = df1.selectExpr("CAST(value AS STRING)")

aircraft = {
    "aircraft_name": "",
    "from": "",
    "to": "",
    "passengers": 0
}

schema = F.schema_of_json(F.lit(json.dumps(aircraft)))

airtraffic = (df2.select(F.from_json(df2.value, schema).alias("jsondata"))
              .select("jsondata.*")
              )

leavers = (airtraffic.groupBy("from")
           .agg({"passengers": "sum"})
           .withColumnRenamed("sum(passengers)", "leavers")
           .withColumnRenamed("from", "city")
           .orderBy("leavers", ascending=False)
           )

(leavers.writeStream
 .format("console")
 .outputMode("complete")
 .start()
 .awaitTermination()
 )

Certifique-se que o produtor ainda esteja rodando para que possamos ver a análise dos dados em tempo real. Dentro do diretório dashboards, execute o seguinte comando:

$ spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 leavers_analysis.py

+------------------+-------+
|              city|leavers|
+------------------+-------+
|      Tokyo, Japan|   2673|
|   Berlin, Germany|   2305|
| São Paulo, Brazil|   2096|
|Seoul, South Korea|   1895|
|       Rome, Italy|   1872|
+------------------+-------+

Análise de dados: aeronaves que mais transportam passageiros

Essa análise é bem mais simples do que as anteriores. Vamos analisar em tempo real as aeronaves que mais transportam passageiros entre as cidades ao redor do mundo. Crie um arquivo chamado aircrafts_analysis.py e adicione o seguinte código dentro dele:

import json
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("Aircrafts Analysis")
         .getOrCreate()
         )

df1 = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "localhost:29092")
       .option("subscribe", "airtraffic")
       .option("startingOffsets", "earliest")
       .load()
       )

df2 = df1.selectExpr("CAST(value AS STRING)")

aircraft = {
    "aircraft_name": "",
    "from": "",
    "to": "",
    "passengers": 0
}

schema = F.schema_of_json(F.lit(json.dumps(aircraft)))

airtraffic = (df2.select(F.from_json(df2.value, schema).alias("jsondata"))
              .select("jsondata.*")
              )

aircrafts = (airtraffic.groupBy("aircraft_name")
             .agg({"passengers": "sum"})
             .withColumnRenamed("sum(passengers)", "total_passengers")
             .orderBy("total_passengers", ascending=False)
             )

(aircrafts.writeStream
 .format("console")
 .outputMode("complete")
 .start()
 .awaitTermination()
 )

Certifique-se que o produtor ainda esteja rodando para que possamos ver a análise dos dados em tempo real. Dentro do diretório dashboards, execute o seguinte comando:

$ spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 aircrafts_analysis.py

+--------------------+----------------+
|       aircraft_name|total_passengers|
+--------------------+----------------+
|McDonnell XF-85 G...|            2533|
|Boeing B-52 Strat...|            2345|
|Convair B-36 Peac...|            2012|
| Lockheed C-5 Galaxy|            2002|
| Northrop B-2 Spirit|            1949|
+--------------------+----------------+

Considerações finais

E finalizamos por aqui, pessoal! Neste artigo desenvolvemos uma análise de dados em tempo real com base em dados fictícios de um tráfego aéreo utilizando o Spark Structured Streaming e o Apache Kafka.

Para isso, desenvolvemos um produtor que envia esses dados em tempo real ao tópico do kafka, e depois desenvolvemos 3 dashboards para analisar esses dados em tempo real.

Espero que tenham gostado. Até a próxima 💚.

Data Engineering with Scala: Mastering Real-Time Data Processing with Apache Flink and Google Pub/Sub

Geazi Anc — Fri, 18 Oct 2024 00:28:07 +0000

Note: this article is also available in brazilian portuguese 🌎

Apache Flink is a distributed data processing framework for both batch and streaming processing. It can be used to develop event-driven applications; perform batch and streaming data analysis; and can be used to develop ETL data pipelines.

Pub/Sub is a scalable, asynchronous messaging service from Google that separates the services that produce messages from the services that process them. It is used for streaming analytics and data integration pipelines to load and distribute data, and is equally effective as a messaging middleware for service integration or as a queue to load tasks in parallel.

In this article, we will develop a very simple real-time data pipeline using Apache Flink in conjunction with version 3 of the Scala programming language, using Pub/Sub as a message broker. Before we begin, let's align expectations?

First, this article is not intended to be an introductory article to Apache Flink. If you have never heard of it before, I suggest you read the first steps from the official documentation. Read it without fear! The Apache Flink documentation is excellent!

Second, although Apache Flink has an official API for the Scala language, it has been deprecated and will be removed in future versions. You can learn more about this here. However, since Scala is a JVM-based language and Apache Flink is developed in Java, it is perfectly possible to still use the Scala language for development with Apache Flink, but using the Java APIs. Yes, I also turned my nose up at that. Nobody deserves that! But, to make our lives easier, we will use the Flink Scala API library, which is nothing less than a fork of the official Flink Scala API, but completely maintained by the community. I highly recommend this library!

Third, finally, we will develop a very simple real-time data pipeline. The goal is not to provide a complex example, but rather to provide a guide to working with Apache Flink with the Scala language plus Pub/Sub as a message broker. I had a hard time finding a decent article that used these three technologies together.

What will we see in this article?

Data Engineering with Scala: Mastering Real-Time Data Processing with Apache Flink and Google Pub/Sub
- 1. Problem definition
- 2. Setup
- 2.1. Creating topics and subscriptions in Pub/Sub
- 2.2. Installing dependencies
- 3. Data pipeline development
- 3.1. Business models and requirements
- 3.2. Defining serializers and deserializers
- 3.3. Pipeline arguments
- 3.4. Pub/Sub source
- 3.5. Pub/Sub Sink
- 3.6. Data pipeline and application of business requirements
- 4. Running the data pipeline
- 5. Conclusion

Now, enough talk. Let's get started!

1. Problem definition

A web application is responsible for receiving the initial registration of new customers from a large Brazilian retail company called My Awesome Company, hereinafter MAC, mac.br. The application sends the initial registration of new customers in real time to a Pub/Sub topic, and you must develop a data pipeline that processes this data in real time, enriches the initial customer registration with some relevant business information and, finally, sends it to a final topic in Pub/Sub. Pretty simple, right?

The web application sends the following payload to Pub/Sub:

{
  "fullName": "string",
  "birthDate": "string"
}

Where:

fullName is the client's full name (dann!);
birthDate is the customer's date of birth, in the format _year-month-day*;

The data pipeline must enrich this basic customer registration with some relevant business information:

It is necessary to split the client's full name into first name and last name;
The client's current age must be calculated based on their date of birth;
If the customer is over 30 years old, registration should not be carried out and the customer should be listed as inactive;
Add a createdAt field, related to the customer creation date.

With this understanding, let's start coding!

2. Setup

Hold on! Let's not start coding yet 🙍🏼 . We'll need to configure a few things first. The initial configurations we'll have to do are the following:

Creation of topics and subscriptions in Pub/Sub;
Installation of the dependencies required for the data pipeline to work;

2.1. Creating topics and subscriptions in Pub/Sub

To create topics and subscriptions in Pub/Sub, we will be using the official Google Cloud CLI, gcloud. Follow these instructions if you do not already have the CLI properly configured on your machine.

Now, what topics need to be created?

created-customer: the topic where the MAC web application will send the payloads relating to the initial customer registrations;
registered-customer: the final topic where our data pipeline will send customers with their respective registrations duly enriched;

Let's start with the created-customer topic. For this topic, we also need to create a standard subscription of type pull:

# creating the topic created-customer
$ gcloud pubsub topics create created-customer
Created topic [projects/my-project-id/topics/created-customer].

# now, creating a pull subscription to the topic created-customer
$ gcloud pubsub subscriptions create created-customer-sub --topic=created-customer
Created subscription [projects/my-project-id/subscriptions/created-customer-sub].

Now, let's create the registered-customer topic. For this topic, we also need to create a default subscription of type pull:

# creating the registered-customer topic
$ gcloud pubsub topics create registered-customer
Created topic [projects/my-project-id/topics/registered-customer].

# now, creating a pull subscription to the registered-customer topic
$ gcloud pubsub subscriptions create registered-customer-sub --topic=registered-customer
Created subscription [projects/my-project-id/subscriptions/registered-customer-sub].

2.2. Installing dependencies

Now yes! Time to code! 🎉

First of all, the development of our data pipeline will not be based on SBT projects. We will use the Scala CLI, a command-line tool that allows compile, run, test and package Scala code. Based on the Scala CLI, we can develop Scala scripts in a very practical and fast way!

To install dependencies, we will use a Scala CLI feature called directives. Directives are ways of defining configurations within the own source code, without needing a build tool like SBT for this. One of the directives we will use is to define the dependencies that our pipeline will use, namely:

Apache Flink Client (Apache Flink's own dependency);
Flink Scala API (a community-maintained library that allows us to develop code in Apache Flink with Scala APIs);
Flink Connector GCP PubSub: the official Apache Flink connector that allows us to send and receive Pub/Sub messages;
Toolkit: a set of useful libraries for everyday tasks, including the uPickle library, used to serialize and deserialize JSON;

To begin, create a directory called br-mac, and a file called Customers.sc inside it:

$ mkdir br-mac
...
$ cd br-mac
...
$ touch Customers.sc
...

Now, inside the Customers.sc file, add the following lines that are related to the directives for installing the necessary dependencies:

//> using toolkit default
//> using dep "org.flinkextended::flink-scala-api:1.18.1_1.1.6"
//> using dep "org.apache.flink:flink-clients:1.18.1"
//> using dep org.apache.flink:flink-connector-gcp-pubsub:3.1.0-1.18

And add the imports that will be used later:

import br.mac.customers.models.*
import br.mac.customers.serializations.*
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.connectors.gcp.pubsub.{PubSubSink, PubSubSource}
import org.apache.flinkx.api.*
import org.apache.flinkx.api.serializers.*

Done! Dependencies and imports have been defined. Let's move on.

3. Data pipeline development

Now it's time to build the data pipeline itself with Apache Flink! This build will consist of six parts:

Business models and requirements;
Development of serializers and deserializers;
Using ParameterTool so that we can get some relevant information for our pipeline through command line arguments;
Development of PubSubSource so that Apache Flink can read data from the Pub/Sub created-customer topic;
Development of PubSubSink so that Apache Flink can send the processed data to the registered-customer topic in Pub/Sub;
Development of the data pipeline core applying business requirements;

Let's go?

3.1. Business models and requirements

Business models are the information that we will receive and send to Pub/Sub. As mentioned before, we will receive a payload in JSON format from Pub/Sub, and send a payload to Pub/Sub also in JSON format. We need to model this payload in Scala classes.

Since these classes are representations of JSON payloads, we will use the uPickle library to serialize them.
and deserialize them into JSON format. If you are not familiar with the uPickle library, I highly recommend you read the documentation. It is also an excellent library!

An example of a payload that we will receive, related to the initial customer registration, is the following:

{
  "fullName": "John Doe",
  "birthDate": "1995-01-01"
}

An example of a payload that we will send to Pub/Sub, related to the final customer registration, is the following:

{
  "firstName": "John",
  "lastName": "Doe",
  "age": 29,
  "isActive": true,
  "createdAt": "2024-08-08T18:07:44.167635713Z"
}

Create another file called Models.scala. Note that this time the file extension is .scala, not .sc. This is because this file is a Scala module, not a Scala script.

In the file, add the following lines:

package br.mac.customers.models

import upickle.default.*

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter
final case class RegisteredCustomer(firstName: String, lastName: String, age: Int, isActive: Boolean, createdAt: String)
    derives ReadWriter

Done! However, we are not finished with our models yet. We need to define some methods so that we can satisfy the business requirements that were defined, which are:

It is necessary to split the client's full name into first name and last name;
The customers's current age must be calculated based on their date of birth;
If the customer is over 30 years old, registration should not be carried out and the customer should be listed as inactive;
Add a createdAt field, related to the customer creation date.

The first and second business requirements can be defined as methods in the CreatedCustomer class. For the third, we can define a constructor for the RegisteredCustomer class that creates an instance of the class with the isActive attribute set to true and the createdAt attribute set to the current time. The fourth requirement will be addressed in the data pipeline itself.

For the first and second requirement, we need to make some imports in the Models.scala file:

import java.time.temporal.ChronoUnit
import java.time.{Instant, LocalDate}

And we can now define the methods in the CreatedCustomer class:

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter:
def firstName: String = fullName.split(" ").head
def lastName: String = fullName.split(" ").last
def age: Int = ChronoUnit.YEARS.between(LocalDate.parse(birthDate), LocalDate.now()).toInt

Finally, let's declare the constructor for the RegisteredCustomer class. We'll do this by defining the apply method on the companion object:

object RegisteredCustomer:
def apply(firstName: String, lastName: String, age: Int): RegisteredCustomer =
RegisteredCustomer(firstName, lastName, age, true, Instant.now().toString)

So the final code for the Models.scala file looks like this:

package br.mac.customers.models

import upickle.default.*

import java.time.temporal.ChronoUnit
import java.time.{Instant, LocalDate}

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter:
def firstName: String = fullName.split(" ").head
def lastName: String = fullName.split(" ").last
def age: Int = ChronoUnit.YEARS.between(LocalDate.parse(birthDate), LocalDate.now()).toInt

final case class RegisteredCustomer(firstName: String, lastName: String, age: Int, isActive: Boolean, createdAt: String)
derives ReadWriter

object RegisteredCustomer:
def apply(firstName: String, lastName: String, age: Int): RegisteredCustomer =
RegisteredCustomer(firstName, lastName, age, true, Instant.now().toString)

3.2. Defining serializers and deserializers

When we talk about Apache Flink connectors, as is the case with the Apache Flink connector for Pub/Sub, we need to keep in mind two fundamental concepts: serializers and deserializers. In other words, serializations.

Serializers are responsible for transforming primitive data types, both from Java and Scala, to be sent to the destination in binary format. Deserializers are responsible for transforming the data received from the source and transforming it into object instances. of the programming languages used.

In our case, we need to create a serializer that receives an instance of one of our newly created classes, transforms them into JSON strings, and transforms them into binary so that they can then be sent to Pub/Sub. The process is exactly the opposite for deserializers. We need to transform a message, a JSON string, that Pub/Sub sends in binary format and transform this message into an instance of the newly created classes.

It's a relatively simple process. To deserialize the JSON string into an instance of the case class, we'll use uPickle. If you're already familiar with Flink, you might be wondering why we don't do this process with the flink-json library. Simple, I had a lot of problems using it to deserialize the JSON strings into the case classes. Therefore, I found it more practical to create a custom deserializer that uses the uPickle library for this process.

Enough talk! Let's code!

Create another file in the directory called Serializations.scala and add the following lines inside it:

package br.mac.customers.serializations

import br.mac.customers.models.*
import org.apache.flink.api.common.serialization.{AbstractDeserializationSchema, SerializationSchema}
import upickle.default.{read, write}

Let's create the deserializer for the CreatedCustomer class. To do this, simply define a class that extends the AbstractDeserializationSchema abstract class, and define the deserialize method. For more information, see this documentation.

class CreatedCustomerDeserializer extends AbstractDeserializationSchema[CreatedCustomer]:
  override def deserialize(message: Array[Byte]): CreatedCustomer = read[CreatedCustomer](new String(message, "UTF-8"))

See? I told you it was simple!

Now let's define the serializer for the RegisteredCustomer class.

class RegisteredCustomerSerializer extends SerializationSchema[RegisteredCustomer]:
override def serialize(element: RegisteredCustomer): Array[Byte] =
write[RegisteredCustomer](element).getBytes("UTF-8")

The interesting thing about this approach is that we can use any library we want to serialize and deserialize JSON strings. If we were using the flink-json library, we would be stuck using Java's jackson library. Yes, I also got goosebumps just thinking about it!

The final code for the Serializations.scala file looks like this:

package br.mac.customers.serializations

import br.mac.customers.models.*
import org.apache.flink.api.common.serialization.{AbstractDeserializationSchema, SerializationSchema}
import upickle.default.{read, write}

class CreatedCustomerDeserializer extends AbstractDeserializationSchema[CreatedCustomer]:
override def deserialize(message: Array[Byte]): CreatedCustomer = read[CreatedCustomer](new String(message, "UTF-8"))

class RegisteredCustomerSerializer extends SerializationSchema[RegisteredCustomer]:
override def serialize(element: RegisteredCustomer): Array[Byte] =
write[RegisteredCustomer](element).getBytes("UTF-8")

We end here with serializers and deserializers. Let's continue!

3.3. Pipeline arguments

In order to make our pipeline as flexible as possible, we must have a way to receive some parameters that are relevant to the functioning of our application, without having to hard-code this information. This information is:

Google Cloud Platform project ID;
Name of the Pub/Sub subscription from which Apache Fllink will consume data;
Name of the Pub/Sub topic where Apache Flink will send the processed data;

To do this, we will receive this information through command line arguments. To do this, we will use a built-in Apache Flink utility called ParameterTool. You can learn more about using this utility in this documentation.

Let's get to work! Add the following lines to the Customers.sc file:

val parameters = ParameterTool.fromArgs(args)
val projectName = parameters.get("project")
val subscriptionName = parameters.get("subscription-name")
val topicName = parameters.get("topic-name")

Done! With this, we can pass the project ID, subscription name and topic name to our pipeline through the --project, --subscription-name and --topic-name parameters, respectively.

3.4. Pub/Sub source

The Pub/Sub source, as mentioned, is the way Apache Flink will read data from Pub/Sub. We will build this source using the official Apache Flink connector for Pub/Sub. If you are interested in learning more about this connector, check out this documentation.

The Pub/Sub source constructor requires the following information:

Deserializer: the way Apache Flink will transform the message received from Pub/Sub into Scala language objects. Remember the deserializer for the CreatedCustomer class that we developed above? So, that's what we'll be using;
ProjectName: The name of the GCP project where you created the Pub/Sub topics and subscriptions;
SubscriptionName: the name of the subscription from which Apache Flink will consume data related to the initial registration of customers;

Add the following lines to the file:

val pubsubSource = PubSubSource
.newBuilder()
.withDeserializationSchema(new CreatedCustomerDeserializer())
.withProjectName(projectName)
.withSubscriptionName(subscriptionName)
.build()

And that's it! Pretty simple too, right?

3.5. Pub/Sub Sink

Phew, we're almost done. Let's build the PubSub Sink for our pipeline.

As stated, Pub/Sub Sink is a way for Apache Flink to send processed data to Pub/Sub. The Pub/Sub Sink constructor requires the following information:

Serializer: the way Apache Flink will transform the RegisteredCustomer class instance into a JSON string and then into binary and send it to Pub/Sub. Remember the serializer we created earlier? That's the one we're going to use!
ProjectName: The name of the GCP project where you created the Pub/Sub topics and subscriptions;
TopicName: the name of the topic that Apache Fllink will send the processed data to;

Add the following lines to the file:

val pubsubSink = PubSubSink
.newBuilder()
.withSerializationSchema(new RegisteredCustomerSerializer())
.withProjectName(projectName)
.withTopicName(topicName)
.build()

3.6. Data pipeline and application of business requirements

We have finally reached the last stage of development! Let's build the core of our data pipeline! As a reminder, our data pipeline will:

Read the initial customer registrations from the Pub/Sub created-customer topic;
Apply transformations and rules according to business requirements, such as:
Split the customer's name into first and last name;
Calculate the customer's age based on their date of birth;
Set the client creation date;
If the customer's age is greater than or equal to 30 years, do not register the customer and set the isActive status to false;
Send the processed data to the registered-customer topic in Pub/Sub;

Let's go! Let's get to work!

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(1000L)

env
.addSource(pubsubSource) // reading data from the created-customer topic
.map(cc => RegisteredCustomer(cc.firstName, cc.lastName, cc.age)) // splitting the customer's name into first and last name, calculating the age and setting the creation date
.map(rc => if rc.age >= 30 then rc.copy(isActive = false) else rc) // checking if the client's age is greater than or equal to 30
.addSink(pubsubSink) // sending the processed data to the registered-customer topic

env.execute("customerRegistering")

Is it finished? Yes, it is finished! Here is what the complete code looks like:

//> using toolkit default
//> using dep "org.flinkextended::flink-scala-api:1.18.1_1.1.6"
//> using dep "org.apache.flink:flink-clients:1.18.1"
//> using dep org.apache.flink:flink-connector-gcp-pubsub:3.1.0-1.18

import br.mac.customers.models.*
import br.mac.customers.serializations.*
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.connectors.gcp.pubsub.{PubSubSink, PubSubSource}
import org.apache.flinkx.api.*
import org.apache.flinkx.api.serializers.*

val parameters = ParameterTool.fromArgs(args)
val projectName = parameters.get("project")
val subscriptionName = parameters.get("subscription-name")
val topicName = parameters.get("topic-name")

val pubsubSource = PubSubSource
.newBuilder()
.withDeserializationSchema(new CreatedCustomerDeserializer())
.withProjectName(projectName)
.withSubscriptionName(subscriptionName)
.build()
val pubsubSink = PubSubSink
.newBuilder()
.withSerializationSchema(new RegisteredCustomerSerializer())
.withProjectName(projectName)
.withTopicName(topicName)
.build()

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(1000L)

env
.addSource(pubsubSource)
.map(cc => RegisteredCustomer(cc.firstName, cc.lastName, cc.age))
.map(rc => if rc.age >= 30 then rc.copy(isActive = false) else rc)
.addSink(pubsubSink)

env.execute("customerRegistering")

4. Running the data pipeline

Before running the pipeline, access Pub/Sub through your browser console, access the created-customer topic, and manually send some messages according to the CreatedCustomer payload schema. For example:

{
  "fullName": "John Doe",
  "birthDate": "1995-01-01"
}

Let's see all this in action. To do so, run the data pipeline through the Scala CLI. There is no need to package the data pipeline and upload it to a Flink cluster. We are working here in local mode.

Run the data pipeline with the following command. Note the application parameters as we defined previously:

$ scala-cli . -- \
--project your-project-id-here \
--subscription-name created-customer-sub \
--topic-name registered-customer
# ...
Compiling project (Scala 3.4.2, JVM (11))
Compiled project (Scala 3.4.2, JVM (11))
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.

Running! Open Pub/Sub in your browser, go to the registered-customer topic, and click Pull. This will show you the data that was processed by Apache Flink 🎉 .

Press CTRL + C to stop the pipeline execution.

5. Conclusion

And we've reached the end of the article! Today, we did:

We define the problem of the company My Awesome Company (MAC);
We define the JSON payloads that would be received and sent to the Pub/Sub topics;
We define the business requirements that would be applied to the received data;
We created two topics in Pub/Sub: one to receive the message regarding the initial registration of customers and another to send the data after being processed by Apache Flink;
We developed the data pipeline in Apache Flink, defining the business models for each payload received and sent; serializers and deserializers of JSON strings; and finally the data pipeline itself, applying the previously defined business rules;

That's all for today, guys! If you liked it, give me a little push and hit like and share it with your friends, okay?

See you next time 💚

Engenharia de Dados com Scala: masterizando o processamento de dados em tempo real com Apache Flink e Google Pub/Sub

Geazi Anc — Fri, 09 Aug 2024 00:23:27 +0000

Note: this article is also available in english 🌎

O Apache Flink é um framework de processamento de dados distribuído, tanto para processamento em batch quanto em streaming. Com ele, é possível desenvolver aplicações orientadas a eventos; realizar análise de dados em batch e em streaming; além de poder ser utilizado para o desenvolvimento de pipelines de dados ETL.

Já o Pub/Sub é um serviço de mensagens assíncrono e escalonável da Google que separa os serviços que produzem mensagens dos serviços que as processam. Ele é usado para análises de streaming e pipelines de integração de dados para carregar e distribuir dados, sendo igualmente eficaz como um middleware voltado para mensagens para integração de serviços ou como uma fila para carregar tarefas em paralelo.

Nesse artigo, iremos desenvolver uma pipeline de dados bem simples em tempo real utilizando o Apache Flink em conjunto com a versão 3 da linguagem de programação Scala, fazendo o uso do Pub/Sub como message broker. Antes de começarmos, vamos alinhar as espectativas?

Primeiro, esse artigo não tem a pretenção de ser um artigo introdutório ao Apache Flink. Caso você nunca tenha ouvido falar nele antes, sugiro a leitura do first steps da documentação oficial. Leia sem medo! A documentação do Apache Flink é excelente!

Segundo, apesar do Apache Flink ter uma API oficial para a linguagem Scala, ela foi descontinuada e será removida nas próximas versões. Você pode saber mais sobre isso aqui. Todavia, como o scala é uma linguagem baseada na JVM e o Apache Flink é desenvolvido em java, é perfeitamente possível ainda utilizarmos a linguagem scala para o desenvolvimento com o Apache Fllink, porém utilizando as APIs do Java. Sim, eu também torci o nariz para isso. Ninguém merece! Mas, para deixar nossa vida mais fácil, vamos utilizar a biblioteca Flink Scala API, que é nada menos que um fork da Scala API oficial do Flink, porém completamente mantido pela comunidade. Recomendo muito essa biblioteca!

Terceiro, por fim, iremos desenvolver aqui uma pipeline de dados em tempo real bem simples. O objetivo não é fornecer um exemplo complexo, mas sim fornecer um guia para trabalhar com o Apache Flink com a linguagem Scala mais o Pub/Sub como message broker. Tive muita dificuldade de encontrar um artigo decente que utilizasse essas três tecnologias em conjunto.

O que vamos ver nesse artigo?

1. Definição do problema
2. Setup
- 2.1. Criação dos tópicos e assinaturas no Pub/Sub
- 2.2. Instalação das dependências
3. Desenvolvimento da pipeline de dados
- 3.1. Modelos e requisitos de negócio
- 3.2. Definindo os serializers e deserializers
- 3.3. Argumentos da pipeline
- 3.4. Pub/Sub source
- 3.5. Pub/Sub Sink
- 3.6. Pipeline de dados e aplicação dos requisitos de negócio
4. Executando a pipeline de dados
5. Considerações finais

Agora, chega de papo. Vamos começar!

1. Definição do problema

Uma determinada aplicação web é responsável por receber o cadastro inicial de novos clientes da grande empresa de varejo brasileira chamada de My Awesome Company, doravante MAC, mac.br. A aplicação envia em tempo real o cadastro inicial dos novos clientes para um tópico do Pub/Sub, e você deve desenvolver uma pipeline de dados que processa esse dado em tempo real, enriqueça o cadastro inicial do cliente com algumas informações relevantes de negócio e, por fim, o envie para um tópico final no Pub/Sub. Bem simples, não?

A aplicação web envia o seguinte payload para o Pub/Sub:

{
  "fullName": "string",
  "birthDate": "string"
}

Onde:

fullName é o nome completo do cliente (dann!);
birthDate é a data de nascimento do cliente, no formato _ano-mes-dia*;

A pipeline de dados deve enriquecer esse cadastro básico do cliente com algumas informações relevantes de negócio:

É preciso separar o nome completo do cliente em primeiro nome e último nome;
Deve-se calcular a idade atual do cliente com base em sua data de nascimento;
Se o cliente tiver mais de 30 anos, o cadastro não deve ser realizado e o cliente deve constar como inativo;
Adicionar um campo createdAt, relacionado a data de criação do cliente.

Tendo esse entendimento, vamos começar a codar!

2. Setup

Calma lá! Não vamos começar a codar ainda 🙍🏼. Vamos precisar configurar algumas coisas antes. As configurações iniciais que vamos ter que fazer são as seguintes:

Criação dos tópicos e das assinaturas no Pub/Sub;
Instalação das dependências necessárias para o funcionamento da pipeline de dados;

2.1. Criação dos tópicos e assinaturas no Pub/Sub

Para a criação dos tópicos e assinaturas no Pub/Sub, vamos estar utilizando a CLI oficial da Google Cloud, o gcloud. Siga essas instruções caso ainda não tenha a CLI devidamente configurada em sua máquina.

Agora, quais tópicos precisam ser criados?

created-customer: o tópico onde a aplicação web da MAC irá enviar os payloads referentes aos cadastros iniciais dos clientes;
registered-customer: o tópico final onde nossa pipeline de dados irá enviar os clientes com os respectivos cadastros devidamente enriquecidos;

Vamos começar pelo tópico created-customer. Para esse tópico, também precisamos criar uma assinatura padrão do tipo pull:

# criando o tópico created-customer
$ gcloud pubsub topics create created-customer
Created topic [projects/my-project-id/topics/created-customer].

# agora, criando uma assinatura do tipo pull para o tópico created-customer
$ gcloud pubsub subscriptions create created-customer-sub --topic=created-customer
Created subscription [projects/my-project-id/subscriptions/created-customer-sub].

Agora, vamos criar o tópico registered-customer. Para esse tópico, também precisamos criar uma assinatura padrão do tipo pull:

# criando o tópico registered-customer
$ gcloud pubsub topics create registered-customer
Created topic [projects/my-project-id/topics/registered-customer].

# agora, criando uma assinatura do tipo pull para o tópico registered-customer
$ gcloud pubsub subscriptions create registered-customer-sub --topic=registered-customer
Created subscription [projects/my-project-id/subscriptions/registered-customer-sub].

2.2. Instalação das dependências

Agora sim! Hora de codar! 🎉

Antes de tudo, o desenvolvimento de nossa pipeline de dados não será feito com base em projetos SBT. Vamos utilizar a Scala CLI, uma ferramenta de linha de comando que permite compilar, executar, testar e empacotar códigos Scala. Com base no Scala CLI, podemos desenvolver scripts Scala de forma muito prática e rápida!

Para a instalação das dependências, vamos utilizar um recurso do Scala CLI chamado de diretivas. Diretivas são modos de definirmos configurações dentro do próprio código fonte, sem precisar de uma ferramenta de build como o SBT para tal. Uma das diretivas que vamos utilizar é para definirmos as dependências que nossa pipeline irá utilizar, a saber:

Apache Flink Client (a própria dependência do Apache Flink);
Flink Scala API (uma biblioteca mantida pela comunidade que nos permite desenvolver códigos no Apache Flink com as APIs do Scala);
Flink Connector GCP PubSub: o connector oficial do Apache Flink que nos permite enviar e receber mensagens do Pub/Sub;
Toolkit: um conjunto de bibliotecas úteis para tarefas cotidianas, incluindo a biblioteca uPickle, utilizada para serializar e deserializar JSON;

Para começarmos, crie um diretório chamado br-mac, e um arquivo chamado Customers.sc dentro dele:

$ mkdir br-mac
...
$ cd br-mac
...
$ touch Customers.sc
...

Agora, dentro do arquivo Customers.sc, adicione as seguintes linhas que são relacionadas as diretivas para a instalação das dependências necessárias:

//> using toolkit default
//> using dep "org.flinkextended::flink-scala-api:1.18.1_1.1.6"
//> using dep "org.apache.flink:flink-clients:1.18.1"
//> using dep org.apache.flink:flink-connector-gcp-pubsub:3.1.0-1.18

E já adicione os imports que serão utilizados posteriormente:

import br.mac.customers.models.*
import br.mac.customers.serializations.*
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.connectors.gcp.pubsub.{PubSubSink, PubSubSource}
import org.apache.flinkx.api.*
import org.apache.flinkx.api.serializers.*

Feito! As dependências e os imports foram definidos. Vamos em frente.

3. Desenvolvimento da pipeline de dados

Chegou o momento de desenvolvermos a pipeline de dados em si com o Apache Flink! Esse desenvolvimento irá consistir em seis partes:

Modelos e requisitos de negócio;
Desenvolvimento dos serializers e deserializers;
Utilização do ParameterTool para que possamos pegar algumas informações relevantes para nossa pipeline através de argumentos da linha de comando;
Desenvolvimento do PubSubSource para que o Apache Flink possa ler os dados do tópico created-customer do Pub/Sub;
Desenvolvimento do PubSubSink para que o Apache Flink possa enviar os dados processados para o tópico registered-customer no Pub/Sub;
Desenvolvimento do core da pipeline de dados aplicando os requisitos de negócio;

Vam'bora?

3.1. Modelos e requisitos de negócio

Os modelos de negócio são as informações que iremos receber e enviar para o Pub/Sub. Como dito anteriormente, eremos receber do Pub/Sub um payload no formato JSON, e enviar um payload para o Pub/Sub também no formato JSON. Precisamos modelar esse payload em classes em Scala.

Como essas classes são representações de payloads JSON, vamos utilizar a biblioteca uPickle para que seja possível serializá-las
e deserializá-las no formato JSON. Caso ainda não conheça a biblioteca uPickle, recomendo fortemente que dê uma lida na documentação. Também é uma excelente biblioteca!

Um exemplo de payload que iremos receber, relacionado ao cadastro inicial dos clientes, é o seguinte:

{
  "fullName": "John Doe",
  "birthDate": "1995-01-01"
}

Já um exemplo de payload que iremos enviar para o Pub/Sub, relacionado ao cadastro final do cliente, é o seguinte:

{
  "firstName": "John",
  "lastName": "Doe",
  "age": 29,
  "isActive": true,
  "createdAt": "2024-08-08T18:07:44.167635713Z"
}

Crie um outro arquivo chamado Models.scala. Observe que dessa vez a extensão do arquivo é .scala, e não .sc. Isso porque esse arquivo é um módulo Scala, e não um script Scala.

No arquivo, adicione as seguintes linhas:

package br.mac.customers.models

import upickle.default.*

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter
final case class RegisteredCustomer(firstName: String, lastName: String, age: Int, isActive: Boolean, createdAt: String)
    derives ReadWriter

Feito! Porém, ainda não acabamos com nossos modelos. Precisamos definir alguns métodos para que possamos satisfazer os requisitos de negócio que foram definidos, sendo eles:

É preciso separar o nome completo do cliente em primeiro nome e último nome;
Deve-se calcular a idade atual do cliente com base em sua data de nascimento;
Se o cliente tiver mais de 30 anos, o cadastro não deve ser realizado e o cliente deve constar como inativo;
Adicionar um campo createdAt, relacionado a data de criação do cliente.

O primeiro e o segundo requisito de negócio podemos definir como métodos na classe CreatedCustomer. Já o terceiro, podemos definir um construtor para a classe RegisteredCustomer que cria uma instância da classe com o atributo isActive definido como true e o atributo createdAt definido como o horário atual. O quarto requisito iremos trabalhar na própria pipeline de dados.

Para o primeiro e segundo requisito, precisamos fazer algumas importações no arquivo Models.scala:

import java.time.temporal.ChronoUnit
import java.time.{Instant, LocalDate}

E já podemos definir os métodos na classe CreatedCustomer:

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter:
  def firstName: String = fullName.split(" ").head
  def lastName: String  = fullName.split(" ").last
  def age: Int          = ChronoUnit.YEARS.between(LocalDate.parse(birthDate), LocalDate.now()).toInt

Por fim, vamos declarar o construtor para a classe RegisteredCustomer. Vamos fazer isso definindo o método apply no companion object:

object RegisteredCustomer:
  def apply(firstName: String, lastName: String, age: Int): RegisteredCustomer =
    RegisteredCustomer(firstName, lastName, age, true, Instant.now().toString)

Portanto, o código final para o arquivo Models.scala ficou dessa maneira:

package br.mac.customers.models

import upickle.default.*

import java.time.temporal.ChronoUnit
import java.time.{Instant, LocalDate}

final case class CreatedCustomer(fullName: String, birthDate: String) derives ReadWriter:
  def firstName: String = fullName.split(" ").head
  def lastName: String  = fullName.split(" ").last
  def age: Int          = ChronoUnit.YEARS.between(LocalDate.parse(birthDate), LocalDate.now()).toInt

final case class RegisteredCustomer(firstName: String, lastName: String, age: Int, isActive: Boolean, createdAt: String)
    derives ReadWriter

object RegisteredCustomer:
  def apply(firstName: String, lastName: String, age: Int): RegisteredCustomer =
    RegisteredCustomer(firstName, lastName, age, true, Instant.now().toString)

3.2. Definindo os serializers e deserializers

Quando falamos em connectores do Apache Flink, como é o caso do connector do Apache Flink para o Pub/Sub, precisamos ter em mente dois conceitos fundamentais: serializers e deserializers. Em outras palavras, serializations.

Os serializers são responsáveis por transformar os tipos de dados primitivos, tanto da linguagem Java quanto do Scala, para serem enviados para o destino no formato binário. Já os deserializers são responsáveis por transformar o dado recebido da fonte e transformá-los para instâncias de objetos das linguagens de programação utilizadas.

No nosso caso, é necessário criar um serializer que receba uma instância de uma das nossas classes recém criadas, transforme-as em strings JSON, as transforme para binário para que aí sim elas possam ser enviadas para o Pub/Sub. O processo é exatamente o oposto para os deserializers. Precisamos transformar uma mensagem, uma string JSON, que o Pub/Sub envia no formato binário e transformar essa mensagem em uma instância das classes recém criadas.

É um processo relativamente simples. Para deserializarmos a string JSON para uma instância da case class, vamos usar o uPickle. Se você já tiver familiaridade com o Flink, deve estar se perguntando porque não fazemos esse processo com a biblioteca flink-json. Simples, tive muitos problemas ao utilizá-la para deserializar as strings JSON para as case classes. Portanto, achei mais prático criar um deserializer customizado que utiliza a biblioteca uPickle para esse processo.

Chega de papo! Vamos codar!

Crie um outro arquivo no diretório chamado Serializations.scala e adicione as seguintes linhas dentro dele:

package br.mac.customers.serializations

import br.mac.customers.models.*
import org.apache.flink.api.common.serialization.{AbstractDeserializationSchema, SerializationSchema}
import upickle.default.{read, write}

Vamos criar o deserializer para a classe CreatedCustomer. Para isso, basta definir uma classe que extende a classe abstrata AbstractDeserializationSchema, e definirmos o método deserialize. Para mais informações, consulte essa documentação.

class CreatedCustomerDeserializer extends AbstractDeserializationSchema[CreatedCustomer]:
  override def deserialize(message: Array[Byte]): CreatedCustomer = read[CreatedCustomer](new String(message, "UTF-8"))

Viu só? Eu disse que era simples!

Agora vamos definir o serializer para a classe RegisteredCustomer.

class RegisteredCustomerSerializer extends SerializationSchema[RegisteredCustomer]:
  override def serialize(element: RegisteredCustomer): Array[Byte] =
    write[RegisteredCustomer](element).getBytes("UTF-8")

O interessante dessa abordagem é que podemos utilizar qualquer biblioteca que desejarmos para serializar e deserializar strings JSON. Se estivéssemos utilizando a biblioteca flink-json, estaríamos refém em utilizar a biblioteca jackson do Java. Sim, também senti arrepio só de pensar nisso!

O código final para o arquivo Serializations.scala ficou dessa forma:

package br.mac.customers.serializations

import br.mac.customers.models.*
import org.apache.flink.api.common.serialization.{AbstractDeserializationSchema, SerializationSchema}
import upickle.default.{read, write}

class CreatedCustomerDeserializer extends AbstractDeserializationSchema[CreatedCustomer]:
  override def deserialize(message: Array[Byte]): CreatedCustomer = read[CreatedCustomer](new String(message, "UTF-8"))

class RegisteredCustomerSerializer extends SerializationSchema[RegisteredCustomer]:
  override def serialize(element: RegisteredCustomer): Array[Byte] =
    write[RegisteredCustomer](element).getBytes("UTF-8")

Terminamos por aqui com os serializers e deserializers. Vamos continuar!

3.3. Argumentos da pipeline

Para que possamos deixar nossa pipeline o mais flexível possível, devemos ter um modo de recebermos alguns parâmetros que são relevantes para o funcionamento da nossa aplicação, sem a necessidade de deixarmos essas informações hard-coded. Essas informações são:

ID do projeto do Google Cloud Platform;
Nome da assinatura do Pub/Sub de onde o Apache Fllink irá consumir os dados;
Nome do tópico do Pub/Sub onde o Apache Flink irá enviar os dados processados;

Para isso, vamos receber essas informações através de argumentos da linha de comando. Para isso, vamos utilizar um utillitário built-in do Apache Flink chamado ParameterTool. Você pode aprender mais sobre a utilização desse utilitário nessa documentação.

Mãos a obra! Adicione as seguintes linhas no arquivo Customers.sc:

val parameters       = ParameterTool.fromArgs(args)
val projectName      = parameters.get("project")
val subscriptionName = parameters.get("subscription-name")
val topicName        = parameters.get("topic-name")

Feito! Com isso, podemos passar para nossa pipeline o ID do projeto, o nome da assinatura e o nome do tópico através dos parâmetros --project, --subscription-name e --topic-name, respectivamente.

3.4. Pub/Sub source

O Pub/Sub source, como dito, é o modo que o Apache Flink irá ler os dados do Pub/Sub. Vamos construir esse source através do connector oficial do Apache Flink para o Pub/Sub. Caso tenha interesse em saber mais sobre esse connector, verifique essa documentação.

O construtor do source do Pub/Sub requer as seguintes informações:

Deserializer: o modo que o Apache Flink irá transformar a mensagem recebida do Pub/Sub em objetos da linguagem Scala. Lembra do deserializer para a classe CreatedCustomer que desenvolvemos mais a cima? Então, é ela que vamos estar usando;
ProjectName: o nome do projeto do GCP onde você criou os tópicos e as assinaturas do Pub/Sub;
SubscriptionName: o nome da assinatura de onde o Apache Flink irá consumir os dados relacionados ao cadastro inicial dos clientes;

Adicione as seguintes linhas no arquivo:

val pubsubSource = PubSubSource
  .newBuilder()
  .withDeserializationSchema(new CreatedCustomerDeserializer())
  .withProjectName(projectName)
  .withSubscriptionName(subscriptionName)
  .build()

E é só isso! Bem simples também, né?

3.5. Pub/Sub Sink

Ufa, estamos quase acabando. Vamos construir o PubSub Sink da nossa pipeline.

Como dito, o Pub/Sub Sink é um modo do Apache Flink enviar os dados processados para o Pub/Sub. O construtor do Pub/Sub Sink requer as seguintes informações:

Serializer: o modo que o Apache Flink irá transformar a instância da classe RegisteredCustomer para uma string JSON e em seguida para binário e enviar para o Pub/Sub. Lembra do serializer que criamos anteriormente? É ele que vamos usar!
ProjectName: o nome do projeto do GCP onde você criou os tópicos e as assinaturas do Pub/Sub;
TopicName: o nome do tópico que o Apache Fllink irá enviar os dados processados;

Adicione as seguintes linhas no arquivo:

val pubsubSink   = PubSubSink
  .newBuilder()
  .withSerializationSchema(new RegisteredCustomerSerializer())
  .withProjectName(projectName)
  .withTopicName(topicName)
  .build()

3.6. Pipeline de dados e aplicação dos requisitos de negócio

Chegamos enfim na última etapa de desenvolvimento! Vamos construir o core da nossa pipeline de dados! Recordando, nossa pipeline de dados irá:

Ler os cadastros iniciais dos clientes do tópico created-customer do Pub/Sub;
Aplicar as transformações e regras conforme os requisitos de negócio, como:
- Separar o nome do cliente em primeiro e último nome;
- Calcular a idade do cliente com base na data de nascimento;
- Definir a data de criação do cliente;
- Se a idade do cliente for maior ou igual a 30 anos, não registrar o cliente e definir o status isActive como false;
Enviar os dados processados para o tópico registered-customer no Pub/Sub;

Vamos lá! Mão na massa!

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(1000L)

env
  .addSource(pubsubSource) // lendo os dados do tópico created-customer
  .map(cc => RegisteredCustomer(cc.firstName, cc.lastName, cc.age)) // separando o nome do cliente em primeiro e último nome, calculando a idade e definindo a data de criação
  .map(rc => if rc.age >= 30 then rc.copy(isActive = false) else rc) // verificando se a idade do cliente é maior ou igual a 30
  .addSink(pubsubSink) // enviando os dados processados para o tópico registered-customer

  env.execute("customerRegistering")

Acabou? Sim, acabou! Veja como ficou o código completo:

//> using toolkit default
//> using dep "org.flinkextended::flink-scala-api:1.18.1_1.1.6"
//> using dep "org.apache.flink:flink-clients:1.18.1"
//> using dep org.apache.flink:flink-connector-gcp-pubsub:3.1.0-1.18

import br.mac.customers.models.*
import br.mac.customers.serializations.*
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.connectors.gcp.pubsub.{PubSubSink, PubSubSource}
import org.apache.flinkx.api.*
import org.apache.flinkx.api.serializers.*

val parameters       = ParameterTool.fromArgs(args)
val projectName      = parameters.get("project")
val subscriptionName = parameters.get("subscription-name")
val topicName        = parameters.get("topic-name")

val pubsubSource = PubSubSource
  .newBuilder()
  .withDeserializationSchema(new CreatedCustomerDeserializer())
  .withProjectName(projectName)
  .withSubscriptionName(subscriptionName)
  .build()
val pubsubSink   = PubSubSink
  .newBuilder()
  .withSerializationSchema(new RegisteredCustomerSerializer())
  .withProjectName(projectName)
  .withTopicName(topicName)
  .build()

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(1000L)

env
  .addSource(pubsubSource)
  .map(cc => RegisteredCustomer(cc.firstName, cc.lastName, cc.age))
  .map(rc => if rc.age >= 30 then rc.copy(isActive = false) else rc)
  .addSink(pubsubSink)

env.execute("customerRegistering")

4. Executando a pipeline de dados

Antes de executar a pipeline, acesse o Pub/Sub através do console de seu navegador, acesse o tópico created-customer e envie manualmente algumas mensagens conforme o schema do payload CreatedCustomer. Por exemplo:

{
  "fullName": "John Doe",
  "birthDate": "1995-01-01"
}

Vamos ver tudo isso em ação? Para isso, execute a pipeline de dados através do Scala CLI. Não é necessário empacotar a pipeline de dados e subir em um cluster Flink. Estamos trabalhando aqui no modo local.

Execute a pipeline de dados com o seguinte comando. Observe os parâmetros da aplicação, conforme definimos anteriormente:

$ scala-cli . -- \
  --project seu-project-id-aqui \
  --subscription-name created-customer-sub \
  --topic-name registered-customer
# ...
Compiling project (Scala 3.4.2, JVM (11))
Compiled project (Scala 3.4.2, JVM (11))
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.

Execução em andamento! Abra o Pub/Sub em seu navegador, vá até o tópico registered-customer, e clique em Efetuar pull. Com isso você irá visualizar os dados que foram processados pelo Apache Flink 🎉.

Pressione CTRL + C para interromper a execução da pipeline.

5. Considerações finais

E chegamos ao fim do artigo! Hoje, fizemos:

Definimos o problema da empresa My Awesome Company (MAC);
Definimos os payloads JSON que seria recebido e enviado aos tópicos do Pub/Sub;
Definimos os requisitos de negócio que seriam aplicados aos dados recebidos;
Criamos dois tópicos no Pub/Sub: um para receber a mensagem referente ao cadastro inicial dos clientes e outro para o envio do dado após ser processado pelo Apache Flink;
Desenvolvemos a pipeline de dados no Apache Flink, definindo os modelos de negócio referente cada payload recebido e enviado; serializers e deserializers das strings JSON; e por fim a pipeline de dados em si, aplicando as regras de negócio anteriormente definidas;

Por hoje é só, "pe-pe-pessoal!" 🐷! Se você gostou, me dê uma forcinha e senta o dedo no like aí e compartilhe com seus amigos, combinado?

Até a próxima 💚

Integrando uma Web API com Datastore Emulator

Geazi Anc — Tue, 21 Feb 2023 15:12:12 +0000

O custo elevado do faturamento associado aos projetos do Google Cloud Platform (GCP) é algo que sempre devemos ter em mente durante todo o ciclo de desenvolvimento de um produto.

A fim de mitigar esse problema, uma das abordagens é o uso de emuladores que simulam alguns serviços localmente, acarretando em custo zero para o projeto.

Hoje, iremos ver como rodar o emulador oficial do Datastore localmente com Docker, e como integrá-lo com uma Web API!

Desenvolvimento da solução

Iremos desenvolver uma breve solução através do notebook para demonstrar o funcionamento do emulator do Datastore. Para tal, iremos desenvolver uma Web API bem simples que irá se integrar com o Datastore local.

Toda a solução será desenvolvida por meio de contêineres via Docker.

Web API

Desenvolvimento de uma API responsável pelo cadastro de usuários. Os dados serão persistidos localmente no contêiner do Datastore.

POST /users: salva um usuário no Datastore;
GET /users: recupera todos os usuários persistidos no Datastore;

Também será desenvolvido um arquivo Dockerfile para fazer a instalação das bibliotecas necessárias e subir o servidor de desenvolvimento da API.

$ mkdir src
$ touch src/app.py

import os

from fastapi import FastAPI
from google.cloud import datastore
from pydantic import BaseModel


PROJECT_ID = os.getenv("DATASTORE_PROJECT_ID")

client = datastore.Client(PROJECT_ID)
app = FastAPI()


class User(BaseModel):
    name: str
    age: int


@app.post("/users", status_code=201)
def create_user(user: User):
    kind = "users"

    key = client.key(kind)
    user_entity = client.entity(key)
    user_entity.update(user)

    client.put(user_entity)
    return user


@app.get("/users")
def get_users():
    query = client.query(kind="users")
    users = list(query.fetch())

    return users

$ touch requirements.txt
fastapi==0.91.0
uvicorn==0.20.0
google-cloud-datastore==2.13.2

$touch web-api.Dockerfile
FROM python:3.11-alpine


WORKDIR /app

COPY requirements.txt requirements.txt
RUN pip install --no-cache-dir --upgrade -r requirements.txt

COPY . .

CMD exec python -m uvicorn app:app --reload --app-dir=./src --host=0.0.0.0 --port=8000

Datastore

O desenvolvimento de um contêiner do Datastore consiste nos seguintes passos:

Desenvolvimento de uma imagem customizada tendo como base a imagem oficial dos emuladores do GCP;

- Iniciar o servidor web do Datastore;

$ touch datastore.Dockerfile
FROM gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators


WORKDIR /datastore

CMD exec gcloud beta emulators datastore start --project my-local-project --host-port 0.0.0.0

Desenvolvimento dos contêineres

Por fim, vamos desenvolver um arquivo docker-compose que irá orquestrar os contêineres construídos com base em nossas imagens.

Nota: observe que é necessário configurar as variáveis de ambiente do Datastore no contêiner da API. Isso se faz necessário para que o SDK do Datastore envie as solicitações diretamente para o contêiner local, não para os servidores do GCP.

$ touch docker-compose.yml
version: '3.9'

services:
  web-api:
    build:
      dockerfile: ./web-api.Dockerfile
    environment:
      DATASTORE_DATASET: my-local-dataset
      DATASTORE_EMULATOR_HOST: datastore:8081
      DATASTORE_EMULATOR_HOST_PATH: datastore:8081/datastore
      DATASTORE_HOST: http://datastore:8081
      DATASTORE_PROJECT_ID: my-local-project
    volumes:
      - ./:/app
    ports:
      - 8000:8000

  datastore:
    build:
      dockerfile: ./datastore.Dockerfile
    ports:
      - 8081:8081

$ docker compose up -d
$ docker compose ps

Vamos testar!

Iremos fazer três solicitações para nossa API. As duas primeiras serão solicitações POST, que irá salvar dois usuários no Datastore, e a última será uma solicitação GET que irá recuperar os dois usuários persistidos no banco de dados.

$ curl -X 'POST' \
  'http://localhost:8000/users' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "John",
  "age": 20
}'

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100    57  100    24  100    33     54     75 --:--:-- --:--:-- --:--:--   130
{"name":"John","age":20}

$ curl -X 'POST' \
  'http://localhost:8000/users' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "Mary",
  "age": 18
}'

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100    57  100    24  100    33   1411   1941 --:--:-- --:--:-- --:--:--  3352
{"name":"Mary","age":18}

$ curl -X 'GET' \
  'http://localhost:8000/users' \
  -H 'accept: application/json'

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100    51  100    51    0     0    850      0 --:--:-- --:--:-- --:--:--   850
[{"name":"John","age":20},{"name":"Mary","age":18}]

Considerações finais

O uso dos emuladores do Google Cloud Platform podem ser uma abordagem interessante durante o ciclo de desenvolvimento de um produto. Com eles, podemos testar nossas soluções quantas vezes forem necessárias sem acarretar em um custo elevado no faturamento do projeto, visto que todas as solicitações das bibliotecas do Google Cloud serão feitas localmente ao invés de serem feitas para os servidores da nuvem.

Obrigado por ter me acompanhado até aqui 💚. Até a próxima!

Referências

PySpark: A brief analysis to the most common words in Dracula, by Bram Stoker

Geazi Anc — Wed, 11 Jan 2023 17:06:56 +0000

Note: this article is also available in portuguese 🌎.

A landmark in Gothic literature, the iconic novel Dracula, written by Bram Stoker in 1897, stirs the emotions of people across the world. Today, to introduce Spark's new concepts and features, we will develop a brief notebook to analyze the most common words in this classic book 🧛🏼‍♂️.

To do this, we will write a notebook in Google Colab, a cloud service built by Google to encourage machine learning and artificial intelligence researches.

This notebook is also available in my GitHub 😉.

This novel was obtained through Project Gutenberg, a digital library that centralizes public books around the world.

Before get start

Before start, we need to install PySpark library.

The PySpark is the official API of Apache Spark for Python. We will develop our data analysis using it 🎲.

So, create a new code cell in Colab and add the following line:

!pip install pyspark

Step one: running Apache Spark

After the installation is complete, we need to run Apache Spark. To do this, create a new code cell and add the following code block:

         from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("The top most common words in Dracula, by Bram Stoker")
         .getOrCreate()
         )

Step two: downloading and reading

In this step, we will download the novel from Guttenberg project and, after that, load it using PySpark.

We will use wget tool to do this, passing the URL book for it and saving it in local directory, and renaming to Dracula – Bram Stoker.txt.

Again, create a new code cell in Colab and add the following code line:

!wget https: // www.gutenberg.org/cache/epub/345/pg345.txt -O "Dracula - Bram Stoker.txt"

Step three: stopwords downloading

In this section, we will download the list of stopwords used in English language. These stops words normally include prepositions, particles, interjections, unions, adverbs, pronouns, introductory words, numbers from 0 to 9 (unambiguous), other frequently used official, independent parts of speech, symbols, punctuation. Relatively recently, this list was supplemented by such commonly used on the Internet sequences of symbols as www, com, http, etc.

This list was obtained through CountWordsFree, a website that centralizes the stopwords used in many languages across the world.

get to work! Create a new code cell in Colab and add the following code line:

!wget https://countwordsfree.com/stopwords/english/txt -O "stop_words_english.txt"

After that, let’s load the book using Spark. Create a new code cell and add the following code block:

book = spark.read.text("Dracula - Bram Stoker.txt")

And let’s load the stopwords as well. The stopwords will are stored in a list, in stopwords variable.

with open("stop_words_english.txt", "r") as f:
    text = f.read()
    stopwords = text.splitlines()

len(stopwords), stopwords[:15]

Output

(851,
 ['able',
  'about',
  'above',
  'abroad',
  'according',
  'accordingly',
  'across',
  'actually',
  'adj',
  'after',
  'afterwards',
  'again',
  'against',
  'ago',
  'ahead']t)

Step four: extracting words

After load is completed, we need to extract the words to a dataframe column.

To do this, use the split function to each line, will split them using blank spaces between them. The result is a list of words.

from pyspark.sql.functions import split

lines = book.select(split(book.value, " ").alias("line"))
lines.show(5)

Output

+--------------------+
|                line|
+--------------------+
|[The, Project, Gu...|
|                  []|
|[This, eBook, is,...|
|[most, other, par...|
|[whatsoever., You...|
+--------------------+
only showing top 5 rows

Step five: exploding list words

Now, let’s convert this list of words in dataframe column, using explode function.

from pyspark.sql.functions import explode, col

words = lines.select(explode(col("line")).alias("word"))
words.show(15)

Output

+---------+
|     word|
+---------+
|      The|
|  Project|
|Gutenberg|
|    eBook|
|       of|
| Dracula,|
|       by|
|     Bram|
|   Stoker|
|         |
|     This|
|    eBook|
|       is|
|      for|
|      the|
+---------+
only showing top 15 rows

Step six: words to lowercase

This is a simple step. We don't want the same word to be different because of capital letters, so we convert these words to lowercase, using lower function.

from pyspark.sql.functions import lower

words_lower = words.select(lower(col("word")).alias("word_lower"))
words_lower.show()

Output

+----------+
|word_lower|
+----------+
|       the|
|   project|
| gutenberg|
|     ebook|
|        of|
|  dracula,|
|        by|
|      bram|
|    stoker|
|          |
|      this|
|     ebook|
|        is|
|       for|
|       the|
|       use|
|        of|
|    anyone|
|  anywhere|
|        in|
+----------+
only showing top 20 rows

Step seven: removing punctuations

so that the same word is not different because of the punctuation at the end of them, is necessary to remove these punctuations.

We'll do this using the regexp_extract function, which extracts words from a string using a regex.

from pyspark.sql.functions import regexp_extract

words_clean = words_lower.select(
    regexp_extract(col("word_lower"), "[a-z]+", 0).alias("word")
)

words_clean.show()

Output

+---------+
|     word|
+---------+
|      the|
|  project|
|gutenberg|
|    ebook|
|       of|
|  dracula|
|       by|
|     bram|
|   stoker|
|         |
|     this|
|    ebook|
|       is|
|      for|
|      the|
|      use|
|       of|
|   anyone|
| anywhere|
|       in|
+---------+
only showing top 20 rows

Step eight: removing null values

However, how you see, there are null values yet, in other words, blank spaces.

It is necessary remove them so that these blanks values are not analyzed.

words_nonull = words_clean.filter(col("word") != "")
words_nonull.show()

Output

+---------+
|     word|
+---------+
|      the|
|  project|
|gutenberg|
|    ebook|
|       of|
|  dracula|
|       by|
|     bram|
|   stoker|
|     this|
|    ebook|
|       is|
|      for|
|      the|
|      use|
|       of|
|   anyone|
| anywhere|
|       in|
|      the|
+---------+
only showing top 20 rows

Step nine: removing stopwords

We are almost there! The last step is removes the stopwords so that, again, these words are not analyzed.

words_without_stopwords = words_nonull.filter(
    ~words_nonull.word.isin(stopwords))

words_count_before_removing = words_nonull.count()
words_count_after_removing = words_without_stopwords.count()

words_count_before_removing, words_count_after_removing

Output

(163399, 50222)

Step ten: analyzing the most common words in Dracula, finally!

And, finally, our data are completely cleared. So, now we could to analyze the most common words in our book.

At first, we’ll group the words and after use an aggregate function to count them.

words_count = (words_without_stopwords.groupby("word")
               .count()
               .orderBy("count", ascending=False)
               )

After, show the top 20 most common words. This value may be changed through rank variable.

rank = 20
words_count.show(rank)

Output

+--------+-----+
|    word|count|
+--------+-----+
|    time|  381|
| helsing|  323|
|     van|  322|
|    lucy|  297|
|    good|  256|
|     man|  255|
|    mina|  240|
|    dear|  224|
|   night|  224|
|    hand|  209|
|    room|  207|
|    face|  206|
|jonathan|  206|
|   count|  197|
|    door|  197|
|   sleep|  192|
|    poor|  191|
|    eyes|  188|
|    work|  188|
|      dr|  187|
+--------+-----+
only showing top 20 rows

Conclusion

That’s all for now, folks! In this article, we analyzed the most common words in Dracula, written by Bram Stoker. To do this, we cleared the words: removing punctuations; converting from uppercase letters to lowercase; and removing stopwords.

I hope you enjoyed it. Keep those stakes sharp, watch out for the shadows that walk at night, and see you in next time 🧛🏼‍♂️🍷.

bibliography

RIOUX, Jonathan. Data Analysis with Python and PySpark.

STOKER, Bram. Dracula.

Uma breve Introdução ao processamento de dados em tempo real com Spark Structured Streaming e Apache Kafka

Geazi Anc — Thu, 29 Sep 2022 18:40:13 +0000

O processamento de dados em tempo real, como o próprio nome diz, é a prática de lidar com o fluxo de dados capturados em tempo real e processados com latência mínima para gerar relatórios instantâneos ou, até mesmo, para produzir respostas automatizadas à um determinado evento.

Hoje, vamos desenvolver uma aplicação bem simples para a ingestão e o processamento de dados em tempo real com o Spark Structured Streaming e o Apache Kafka 🎲. Como este tutorial tem como objetivo ser uma breve introdução à essas tecnologias, vamos desenvolver um simples contador de palavras. Nada muito elaborado ou complexo 😥.

Caso queira ver um exemplo mais completo, confira um outro artigo que escrevi: uma análise de dados em tempo real com base em dados de tráfego aéreo.

Análise de dados de tráfego aéreo em tempo real com Spark Structured Streaming e Apache Kafka

Geazi Anc ・ Oct 28

#dataengineering #python #braziliandevs #spark

E aí, se interessou? Então continue lendo!

Você também pode conferir este projeto em meu GitHub 😉.

O que é Spark Structured Streaming e Apache Kafka?

O Spark Structured Streaming é um módulo do PySpark que facilita a criação de aplicativos e pipelines de streaming com as mesmas e familiares APIs do Spark. O Spark Structured Streaming abstrai conceitos complexos de streaming, como processamento incremental, pontos de verificação e marcas d'água, para que você possa criar aplicativos e pipelines de streaming sem aprender novos conceitos ou ferramentas ❇.

Já o O Apache Kafka é uma plataforma de streaming de eventos distribuídos de código aberto usada por milhares de empresas para pipelines de dados de alto desempenho, análise de streaming, integração de dados e aplicativos de missão crítica.

Os eventos, que podem ser dados capturados em tempo real, são enviados à um tópico do Kafka. Fazendo uma analogia: um tópico é como se fosse uma pasta de arquivos em seu computador, e os eventos são os arquivos desta pasta.

Um produtor, ou producer, é responsável por enviar eventos de streaming à um ou mais tópicos do Kafka. Já um consumidor, ou consumer, é responsável por se inscrever em um ou mais tópicos do Kafka e ler ou processar tais eventos enviados pelo produtor.

Sugiro a leitura da documentação que introduz esses conceitos com mais detalhes 😉.

Agora que já entendemos um pouco sobre os conceitos abordados nesse tutorial, podemos começar a desenvolver nossa aplicação 👏🏼.

Arquitetura

Calma lá! Antes de começarmos, vamos conhecer um pouco sobre a arquitetura de nosso projeto. Ela é composta pelos seguintes componentes lógicos, assim como as tecnologias que serão utilizadas:

Armazenamento de dados analíticos: é onde nossos dados ficarão armazenados. Para isso, iremos criar um tópico no Apache Kafka para que, posteriormente, possamos consumir esses dados que estarão sendo enviados em tempo real pelo produtor.
Ingestão de dados: um produtor desenvolvido em Python que irá enviar palavras aleatórias em tempo real ao tópico criado no Kafka.
Consumo de dados: um consumidor desenvolvido em Python que tem como objetivo apenas monitorar as palavras que estão chegando em tempo real ao tópico do Kafka.
Processamento de fluxo e análise: uma aplicação desenvolvida com o PySpark que irá consumir em tempo real as palavras enviadas pelo produtor ao tópico do Kafka. É esta aplicação que irá agregar as palavras e gerar um relatório da contagem de tais palavras.

Criando o ambiente de desenvolvimento

Este tutorial assume que você já tenha o PySpark instalado em sua máquina. Caso ainda não tenha, confira as etapas na própria documentação.

Já para o Apache Kafka, vamos utilizar ele por meio de conteinerização via Docker 🎉🐳.

E, por fim, vamos utilizar o Python através de um ambiente virtual.

Apache Kafka por conteinerização via Docker

Sem mais delongas, crie uma pasta chamada data-streaming-sample e adicione o arquivo docker-compose.yml dentro dela.

$ mkdir data-streaming-sample
$ cd data-streaming-sample
$ touch docker-compose.yml

Agora, adicione o seguinte conteúdo dentro do arquivo docker-compose:

version: '3.9'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

Feito! Já podemos subir nosso servidor do Kafka. Para isso, digite o seguinte comando no terminal:

$ docker compose up -d
$ docker compose ps

NAME                                COMMAND                  SERVICE             STATUS              PORTS
data-streaming-sample-kafka-1       "/etc/confluent/dock…"   kafka               running             9092/tcp, 0.0.0.0:29092->29092/tcp
data-streaming-sample-zookeeper-1   "/etc/confluent/dock…"   zookeeper           running             2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp

Observação: este tutorial está utilizando a versão 2.0 do Docker Compose. É por este motivo que não há o "-" entre docker e compose ☺.

Agora, precisamos criar um tópico dentro do Kafka que irá armazenar as palavras enviadas em tempo real pelo produtor. Para isso, vamos acessar o Kafka dentro do contêiner:

$ docker compose exec kafka bash

E enfim criar o tópico, chamado de words.

$ kafka-topics --create --topic words --bootstrap-server localhost:29092

Created topic words.

Criação do ambiente virtual

Para desenvolvermos nosso produtor, ou seja, a aplicação que será responsável por enviar as palavras em tempo real para o tópico do Kafka, precisamos fazer o uso da biblioteca kafka-python. O kafka-python é uma biblioteca desenvolvida pela comunidade que nos permite desenvolver produtores e consumidores que se integram com o Apache Kafka.

Primeiro, vamos criar um arquivo chamado requirements.txt e adicionar a seguinte dependência dentro dele:

kafka-python

Segundo, vamos criar um ambiente virtual e instalar as dependências no arquivo requirements.txt:

$ python -m venv venv
$ venv\scripts\activate
$ pip install -r requirements.txt

Feito! Agora sim nosso ambiente já está pronto para o desenvolvimento 🚀.

Desenvolvimento do produtor

Vamos criar nosso produtor. Como foi dito, um produtor é responsável por enviar os dados em tempo real para um tópico no Kafka. Este produtor irá enviar aleatoriamente uma dentre quatro palavras ao tópico words que criamos anteriormente, em um intervalo de tempo aleatório entre um e cinco segundos.

Para isso, criamos uma instância da classe KafkaProducer. Esta classe recebe dois parâmetros:

bootstrap_servers: o servidor onde está rodando o Kafka. Neste caso, ele está rodando no localhost, na porta 29092, conforme configuramos no arquivo docker-compose.
value_serializer: umma função que serializa os dados em bits para serem enviados para o Kafka. Neste caso, a função recebe uma string e retorna um objeto do tipo Bytes.

Depois, utilizamos o método send para enviar os dados ao tópico. Ele recebe dois parâmetros: o tópico que os dados serão enviados e os dados propriamente dito.

Vamos criar um diretório chamado src e um subdiretório chamado kafka. Dentro do diretório kafka, vamos criar um arquivo chamado producer.py e adicionar o seguinte código dentro dele:

import random
from time import sleep
from kafka import KafkaProducer


producer = KafkaProducer(
    bootstrap_servers="localhost:29092",
    value_serializer=lambda x: x.encode("utf-8")
)


while True:
    words = [
        "spark",
        "kafka",
        "streaming",
        "python"
    ]

    word = random.choice(words)
    future = producer.send("words", value=word)

    print(future.get(timeout=60))

    sleep(random.randint(1, 6))

E já podemos executar nosso produtor. Você pode interromper a execução a qualquer momento pressionando CTRL + C.

$ python producer.py

RecordMetadata(topic='words', partition=0, topic_partition=TopicPartition(topic='words', partition=0), offset=0, timestamp=1664469827519, log_start_offset=0, checksum=None, serialized_key_size=-1, serialized_value_size=6, serialized_header_size=-1)
RecordMetadata(topic='words', partition=0, topic_partition=TopicPartition(topic='words', partition=0), offset=1, timestamp=1664469833559, log_start_offset=0, checksum=None, serialized_key_size=-1, serialized_value_size=6, serialized_header_size=-1)
RecordMetadata(topic='words', partition=0, topic_partition=TopicPartition(topic='words', partition=0), offset=2, timestamp=1664469838567, log_start_offset=0, checksum=None, serialized_key_size=-1, serialized_value_size=9, serialized_header_size=-1)
RecordMetadata(topic='words', partition=0, topic_partition=TopicPartition(topic='words', partition=0), offset=3, timestamp=1664469842582, log_start_offset=0, checksum=None, serialized_key_size=-1, serialized_value_size=6, serialized_header_size=-1)
...

Desenvolvimento do consumidor

Vamos criar nosso consumidor. Como foi dito antes, e conforme explicado na arquitetura, um consumidor é responsável por se inscrever em um tópico e ler os eventos que são enviados até ele em tempo real. Nosso consumidor simplesmente irá monitorar as palavras que chegam até o tópico words.

Para isso, vamos criar uma instância da classe KafkaConsumer. Esta classe recebe três parâmetros:

O tópico que queremos que o consumidor se inscreva. Neste caso, o tópico words.
bootstrap_servers: o servidor onde está rodando o Kafka. Neste caso, ele está rodando no localhost, na porta 29092, conforme configuramos no arquivo docker-compose.
value_deserializer: umma função que deserializa os dados de bits para string.

Ainda no diretório kafka, vamos criar um arquivo chamado consumer.py e adicionar o seguinte código dentro dele:

from kafka import KafkaConsumer


consumer = KafkaConsumer(
    "words",
    bootstrap_servers="localhost:29092",
    value_deserializer=lambda x: x.decode("utf-8")
)


for msg in consumer:
    print(msg.value)

Vamos executar nosso consumidor. Certifique-se que o produtor ainda esteja rodando, hein!

$ python consumer.py

kafka
streaming
kafka
streaming
spark
...

E finalizamos o desenvolvimento de nosso produtor e consumidor. Bem simples, não? ☺

Desenvolvimento do processamento de dados em tempo real com Spark Structured Streaming

Esta é uma etapa bem simples, também. Vamos criar uma aplicação com o PySpark que irá se inscrever no tópico words e processar em tempo real os dados que chegam até ele.

Primeiro, com base na instância da classe SparkSession, vamos utilizar o método .writeStream. Depois disso, chamamos uma série de métodos encadeados, entre eles format e options. O format iremos dizer a ele que vamos ler dados em tempo real do Kafka. Já os métodos options informamos o servidor onde está rodando o servidor Kafka, o tópico que ele irá consumir e o modo que ele deve consumir, ou seja, do mais antigo para o mais recente.

Mãos à obra! Vamos criar um novo arquivo no diretório src chamado word_counts.py e adicionar o seguinte código dentro dele:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F


spark = (SparkSession.builder
         .appName("Words Count Analysis")
         .getOrCreate()
         )

df1 = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "localhost:29092")
       .option("subscribe", "words")
       .option("startingOffsets", "earliest")
       .load()
       )

df2 = df1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

lines = df2.select(F.explode(F.split(df2.value, " ")).alias("word"))

word_counts = (lines.groupBy("word")
               .count()
               .orderBy("count", ascending=False)
               )

(word_counts.writeStream
 .format("console")
 .outputMode("complete")
 .start()
 .awaitTermination()
 )

Caso seja a primeira vez que esteja integrando o Apache Spark com o Apache Kafka, talvez a execução do spark-submit demore um pouco. Isso ocorre porque ele precisa fazer o download dos pacotes necessários.

Vamos lá! Sei que seus dedos estão coçando para executar o projeto e ver tudo isso em ação. Portanto, digite no terminal, dentro do diretório src:

`spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 word_counts.py

E veja o resultado!

` +---------+-----+ | word|count| +---------+-----+ | kafka| 11| | python| 10| | spark| 7| |streaming| 6| +---------+-----+ `

Observe que o dataframe será constantemente atualizado conforme o producer envia novos dados ao tópico. Deixe o produtor rodando por um tempo e veja por você mesmo 😉.

Considerações finais

E acabamos por aqui, pessoal. Neste tutorial ensinei como processar dados em tempo real com Spark Structured Streaming e Apache Kafka.

Para isso, desenvolvemos um simples contador de palavras, que agrega as palavras consumidas de um tópico do Kafka e exibe a contagem dessas palavras de forma decrescente. Também desenvolvemos um produtor, que envia dados em tempo real constantemente à um tópico do Kafka e um consumidor, que apenas monitora o tópico na medida que novos dados são enviados pelo produtor.

Espero que tenham gostado. Até a próxima 💚!

PySpark: uma breve análise das palavras mais comuns em Drácula, por Bram Stoker

Geazi Anc — Sat, 24 Sep 2022 16:24:46 +0000

Note: dis article is also available in english 🌎.

Considerado como um marco da literatura gótica, o icônico livro Drácula, escrito em 1897 por Bram Stoker, desperta até hoje o fascínio das pessoas por todo o mundo. Hoje, a fim de introduzir novos conceitos e funcionalidades do Apache Spark, vamos desenvolver uma breve análise das palavras mais comuns encontradas neste clássico livro 🧛🏼‍♂️.

Para isso, vamos desenvolver um notebook no Google Colab, um serviço de nuvem gratuito criado pelo Google para incentivar pesquisas na área de machine learning e inteligência artificial.

Caso não saiba como usar o Google Colab, confira este excelente artigo da Alura escrito pelo Thiago Santos que ensina, de forma muito didática, como usar o Colab e criar seus primeiros códigos!

O notebook deste artigo também está disponível em meu GitHub 😉.

A obra em questão foi obtida por meio do Projeto Gutenberg, um acervo digital que reúne livros de todo o mundo que já se encontram em domínio público. A versão plaintext de Drácula pode ser baixada gratuitamente aqui.

Antes de começar

Antes de iniciarmos o desenvolvimento de nosso notebook, é necessário fazer a instalação da biblioteca PySpark.

A biblioteca PySpark é a API oficial do Python para o Apache Spark. É com ela que vamos realizar nossa análise de dados 🎲.

Crie uma nova célula de código no Colab e execute a seguinte linha:

!pip install pyspark

Passo um: inicialização do Apache Spark

Logo após a instalação, precisamos inicializar o Apache Spark. Para isso, crie uma nova célula de código no Colab e adicione o seguinte bloco:

         from pyspark.sql import SparkSession


spark = (SparkSession.builder
         .appName("The top most common words in Dracula, by Bram Stoker")
         .getOrCreate()
         )

Passo dois: download e leitura de Drácula, por Bram Stoker

Agora sim podemos começar! Nesta etapa iremos fazer o download do livro Drácula do projeto Gutenberg e, logo em seguida, fazer a leitura do arquivo através do PySpark.

O download do livro consiste, basicamente, no uso do utilitário wget, informando a URL que direciona para o livro Drácula no projeto Gutenberg. Depois, salva-se o conteúdo da solicitação, isto é, o próprio livro, no diretório atual, com o nome de Dracula – Bram Stoker.txt.

Crie uma nova célula no colab e adicione o seguinte bloco de código:

!wget https: // www.gutenberg.org/cache/epub/345/pg345.txt -O "Dracula - Bram Stoker.txt"

Passo três: download das stopwords em inglês

A seguir, iremos fazer o download de uma lista das stopwords que são frequentemente usadas no idioma inglês. Essas stopwords normalmente incluem preposições, partículas, interjeições, uniões, advérbios, pronomes, palavras introdutórias, números de 0 a 9 ( inequívocos ), outras partes oficiais da fala, símbolos, pontuação. Recentemente, essa lista foi complementada por sequências de símbolos comumente usadas na Internet como www, com, http, etc.

Essa lista foi adquirida através do site CountWordsFree, um site que, dentre outros utillitários, reúne as stopwords encontradas em diversos idiomas, incluindo o nosso querido português.

Mãos a obra! Crie uma nova célula de código e adicione o seguinte bloco:

!wget https://countwordsfree.com/stopwords/english/txt -O "stop_words_english.txt"

Feito esses downloads, podemos fazer a leitura do livro através do PySpark. Crie uma nova célula no Colab e adicione o seguinte bloco de código:

book = spark.read.text("Dracula - Bram Stoker.txt")

E também vamos fazer a leitura das stopwords que acabamos de baixar. As stopwords serão armazenadas em uma lista, na variável stopwords.

raw_stopwords = spark.read.text("stop_words_english.txt")
stopwords = raw_stopwords.selectExpr("value as stopwords")

stopwords.show()

Output

|  stopwords|
+-----------+
|       able|
|      about|
|      above|
|     abroad|
|  according|
|accordingly|
|     across|
|   actually|
|        adj|
|      after|
| afterwards|
|      again|
|    against|
|        ago|
|      ahead|
|      ain't|
|        all|
|      allow|
|     allows|
|     almost|
+-----------+
only showing top 20 rows

Passo quatro: Extração individual das palavras

Após a leitura do livro, é necessário que transformemos cada uma das palavras em uma coluna no DataFrame.

Para isso, utiliza-se o método split, o qual, para cada uma das linhas, irá separar cada uma das palavras através do espaço em branco entre elas. O resultado será uma lista de palavras.

from pyspark.sql.functions import split


lines = book.select(split(book.value, " ").alias("line"))
lines.show(5)

Output:

+--------------------+
|                line|
+--------------------+
|[The, Project, Gu...|
|                  []|
|[This, eBook, is,...|
|[most, other, par...|
|[whatsoever., You...|
+--------------------+
only showing top 5 rows

Passo cinco: explodindo a lista de palavras em colunas no DataFrame

Depois das palavras terem sido separadas, é necessário que se faça a conversão desta lista de palavras em colunas no DataFrame.

Para tal, usa-se o método explode presente no Apache Spark.

from pyspark.sql.functions import explode, col


words = lines.select(explode(col("line")).alias("word"))
words.show(15)

Output

+---------+
|     word|
+---------+
|      The|
|  Project|
|Gutenberg|
|    eBook|
|       of|
| Dracula,|
|       by|
|     Bram|
|   Stoker|
|         |
|     This|
|    eBook|
|       is|
|      for|
|      the|
+---------+
only showing top 15 rows

Passo seis: transformando todas as palavras em minúsculas

Esta é uma etapa bem simples. Para que não haja distinção da mesma palavra por conta de letras maiúsculas, vamos transformar todas as palavras no DataFrame para letras minúsculas, fazendo o uso da função lower.

from pyspark.sql.functions import lower


words_lower = words.select(lower(col("word")).alias("word_lower"))
words_lower.show()

Output

+----------+
|word_lower|
+----------+
|       the|
|   project|
| gutenberg|
|     ebook|
|        of|
|  dracula,|
|        by|
|      bram|
|    stoker|
|          |
|      this|
|     ebook|
|        is|
|       for|
|       the|
|       use|
|        of|
|    anyone|
|  anywhere|
|        in|
+----------+
only showing top 20 rows

Passo sete: eliminação de pontuação

Para que também não haja distinção da mesma palavra por conta da pontuação presente no final delas, é preciso removê-las.

Isso é feito através do método regexp_extract, o qual extrai palavras de uma string por meio de uma expressão regular.

Calma, não precisa se assustar! A expressão é bem simples. Ela consiste em um conjunto contendo todos os símbolos de A a Z, uma ou mais vezes. Viu, eu te disse que era bem simples 👏🏼.

from pyspark.sql.functions import regexp_extract


words_clean = words_lower.select(
    regexp_extract(col("word_lower"), "[a-z]+", 0).alias("word")
)

words_clean.show()

Output

+---------+
|     word|
+---------+
|      the|
|  project|
|gutenberg|
|    ebook|
|       of|
|  dracula|
|       by|
|     bram|
|   stoker|
|         |
|     this|
|    ebook|
|       is|
|      for|
|      the|
|      use|
|       of|
|   anyone|
| anywhere|
|       in|
+---------+
only showing top 20 rows

Passo oito: remoção de valores nulos

Como visto, mesmo após a remoção das pontuações ainda há colunas com valores nulos, ou seja, espaços em branco.

Para que esses espaços em branco não sejam considerados na análise da frequência de cada palavra presente no livro, é necessário removê-los.

words_nonull = words_clean.filter(col("word") != "")
words_nonull.show()

Output

+---------+
|     word|
+---------+
|      the|
|  project|
|gutenberg|
|    ebook|
|       of|
|  dracula|
|       by|
|     bram|
|   stoker|
|     this|
|    ebook|
|       is|
|      for|
|      the|
|      use|
|       of|
|   anyone|
| anywhere|
|       in|
|      the|
+---------+
only showing top 20 rows

Passo nove: remoção das stopwords

Estamos quase lá! Antes de partirmos para a análise das palavras mais comuns propriamente dita, precisamos remover as stopwords de nosso dataframe, para que elas não sejam levadas em consideração durante a análise.

words_without_stopwords = (
    words_nonull.join(stopwords, words_nonull["word"] == stopwords["stopwords"], how="left")
    .filter("stopwords is null")
    .select("word")
)


words_count_before_removing = words_nonull.count()
words_count_after_removing = words_without_stopwords.count()

words_count_before_removing, words_count_after_removing

Output

(163399, 50222)

Passo dez: análise das palavras mais comuns

E, finalmente, chegamos ao fim da limpesa de nossos dados. Agora sim podemos começar a análise das palavras mais comuns presentes no livro.

Primeiro, é realizado a contagem das palavras mais frequentes no dataframe. Para isso, vamos agrupar cada uma das palavras e depois vamos usar uma função de agregação, count, para determinar quantas vezes elas aparecem.

words_count = (words_without_stopwords.groupby("word")
               .count()
               .orderBy("count", ascending=False)
               )

Depois, vamos exibir as 20 palavras mais comuns. O ranque pode ser ajustado através da variável rank. Sinta-se à vontade para ajustar a variável como preferir.

rank = 20
words_count.show(rank)

Output

+--------+-----+
|    word|count|
+--------+-----+
|    time|  381|
| helsing|  323|
|     van|  322|
|    lucy|  297|
|    good|  256|
|     man|  255|
|    mina|  240|
|    dear|  224|
|   night|  224|
|    hand|  209|
|    room|  207|
|    face|  206|
|jonathan|  206|
|   count|  197|
|    door|  197|
|   sleep|  192|
|    poor|  191|
|    eyes|  188|
|    work|  188|
|      dr|  187|
+--------+-----+
only showing top 20 rows

Considerações finais

É isso por hoje, pessoal. Chegamos no fim de nossa breve análise.

Neste artigo, analisamos as palavras mais comuns do livro Drácula, por Bram Stoker. Para isso, foi necessário fazer uma limpesa nos dados, como dividir as palavras pelos espaços entre elas; explodir a lista de palavras em colunas no dataframe; transformar todas as letras em minúsculas; remover a pontuação de todo o texto através de uma expressão regular; e, por fim, remover as stopwords.

Espero que tenham gostado. Mantenham as estacas afiadas, cuidado com as sombras que andam pela noite, e até a próxima 🧛🏼‍♂️🍷.

Referências

RIOUX, Jonathan. Data Analysis with Python and PySpark.

STOKER, Bram. Dracula.

Introdução à análise de dados com PySpark utilizando os dados dos campeões de League of Legends

Geazi Anc — Thu, 15 Sep 2022 17:57:35 +0000

O League of Legends, também conhecido como lolzinho, para os íntimos, é um jogo ambientado no mundo fantasioso de Runeterra, com batalhas sangrentas e muita magia. Em League of Legends, os jogadores controlam personagens conhecidos como campeões, cada um com suas habilidades e diferentes estilos de jogo.

Neste artigo, iremos analisar algumas estatísticas desses campeões fazendo o uso do PySpark, uma API do framework Apache Spark desenvolvida para a linguagem de programação Python 🐍. Os dados serão extraídos da web API Data Dragon, uma API pública da Riot Games.

Para isso, vamos desenvolver um notebook no Google Colab, um serviço de nuvem gratuito criado pelo Google para incentivar pesquisas na área de machine learning e inteligência artificial.

Caso não saiba como usar o Google Colab, confira este excelente artigo da Alura escrito pelo Thiago Santos que ensina, de forma muito didática, como usar o Colab e criar seus primeiros códigos!

O notebook deste artigo também está disponível em meu GitHub 😉.

Peguem suas espadas, preparem suas magias, e vamos começar ⚔🧙🏼‍♀️!

Instalação

Antes de começarmos, é necessário fazer a instalação de duas bibliotecas: PySpark e Requests.

A biblioteca PySpark, como foi dito, é a API oficial do Python para o Apache Spark. É com ela que vamos realizar nossa análise de dados 🎲.

Já a biblioteca Requests é uma biblioteca que nos permite fazer solicitações HTTP a um determinado website. Mediante a ela que iremos extrair os dados dos campeões através da API pública da Riot Games 🚀.

Crie uma nova célula de código no Colab e execute a seguinte linha:

!pip install pyspark
!pip install requests

Inicialização

Logo após a instalação das bibliotecas, precisamos inicializar o Apache Spark. Para isso, importamos a classe SparkSession dentro do módulo sql da biblioteca pyspark.

Depois da importação, instanciamos a classe SparkSession através de uma série de métodos encadeados, como appName e getOrCreate.

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("Introdução à análise de dados com PySpark utilizando os dados dos campeões de League of Legends")
         .getOrCreate()
         )

Extração de dados dos campeões

A extração dos dados dos campeões de League of Legends é feita através de uma solicitação HTTP à um endpoint da API Data Dragon, uma API pública da Riot Games que centraliza os dados do jogo, como campeões, itens, magias e ETC.

A resposta é um objeto JSON semelhante a este:

{
    "type": "champion",
    "format": "standAloneComplex",
    "version": "12.17.1",
    "data": {
        "Aatrox": {},
        "Ahri": {...},
        "Akali": {...},
        "Akshan": {...},
        "Alistar": {...},
        ...,
    }
}

Observe que os dados que queremos está dentro da chave data. Vamos pegar esses dados, descartando os demais, e exibir apenas o nome de todos os campeões.

Crie uma nova célula de código e execute o seguinte bloco:

import requests

response=requests.get(
"https://ddragon.leagueoflegends.com/cdn/12.17.1/data/pt_BR/champion.json")

champions=response.json().get("data")
champions.keys()

Resultado:

dict_keys(['Aatrox', 'Ahri', 'Akali', 'Akshan', 'Alistar', ...])

Também podemos ver os dados de um campeão em específico. Nesse caso, vamos ver os dados estatísticos da Akali.

champions.get("akali")

Resultado:

{'version': '12.17.1',
 'id': 'Akali',
 'key': '84',
 'name': 'Akali',
 'title': 'a Assassina Renegada',
 'blurb': 'Abandonando a Ordem Kinkou e seu título de Punho das Sombras, Akali agora ataca sozinha, pronta para ser a arma mortal que seu povo precisa. Embora ela mantenha tudo o que aprendeu com seu mestre Shen, ela se comprometeu a defender Ionia de seus...',
 'info': {'attack': 5, 'defense': 3, 'magic': 8, 'difficulty': 7},
 'image': {'full': 'Akali.png',
  'sprite': 'champion0.png',
  'group': 'champion',
  'x': 96,
  'y': 0,
  'w': 48,
  'h': 48},
 'tags': ['Assassin'],
 'partype': 'Energia',
 'stats': {'hp': 570,
  'hpperlevel': 119,
  'mp': 200,
  'mpperlevel': 0,
  'movespeed': 345,
  'armor': 23,
  'armorperlevel': 4.7,
  'spellblock': 37,
  'spellblockperlevel': 2.05,
  'attackrange': 125,
  'hpregen': 9,
  'hpregenperlevel': 0.9,
  'mpregen': 50,
  'mpregenperlevel': 0,
  'crit': 0,
  'critperlevel': 0,
  'attackdamage': 62,
  'attackdamageperlevel': 3.3,
  'attackspeedperlevel': 3.2,
  'attackspeed': 0.625}}

Limpesa dos dados

Antes de começarmos de fato com a análise, é necessário fazermos uma limpesa prévia nos dados. Vamos pegar apenas os que nos interessa, e remover os dicionários dentro de dicionários, deixando um único dicionário para cada campeão com os dados necessários.

champions=[{'name': value['name'], 'title': value['title'], **value['info'], **value['stats']} for key, value in champions.items()]
champions[2]

Resultado:

{'name': 'Akali',
 'title': 'a Assassina Renegada',
 'attack': 5,
 'defense': 3,
 'magic': 8,
 'difficulty': 7,
 'hp': 570,
 'hpperlevel': 119,
 'mp': 200,
 'mpperlevel': 0,
 'movespeed': 345,
 'armor': 23,
 'armorperlevel': 4.7,
 'spellblock': 37,
 'spellblockperlevel': 2.05,
 'attackrange': 125,
 'hpregen': 9,
 'hpregenperlevel': 0.9,
 'mpregen': 50,
 'mpregenperlevel': 0,
 'crit': 0,
 'critperlevel': 0,
 'attackdamage': 62,
 'attackdamageperlevel': 3.3,
 'attackspeedperlevel': 3.2,
 'attackspeed': 0.625}

Criando o DataFrame

Agora sim! Os dados dos campeões estão limpos, então já podemos criar nosso DataFrame com o Spark.

Infelizmente, o Spark é um tanto... seletivo com o tipo de objeto que passamos a ele para criar um DataFrame. Logo, nosso objeto atual champions, que é composto de uma lista de dicionários, não é aceito pelo Spark.

Mas existe uma solução👏🏼. A biblioteca Pandas é muito mais flexível no que se refere a criação de um novo DataFrame. Portanto, é possível criar um DataFrame do Pandas com nosso objeto champions atual, e em seguida criar um DataFrame do Spark com base no DataFrame criado pelo Pandas.

import pandas as pd

df = spark.createDataFrame(pd.DataFrame(champions))

df.select("name", "title").show(5, False)

Resultado:

+-------+-----------------------+
|name   |title                  |
+-------+-----------------------+
|Aatrox |a Espada Darkin        |
|Ahri   |a Raposa de Nove Caudas|
|Akali  |a Assassina Renegada   |
|Akshan |o Sentinela Rebelde    |
|Alistar|o Minotauro            |
+-------+-----------------------+
only showing top 5 rows

Concatenação de colunas

Não sei vocês, mas acho um tanto incômodo ficar selecionando o nome e os títulos dos campeões cada vez que formos visualisar seus dados. Então, vamos concatenar as colunas name e title em uma nova coluna, chamada full_name.

Para isso, vamos primeiramente utilizar o método withColumn. Em resumo, esse método nos permite criar uma nova coluna em nosso DataFrame.

O primeiro parâmetro do método é o nome da nossa coluna. Já o segundo parâmetro são os dados que queremos popular nossa nova coluna. Nesse caso, a concatenação da coluna name com a coluna title.

Para concatenar as colunas de strings, vamos utilizar a função concat.
Esta função recebe como parâmetros o nome das colunas que queremos concatenar. Contudo, não podemos passar apenas o nome dessas colunas. Caso contrário o nome e os títulos ficariam colados um ao outro. Então também usamos a função lit, que cria uma nova coluna literal com o valor que passamos a ela, isto é: ", ".

from pyspark.sql import functions as F

df = df.withColumn("full_name", F.concat(df.name, F.lit(", "), df.title))
df.select("full_name").show(5, False)

Resultado:

+-----------------------------+
|full_name                    |
+-----------------------------+
|Aatrox, a Espada Darkin      |
|Ahri, a Raposa de Nove Caudas|
|Akali, a Assassina Renegada  |
|Akshan, o Sentinela Rebelde  |
|Alistar, o Minotauro         |
+-----------------------------+
only showing top 5 rows

Quem são os campeões mais poderosos de League of Legends?

Curioso para saber quem são os campeões mais poderosos de League of Legends? Pois é, eu também estou. Vamos descobrir 👀!

Para esta análise, considere que o que determina o nível de poder de um campeão são seus valores de ataque, armadura, vida e mana.

Então, para vermos quem são os campeões mais poderosos, basta ordenarmos nosso DataFframe com base nessas colunas, de modo decrescente.

Uma pequena observação: atualmente todos os campeões estão no nível um.

base_columns = ["attackdamage", "armor", "hp", "mp"]

(df.orderBy(*base_columns, ascending=False)
 .select("full_name", *base_columns)
 .show(5, False)
 )

Resultado:

+---------------------------------+------------+-----+-----+-----+
|full_name                        |attackdamage|armor|hp   |mp   |
+---------------------------------+------------+-----+-----+-----+
|Tryndamere, o Rei Bárbaro        |72.0        |33   |696.0|100.0|
|Cho'Gath, o Terror do Vazio      |69.0        |38   |644.0|270.0|
|Renekton, o Carniceiro das Areias|69.0        |35   |660.0|100.0|
|Ornn, O Fogo sob a Montanha      |69.0        |33   |660.0|340.6|
|Kayn, o Ceifador das Sombras     |68.0        |38   |655.0|410.0|
+---------------------------------+------------+-----+-----+-----+
only showing top 5 rows

Level up!

Como dito, atualmente nossos campeões estão no nível 1. Vamos alterar o nível deles para o nível 10.

Observe que as estatísticas dos campeões devem acompanhar seus crescimentos conforme o passar dos níveis. Nesta análise, vamos alterar apenas os valores de dano, armadura, vida e mana.

Para alterarmos esses valores, vamos fazer o uso do método withColumns.
Este método recebe um objeto do tipo dicionário, onde as chaves são os nomes das colunas, e seus valores são as colunas com os dados alterados.

level = 10

df2 = df.withColumns({
    "attackdamage": df.attackdamage+df.attackdamageperlevel*level,
    "armor": df.armor+df.armorperlevel*level,
    "hp": df.hp+df.hpperlevel*level,
    "mp": df.mp+df.mpperlevel*level
})

Quem são os campeões mais poderosos de League of Legends (de novo)?

Com todos os campeões já no nível 10, vamos ver se o rank de poder da análise anterior se manteve ou se houve mudança.
Lembrando que ainda estamos analisando o nível de poder apenas com base nas colunas dano, armadura, vida e mana.

(df2.orderBy(*base_columns, ascending=False)
 .select("full_name", *base_columns)
 .show(5, False)
 )

Resultado:

+-----------------------------+------------+-----+------+-----+
|full_name                    |attackdamage|armor|hp    |mp   |
+-----------------------------+------------+-----+------+-----+
|Illaoi, a Sacerdotisa Cráquem|118.0       |85.0 |1746.0|800.0|
|Olaf, o Berserker            |115.0       |77.0 |1835.0|816.0|
|Darius, a Mão de Noxus       |114.0       |91.0 |1792.0|838.0|
|Yorick, o Pastor de Almas    |112.0       |91.0 |1790.0|900.0|
|Cho'Gath, o Terror do Vazio  |111.0       |85.0 |1584.0|870.0|
+-----------------------------+------------+-----+------+-----+
only showing top 5 rows

Estatísticas dos níveis de poderes

Para finalizar, vamos ver algumas estatísticas simples de todos os nossos campeões no nível 10.

Vamos determinar a média do dano, o máximo do hp e da mana, e o mínimo da armadura.

Utilizaremos o método agg. Este método recebe como parâmetro um dicionário, onde as chaves são o nome das colunas que queremos analisar e os valores são as funções que queremos aplicar sobre elas.

(df2.agg({
    "attackdamage": "mean",
    "hp": "max",
    "mp": "max",
    "armor": "min"
})
    .show()
)

Resultado:

+-------+----------+-----------------+-------+
|max(mp)|min(armor)|avg(attackdamage)|max(hp)|
+-------+----------+-----------------+-------+
|10000.0|      28.0|91.40481987577641| 1892.0|
+-------+----------+-----------------+-------+

Considerações finais

É isso, meus amigos. Finalizamos nossa análise por aqui 🎆.

Neste artigo demonstrei como aplicar uma análise bem simples sobre os dados dos campeões de League of Legends. Fizemos a extração dos dados por meio da API públilca da Riot Games; fizemos uma limpesa prévia nos dados; criamos uma nova coluna com o resultado da concatenação dos nomes dos campeões e seus títulos; ranqueamos os campeões mais poderosos com base em seus níveis de poder; e, por fim, fizemos uma análise das estatísticas dos campeões tanto no nível 1 quanto no nível 10.

Espero que tenham gostado. Até a próxima 💚!

Pokemons Flow: desenvolvendo uma pipeline de dados com apache airflow para extração de pokemon via API

Geazi Anc — Tue, 13 Sep 2022 13:51:47 +0000

O Apache Airflow é uma plataforma desenvolvida pela comunidade para criar, agendar e monitorar fluxos de trabalho, tudo feito programaticamente. Com ela, os pipelines do Airflow são definidos em Python, permitindo a geração dinâmica de pipeline, sem sair da sintaxe que já conhecemos 🐍.

Saber desenvolver pipeline de dados com o Apache Airflow é um requisito mais do que essencial caso você almege uma carreira em engenharia de dados. Portanto, caso queira saber mais sobre essa poderosa ferramenta, continue lendo 😉.

Neste artigo irei ensinar como desenvolver uma pipeline de dados para extrair os famosos monstrinhos de bolso, os pokemons, da API PokeAPI. Depois da extração será aplicado algumas transformações bem simples nos dados para que, por fim, possam ser salvos localmente, simulando o carregamento dos dados em um data warehouse.

O código completo desenvolvido neste artigo pode ser conferido em meu GitHub 🤖.

Desenvolvimento baseado em conteinerização via Docker Compose

Existe muitas maneiras de instalar o Airflow em sua máquina. Todas elas podem ser conferidas na própria documentação oficial.

Para este artigo, no entanto, iremos desenvolver nossa pipeline baseada em conteinerização via Docker Compose.
Caso não tenha o Docker instalado, confira o guia de instalação na documentação.

O arquivo YML que irá subir nosso cluster pode ser baixado aqui, diretamente da página oficial do Airflow. Nenhuma mudança no arquivo é necessária para este projeto 💚.

O download deste arquivo é necessário para as próximas etapas.

Inicializando o cluster

Primeiro, vamos criar um diretório chamado pokemonsflow e adicionar o arquivo docker-compose.yml.

Depois disso, abra o terminal no diretório e digite os seguintes comandos para inicializar o cluster do Airflow:

$ docker-compose up airflow-init
$ docker-compose up -d
$ docker-compose ps

Com isso, seu cluster do Airflow já está ativo e o ambiente está pronto para o desenvolvimento 🚀.

Note que agora seu diretório pokemonsflow possue três novos subdiretórios:

dags/
logs/
plugins/
docker-compose.yml

Para finalizar, crie um subdiretório chamado data no diretório dags. Este diretório é onde ficará salvo os dados extraídos pela pipeline:

$ mkdir dags/data

Desenvolvimento da pipeline de dados

Agora que nosso ambiente de desenvolvimento via Docker já está inicializado, podemos começar a dar os primeiros passos na estruturação de nosso DAG 🎆.

Antes de tudo, crie um arquivo chamado pokemonsflow_dag.py no subdiretório dags. Note que o sufixo _dag no arquivo é necessário para o Airflow reconhecer automaticamente nosso DAG 😉:

$ cd dags
$ touch pokemonsflow_dag.py

Depois disso, adicione o seguinte código no arquivo:

import pandas as pd
import pendulum
import requests
from airflow.decorators import dag, task


@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2022, 1, 1, tz='UTC'),
    catchup=False
)
def pokemonsflow_dag():

    @task
    def extract() -> list:
        pass

    @task
    def transform(pokemons: list) -> list:
        pass

    @task
    def load(pokemons: list):
        pass


dag = pokemonsflow_dag()

Com isso já temos a estrutura inicial de nosso DAG. A biblioteca requests e pandas são necessárias para as tarefas de extração e transformação dos dados.

No DAG temos três tarefas principais, onde:

Extract: irá fazer a extração de vinte pokemons da PokeAPI;
Transform: irá selecionar apenas cinco campos dos pokemons extraídos, e ordená-los de forma decrescente pelo campo base_experience;
Load: por fim, esta task irá salvar os dados transformados no subdiretório /dags/data/, no formato CSV;

Como pôde perceber, tanto a task transform quanto a task load dependem dos dados extraídos ou transformados pela task anterior. Para que a transição dos dados seja feita entre as tasks, o Airflow usa um mecanismo interno chamado XComs, uma abreviação para Comunicação Cruzada.

Antes do Airflow 2.0, o compartilhamento de dados entre tasks usando XComs era um tanto... verbosa. Contudo, com a chegada do Airflow 2.0, podemos compartilhar os dados entre tasks apenas passando eles como se fossem parâmetros de funções. Simples, não? ☺

Agora vamos desenvolver individualmente cada task de nosso DAG.

Extração de dados

A extração de pokemons da API é feita através das seguintes etapas:

Faz uma chamada GET no endpoint /api/v2/pokemon, com o parâmetro limit=20 para restringirmos os resultados. O resultado será um json com o campo result, semelhante a este:


[
  {
    'name': 'bulbasaur',
    'url': 'https://pokeapi.co/api/v2/pokemon/1/'
  },
  {
    'name': 'ivysaur',
    'url': 'https://pokeapi.co/api/v2/pokemon/2/'
  },
  {
    'name': 'venusaur',
    'url': 'https://pokeapi.co/api/v2/pokemon/3/'
  },
  {
    'name': 'charmander',
    'url': 'https://pokeapi.co/api/v2/pokemon/4/'
  },
  ...
]

Depois, acessamos o campo results e fazemos uma chamada GET para cada uma das URLS. O resultado será uma lista com os vinte pokemons extraídos da API;

Mãos à obra! Adicione o seguinte código dentro da função extract dentro de nosso DAG, não esquecendo, é claro, de remover a palavra reservada pass:

    @task
    def extract() -> list:
        url = 'http://pokeapi.co/api/v2/pokemon'

        params = {
            'limit': 20
        }

        response = requests.get(url=url, params=params)

        json_response = response.json()
        results = json_response['results']

        pokemons = [requests.get(url=result['url']).json()
                    for result in results]

        return pokemons

Parabéns! A task para extração dos vinte pokemons da API está concluída 🎉.

Transformação dos dados

Caso faça a requisição do endpoint via Postman, por exemplo, você vai notar que cada pokemon contém inúmeros campos, tais como forms, stats, ETC. Contudo, iremos pegar apenas cinco campos desse json.

As etapas da task transform são as seguintes:

Pega os dados extraídos pela task anterior, isto é, a task extract;
Cria um DataFrame do Pandas com os dados e seleciona cinco colunas, descarrtando as demais;
Ordena os dados do DataFrame pela coluna base_experience, de forma decrescente;
Converte o DataFrame para uma lista de dicionários do Python, para que os dados possam ser transferidos para a task posterior;

Agora, mãos à obra, de novo! Adicione o seguinte código dentro da função transform dentro de nosso DAG, não esquecendo, é claro, de remover a palavra reservada pass:

    @task
    def transform(pokemons: list) -> list:

        columns = [
            'name',
            'order',
            'base_experience',
            'height',
            'weight'
        ]

        df = pd.DataFrame(data=pokemons, columns=columns)
        df = df.sort_values(['base_experience'], ascending=False)

        pokemons = df.to_dict('records')

        return pokemons

Feito! A task para a transformação dos dados está concluída 🎉.

Carregamento dos dados

Por fim, só o que nos resta é desenvolver a task load. As etapas desta task são:

Pega os dados transformados pela task anterior;
Cria um DataFrame do Pandas com os dados transformados;
Salva os dados do DataFrame no diretório /dags/data/, no formato CSV;

Adicione o seguinte código dentro da função load dentro de nosso DAG, não esquecendo, é claro, de remover a palavra reservada pass:

    @task
    def load(pokemons: list):

        df = pd.DataFrame(data=pokemons)
        df.to_csv('./dags/data/pokemons_dataset.csv', index=False)

Parabéns! Todas as tasks de nosso DAG foram desenvolvidas 🎉.

Orquestração das tarefas e transferência de dados

Calma, ainda não acabou! Antes de finalizar, precisamos dizer ao Airflow a ordem para executar essas tarefas, e transferir os dados de uma task para a outra. Abaixo das funções, adicione o seguinte código:

    # ETL pipeline

    # extract
    extracted_pokemons = extract()

    # transform
    transformed_pokemons = transform(pokemons=extracted_pokemons)

    # load
    load(pokemons=transformed_pokemons)

Agora sim acabamos 👏🏼. Note que a transferência dos dados entre as tasks é feito passando-os como parâmetros das funções. Muito, muito simples!

O código final de nosso DAG ficou assim:

import pandas as pd
import pendulum
import requests
from airflow.decorators import dag, task


@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2022, 1, 1, tz='UTC'),
    catchup=False
)
def pokemonsflow_dag():

    @task
    def extract() -> list:
        url = 'http://pokeapi.co/api/v2/pokemon'

        params = {
            'limit': 20
        }

        response = requests.get(url=url, params=params)

        json_response = response.json()
        results = json_response['results']

        pokemons = [requests.get(url=result['url']).json()
                    for result in results]

        return pokemons

    @task
    def transform(pokemons: list) -> list:

        columns = [
            'name',
            'order',
            'base_experience',
            'height',
            'weight'
        ]

        df = pd.DataFrame(data=pokemons, columns=columns)
        df = df.sort_values(['base_experience'], ascending=False)

        pokemons = df.to_dict('records')

        return pokemons

    @task
    def load(pokemons: list):

        df = pd.DataFrame(data=pokemons)
        df.to_csv('./dags/data/pokemons_dataset.csv', index=False)

    # ETL pipeline

    # extract
    extracted_pokemons = extract()

    # transform
    transformed_pokemons = transform(pokemons=extracted_pokemons)

    # load
    load(pokemons=transformed_pokemons)


dag = pokemonsflow_dag()

Testano o DAG

Já podemos testar se nosso DAG está funcionando conforme o esperado. Para isso, vá até o terminal que está aberto na raiz do diretório e digite os seguintes comandos:


$ docker-compose exec airflow-worker bash
$ airflow dags test pokemonsflow_dag 2022-01-01

Quando a execução do DAG for finalizada, vá até o diretório /dags/data/ e confira o arquivo pokemons_dataset.csv. Todos os vinte pokemons estarão ordenados conforme a coluna base_experience.

Considerações finais

Como vimos, o Apache Airflow é uma poderosa ferramenta para a orquestração de tarefas de uma pipeline de dados. Não esqueça de conferir as outras inúmeras funcionalidades do Airflow 😉.

Neste artigo ensinei como desenvolver um DAG para extração de vinte pokemons da API PokeAPI. Foi aplicado algumas transformações bem simples nos dados antes de serem salvos localmente em formato CSV.

Se você gostou desse artigo, não esqueça de curtir e compartilhar nas redes sociais 💚.

Até a próxima!