DEV Community: Mayara Machado

Pytest Mocks, o que são?

Mayara Machado — Wed, 30 Oct 2024 12:23:15 +0000

Esse texto faz é o primeiro de uma série de textos sobre testes em aplicações de processamento de dados que estarei trazendo aqui e em meu blog pessoal.

Quando fiz minha transição de carreira de engenheira de software para engenheira de dados, passei a ter conversas com pessoas da área de dados que não possuíam uma formação em engenharia de software. Nessas conversas, um questionamento surgia repetidamente: como escrever testes?

Escrever testes pode, de fato, parecer uma tarefa complexa para quem não está acostumado, já que exige uma mudança na forma de escrever código. A verdade é que não há mistério algum, mas sim uma questão de prática e repetição. Meu objetivo principal neste artigo é guiar você, que está começando, em um processo que mostra como podemos criar testes para aplicações que processam dados, garantindo qualidade e confiabilidade no código.

Esse texto faz parte de uma série que estarei trazendo ao longo das próximas semanas onde compartilho como escrever testes automatizados em código voltado para engenharia de dados. No artigo de hoje quero explorar um pouco sobre mocks. Em diversos cenários de código, uma pipeline de dados estará realizando conexões, chamadas de API, integração com serviços da Cloud e etc, o que pode causar uma certa confusão em como podemos testar essa aplicação. Vamos explorar hoje algumas bibliotecas interessantes para escrita de testes focando no uso dos mocks.

Afinal, o que são Mocks?

Mocks são objetos simulados usados em testes para imitar o comportamento de dependências externas ou componentes que não são o foco do teste. Eles permitem que você isole a unidade de código que está sendo testada, garantindo que o teste seja mais controlável e preditivo. A utilização de mocks é uma prática comum em testes unitários e testes de integração.

E devemos usar os mocks quando:

A dependência é irrelevante para o teste;
A dependência não está disponível;
Queremos testar comportamentos especiais, simulações de erros ou respostas específicas.

Em pipelines de dados, o Mocking permite criar representações de componentes externos – como um banco de dados, um serviço de mensageria ou uma API – sem depender de suas infraestruturas reais. Isso é particularmente útil em ambientes de processamento de dados, que integram várias tecnologias, como PySpark para processamento distribuído, Kafka para mensageria, além de serviços em nuvem, como AWS e GCP.

Nesses cenários onde temos pipelines de dados, o Mocking facilita a execução de testes isolados e rápidos, minimizando custos e tempo de execução. Ele permite que cada parte do pipeline seja verificada com precisão, sem falhas intermitentes causadas por conexões reais ou infraestrutura externa, e com a segurança de que cada integração funciona como esperado.

Podemos encontrar em cada linguagem de programação, módulo internos que já disponibilizam funções de Mock para serem implementados. Em Python, a biblioteca nativa unittest.mock é a principal ferramenta para criar mocks, permitindo simular objetos e funções com facilidade e controle. Em Go, o processo de Mocking é comumente suportado por pacotes externos, como o mockery, pois a linguagem não possui uma biblioteca nativa de Mock; o mockery é especialmente útil para gerar mocks a partir de interfaces, um recurso nativo de Go. Já em Java, o Mockito se destaca como uma biblioteca popular e poderosa para criar mocks, integrando-se ao JUnit para facilitar testes unitários robustos. Essas bibliotecas oferecem uma base essencial para testar componentes isolados, especialmente em pipelines de dados e sistemas distribuídos onde a simulação de fontes de dados e APIs externas é crítica.

Implementando Mocks

Vamos começar com um exemplo básico de como podemos usar Mocks. Supondo que temos uma função que realiza chamadas de API e precisamos realizar a escrita de testes unitários para essa função:

def get_data_from_api(url):
    import requests
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

Para abordar corretamente os cenários de teste, precisamos primeiro entender quais situações devem ser cobertas. Como nossa função realiza chamadas REST, os testes devem contemplar, no mínimo, dois cenários principais: um em que a requisição é bem-sucedida e outro em que a resposta não é a esperada. Poderíamos executar o código com uma URL real para observar o comportamento, mas essa abordagem apresenta desvantagens, pois não teríamos controle sobre os diferentes tipos de resposta, além de deixar o teste vulnerável a mudanças na resposta da URL ou a sua eventual indisponibilidade. Para evitar essas inconsistências, utilizaremos Mocks.

from unittest import mock

@mock.patch('requests.get')
    def test_get_data_from_api_success(mock_get):
        # Configura o mock para retornar uma resposta simulada
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"key": "value"}

        # Chama a função com o mock ativo
        result = get_data_from_api("http://fakeurl.com")

        # Verifica se o mock foi chamado corretamente e o resultado é o esperado
        mock_get.assert_called_once_with("http://fakeurl.com")
        self.assertEqual(result, {"key": "value"})

Com a decoração @mock.patch da biblioteca unittest do Python, podemos substituir a chamada requests.get por um mock, um "objeto falso" que simula o comportamento da função get dentro do contexto de teste, eliminando a dependência externa.

Ao definir valores para o return_value do mock, podemos especificar exatamente o que esperamos que o objeto retorne ao ser chamado na função que estamos testando. É importante que a estrutura do return_value siga a mesma dos objetos reais que estamos substituindo. Por exemplo, um objeto response do módulo requests possui atributos como status_code e métodos como json(). Assim, para simularmos uma resposta da função requests.get, podemos atribuir o valor esperado a esses atributos e métodos diretamente no mock.

response.status_code = mock_get.return_value.status_code
response.json() = mock_get.return_value.json.return_value

Neste caso específico, o foco é simular a resposta da requisição, ou seja, testar o comportamento da função com diferentes resultados esperados sem depender de uma URL externa e sem impacto em nosso ambiente de testes.

from unittest import TestCase, mock

# Função que será testada
def get_data_from_api(url):
    import requests
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

# Teste usando Mock
class TestGetDataFromApi(TestCase):
    @mock.patch('requests.get')
    def test_get_data_from_api_success(self, mock_get):
        # Configura o mock para retornar uma resposta simulada
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"key": "value"}

        # Chama a função com o mock ativo
        result = get_data_from_api("http://url.com")

        # Verifica se o mock foi chamado corretamente e o resultado é o esperado
        mock_get.assert_called_once_with("http://url.com")
        self.assertEqual(result, {"key": "value"})

    @mock.patch('requests.get')
    def test_get_data_from_api_failure(self, mock_get):
        # Configura o mock para simular uma resposta com erro
        mock_get.return_value.status_code = 404

        # Chama a função com o mock ativo
        result = get_data_from_api("http://url.com")

        # Verifica o comportamento em caso de erro
        mock_get.assert_called_once_with("http://url.com")
        self.assertIsNone(result)

# Execução dos testes (se for usado em script)
if __name__ == '__main__':
    import unittest
    unittest.main()

Ao simular respostas de erro da API em testes, podemos ir além do básico e verificar o comportamento da aplicação em relação a diferentes tipos de códigos de status HTTP, como 404, 401, 500 e 503. Isso proporciona uma cobertura mais ampla e assegura que a aplicação lida adequadamente com cada tipo de falha, entendo de que forma essas variações na chamada podem impactar nossa aplicação/processamento de dados. Em chamadas de método POST, podemos adicionar uma camada extra de validação, verificando não apenas o status_code e o funcionamento básico da chamada, mas também o schema do envio e da resposta recebida, garantindo que os dados retornados sigam o formato esperado. Essa abordagem de teste mais detalhada ajuda a prevenir problemas futuros, assegurando que a aplicação esteja preparada para lidar com uma variedade de cenários de erro e que os dados recebidos estejam sempre de acordo com o que foi projetado.

Implementando Mocks com PySpark

Agora que vimos um caso simples de utilização de Mocks em um código puramente python, vamos expandir nossos casos para um trecho de código que usa Pyspark.

Para testar funções PySpark, especialmente operações em DataFrame como filter, groupBy, e join, o uso de mocks é uma abordagem eficaz que elimina a necessidade de executar o Spark real, reduzindo o tempo de teste e simplificando o ambiente de desenvolvimento. A biblioteca unittest.mock do Python permite simular comportamentos desses métodos, possibilitando verificar o fluxo e a lógica de código sem dependência da infraestrutura Spark.

Vejamos, dado a seguinte função onde temos uma transformação que realiza operação de filtro, groupBy e join em dataframes em Spark.

def transform_data(df, df_other):
    filtered_df = df.filter(df['value'] > 10)
    grouped_df = filtered_df.groupBy('category').count()
    result_df = grouped_df.join(df_other, on="category", how="inner")
    return result_df

Para executar um teste PySpark precisamos que a configuração do Spark seja feita localmente. Essa configuração é feita no método setUpClass, que cria uma instância do Spark que será utilizada em todos os testes da classe. Isso nos permite executar PySpark de forma isolada, possibilitando a realização de operações de transformação reais sem depender de um cluster completo. Após a conclusão dos testes, o método tearDownClass é responsável por finalizar a sessão do Spark, garantindo que todos os recursos sejam liberados adequadamente e que o ambiente de teste seja limpo.

from pyspark.sql import SparkSession
from pyspark.sql import DataFrame
from pyspark.sql import functions as F
import unittest
from unittest import mock

class TestTransformData(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.spark = SparkSession.builder.master("local[1]").appName("PySparkTest").getOrCreate()

    @classmethod
    def tearDownClass(cls):
        cls.spark.stop()

    def test_transform_data(self):
        # Cria dados de entrada
        df_data = [
            ("A", 15),
            ("B", 5),
            ("A", 20),
            ("C", 25)
        ]
        df_other_data = [("A", "description A"), ("C", "description C")]

        # Cria DataFrames a partir dos dados simulados
        df = self.spark.createDataFrame(df_data, ["category", "value"])
        df_other = self.spark.createDataFrame(df_other_data, ["category", "description"])

        # Executa a função com processamento real
        result_df = transform_data(df, df_other)

        # Realiza verificações nas operações do DataFrame resultante
        result_data = result_df.collect()  # Pega todos os dados do DataFrame em uma lista de linhas
        expected_data = [("A", 2, "description A"), ("C", 1, "description C")]

        # Converte os resultados em uma lista de tuplas para fácil comparação
        result_tuples = [(row.category, row['count'], row.description) for row in result_data]
        self.assertEqual(result_tuples, expected_data)

No teste test_transform_data, começamos criando DataFrames de exemplo para df e df_other, os quais contêm os dados que serão utilizados nas transformações. Em seguida, executamos a função transform_data sem aplicar mocks, permitindo que as operações de filter, groupBy e join ocorram de fato e resultem em um novo DataFrame. Após a execução, utilizamos o método collect() para extrair os dados do DataFrame resultante, o que nos permite comparar esses dados com os valores esperados e, assim, validar a transformação realizada de maneira real e precisa.

Mas também podemos ter cenários onde queremos testar o resultado de uma dessas funções do pyspark. Que seja necessário mockar uma outra parte do código que possa estar representando um gargalo no tempo de execução e que não represente um risco para o nosso processo. Sendo assim, podemos nos apoiar da técnica de mockar uma função/módulo, conforme vimos no exemplo anterior utilizando requests.

from pyspark.sql import SparkSession
from pyspark.sql import DataFrame
from pyspark.sql import functions as F
import unittest
from unittest import mock

class TestTransformData(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.spark = SparkSession.builder.master("local[1]").appName("PySparkTest").getOrCreate()

    @classmethod
    def tearDownClass(cls):
        cls.spark.stop()

    @mock.patch('pyspark.sql.DataFrame.filter')
    def test_transform_data_with_mocked_filter(self, mock_filter):
        # Mock para a operação filter, mantendo o processamento real nas outras etapas
        mock_filtered_df = self.spark.createDataFrame([("A", 15), ("A", 20), ("C", 25)], ["category", "value"])
        mock_filter.return_value = mock_filtered_df

        df_data = [("A", 15), ("B", 5), ("A", 20), ("C", 25)]
        df_other_data = [("A", "description A"), ("C", "description C")]

        df = self.spark.createDataFrame(df_data, ["category", "value"])
        df_other = self.spark.createDataFrame(df_other_data, ["category", "description"])

        # Executa a função, agora com filter mockado
        result_df = transform_data(df, df_other)
        result_data = result_df.collect()

        # Compara com o dado esperado
        expected_data = [("A", 2, "description A"), ("C", 1, "description C")]  # O resultado deve ser ajustado conforme a lógica do teste
        result_tuples = [(row.category, row['count'], row.description) for row in result_data]
        self.assertEqual(result_tuples, expected_data)

O teste com Mock para operação específica foi realizado no método test_transform_data_with_mocked_join, onde aplicamos um mock especificamente para o método filter. Esse mock substitui o resultado da operação de join por um DataFrame simulado, permitindo que as operações anteriores, como groupBy e join, sejam executadas de forma real. O teste, então, compara o DataFrame resultante com o valor esperado, garantindo que o mock para join foi utilizado corretamente, sem interferir nas demais transformações realizadas.

Essa abordagem híbrida traz várias vantagens. Ao garantir que as operações reais de PySpark, como join e groupBy, sejam mantidas, podemos validar a lógica das transformações sem perder a flexibilidade de substituir operações específicas, como filter, com mocks. Isso resulta em testes mais robustos e rápidos, eliminando a necessidade de um cluster Spark completo, o que facilita o desenvolvimento e a validação contínua do código.

É importante ressaltar que essa estratégia deve ser utilizada com cautela e somente em cenários onde não se cria um viés nos resultados. O objetivo do teste é assegurar que o processamento ocorra de forma correta; não devemos simplesmente atribuir valores sem testar a função de fato. Embora seja válido mockar trechos que podemos garantir que não afetarão o processo do teste unitário, é essencial lembrar que a função deve ser executada para validar seu comportamento real.

Assim, a abordagem híbrida faz muito mais sentido quando temos outros tipos de processamento agregado nessa função. Essa estratégia permite uma combinação eficaz de operações reais e simuladas, garantindo testes mais robustos e confiáveis

Concluindo

Os mocks são aliados valiosos na criação de testes eficazes, especialmente quando se trata de trabalhar com PySpark e outros serviços na nuvem. A implementação que exploramos usando unittest em Python não apenas nos ajudou a simular operações, mas também a manter a integridade dos nossos dados e processos. Com a flexibilidade que os mocks oferecem, podemos testar nossos pipelines sem o medo de causar estragos em ambientes de produção. E aí, preparados para o próximo desafio? No nosso próximo texto, vamos mergulhar no mundo das integrações com serviços da AWS e GCP, mostrando como mockar essas chamadas e garantir que suas pipelines funcionem perfeitamente. Até a próxima!

ETL com Apache Airflow, Web Scraping, AWS S3, Apache Spark e Redshift | Parte 1

Mayara Machado — Mon, 04 Jan 2021 15:41:55 +0000

Nesse post irei apresentar como utilizar o Apache Airflow, uma das mais conhecidas ferramenta para gerenciamente de fluxos, para automatizar um fluxo de Extração, Transformação e Carregamento do inglês Extract, Transform, Load (ETL). Para abordar melhor os tópicos envolvidos nesse projeto, o mesmo foi dividido em duas partes que irão originar duas publicações.

Nessa primeira parte mostrarei como realizar o desenvolvimento do início de um processo ETL utilizando Airflow. Constuíremos um Web Scraper utilizando BeautifulSoup para extrair informações de avaliação de filmes do site AdoroCinema, com os dados obtidos nós iremos escrever um arquivo .csv e enviar para o AWS S3. Na segunda parte desse projeto, iremos realizar a integração do Airflow com o Apache Spark para realizar um pequeno processamento em cima dos arquivos enviados para a AWS S3 e então armazenaremos o resultado do processamento na Amazon Redshift.

O código do projeto está disponível aqui.

Airflow

Foi um projeto desenvolvido internamente pela equipe de desenvolvimento da empresa Aibnb, surgiu como uma ferramenta para solucionar o problema que o setor encontrava com a crescente complexidade no ambiente de dados da empresa. O Airflow surgiu no final de 2014 e foi apresentado no blog da Airbnb em 2015, e desde então se tornou popular no meio da comunidade de engenharia de dados. Por fim, depois de algum tempo o projeto foi cedida para o Apache que hoje é o responsável por mantér o projeto, hoje o projeto se chama Apache Airflow.

Configuração inicial do projeto

Para esta etapa do desenvolvimento será necessário ter disponível:

docker e docker-compose em sua máquina.
Uma conta AWS.
aws cli instalado em sua máquina.
Bucket S3 configurado.

A imagem do docker utilizada foi a puckel/docker-airflow onde acrescentei o BeautifulSoup como dependência para criação da imagem em minha máquina.

As configurações de credenciais da AWS utilizadas pelo Airflow dentro do container Docker são obtidas do arquivo local .aws gerado no momento que você executa o seguinte comando no bash:

$ aws configure

Fluxo a ser implementado

O projeto consistirá em duas etapas, nessa primeira publicação implementaremos o fluxo da parte 1.

Web Scraping com BeautifulSoup

Web Scraping é o processo de obter informações que estão disponíveis na internet de forma automática. Para realizar esse processo, antes precisamos conhecer melhor sobre a página web que iremos explorar, averiguar o HTML da página é um passo muito importante para poder entender quais atributos você irá buscar durante a construção do seu Web Scraper. Para isso você pode utilizar a funcionalidade Developer Tools presente no seu navegador e explorar a estrutura do HTML da informação que você gostaria de obter.

Para a minha coleta escolhi selecionar informações referente ao título, gênero, url da foto do poster, sinopse resumida, data de lançamento, sinope completa, as avaliações disponíveis e o link para a publicação do adorocinema sobre o dado filme. A sinopse completa foi obtida a partir do link da publicação, visto que nessa página de listagem não é mostrada a sinopse completa. As vezes navegar por links para montar uma resposta é necessário pois a informação pode não está sendo exibida de forma completa na página que você iniciou a busca.

Além disso outro tópico que levei em consideração foi as avaliações, eu queria obter todas as notas disponíveis e é importante ressaltar que existem 3 classificações de avaliação: AdoroCinema, Imprensa e Leitores e que para alguns filmes uma classificação ou mais pode não possuir nota. Para isso foi necessário fazer um tratamento no momento de selecionar o componente HTML correto, para o avaliador que não possuísse classificação eu completei com uma informação nula pois futuramente posso aplicar algumas técnicas de Data Cleaning.

from bs4 import BeautifulSoup
from urllib.request import urlopen, Request

def scraping_avaliacao_adoro_cinema(page = 1):
    URL = f"http://www.adorocinema.com/filmes/todos-filmes/notas-espectadores/?page={page}"

    html_doc = urlopen(URL).read()
    soup = BeautifulSoup(html_doc, "html.parser")
    data = []

    for dataBox in soup.find_all("div", class_="data_box"):
        titleObj = dataBox.find("a", class_="no_underline")
        imgObj = dataBox.find(class_="img_side_content").find_all(class_="acLnk")[0]
        sinopseObj = dataBox.find("div", class_="content").find_all("p")[0]
        dateObj = dataBox.find("div", class_="content").find("div", class_="oflow_a")
        movieLinkObj = dataBox.find(class_="img_side_content").find_all("a")[0]
        generoObj = dataBox.find("div", class_="content").find_all('li')[3].find('div',class_="oflow_a")
        detailsLink = 'http://www.adorocinema.com' + movieLinkObj.attrs['href']

        avaliacoesMeios = dataBox.find("div", class_="margin_10v").find_all('span', class_="acLnk")
        avaliacoesNotas = dataBox.find("div", class_="margin_10v").find_all('span', class_="note")

        # tratar a lista de avaliações
        avaliadores = [ elem.text.strip() for elem in avaliacoesMeios if elem.text.strip() != "" ]

        # vincular com notas
        avaliacoesValores = {"AdoroCinema" : None, "Leitores" : None, "Imprensa" : None}
        for index, avaliador in enumerate(avaliadores):
          avaliacoesValores[avaliador] = avaliacoesNotas[index].text.strip()     

        #Carregar a sinopse completa 
        htmldocMovieDetail = urlopen(detailsLink).read()
        soupMovieDetail = BeautifulSoup(htmldocMovieDetail, "html.parser")
        fullSinopse = soupMovieDetail.find(class_="content-txt")     
        fullImgObj = soupMovieDetail.find("meta",  property="og:image")   

        data.append({'titulo': titleObj.text.strip(),
                    'genero': generoObj.text.replace('\n','').strip(),
                    'poster' : fullImgObj["content"], 
                    'sinopse' : sinopseObj.text.strip(),
                    'data' :  dateObj.text[0:11].strip(),
                    'link' : detailsLink,
                    'avaliacoes' : avaliacoesValores,
                    'sinopseFull': fullSinopse.text})

    return data

Aqui você pode conferir um artigo muito completo explicando a utilização do BeautifulSoup.

DAGs

O Airflow utiliza uma estrutura chamada DAG - Directed Acyclic Graph, que de modo geral é um grafo acíclico onde todas as tarefas a serem executadas são estruturadas refletindo suas dependências e relacionamentos. Existem muitas configuração que podem ser feitas, como aplicação de condicionais de uso como executar um nó apenas se o anterior tiver resultado em sucesso e coisas assim, é possível configurar timeout, tempo de reiniciação e etc.

Por default o Aifrlow identifica as DAGs do projeto como todo arquivo python armazenado na pasta dags, mas essa é uma configuração que pode ser sobrescrita caso você deseje mudar a localização das suas DAGs.

Criando uma DAG

Para criar uma DAG precisamos definir alguns atributos iniciais, como a dag_id, schedule_interval e alguns args.

dag_id: é o identificador único do fluxo, aqui chamamos de 'etl_movie_review'.
schedule_interval: indica a periodicidade da execução das tasks.
default_args: são informações que serão refletidos em todos os nós do DAGs.

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

from etl_scripts.load_to_s3 import local_to_s3
from etl_scripts.web_scraping import collect_adoro_cinema_data
from etl_scripts.remove_local_file import remove_local_file

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2021, 1, 1),
    "retries": 0,
}

with DAG(
    "etl_movie_review", 
    schedule_interval=timedelta(minutes=1),
    catchup=False,
    default_args=default_args
) as dag:
    (...)

Criando as tasks da DAG

Para construir nossos nós o Airflow utiliza os Operators, eles são classes que encapsulam a lógica para realizar o trabalho para determinados tipos de conexão. No nosso projeto utilizaremos por enquanto apenas o PythonOperator e o DummyOperator, para realizar chamadas de execução de código python e para sinalizar inicio e fim de fluxos, respectivamente. Além desses existem alguns outros, você pode encontrar a lista de Operators aqui.

(...)
   # Inicio do Pipeline
    start_of_data_pipeline = DummyOperator(task_id='start_of_data_pipeline', dag=dag)

    # Definindo a tarefa para realização de web scrapping
    movie_review_web_scraping_stage = PythonOperator(
        task_id='movie_review_web_scraping_stage',
        python_callable=collect_adoro_cinema_data,
        op_kwargs={
            'quantidade_paginas': 1,
        },
    )

    # definindo a tarefa para enviar o arquivo csv para S3.
    movie_review_to_s3_stage = PythonOperator(
        task_id='movie_review_to_s3_stage',
        python_callable=local_to_s3,
        op_kwargs={
            'bucket_name': '<nome-do-seu-bucket>'
        },
    )

    movie_review_remove_local = PythonOperator(
        task_id='movie_review_remove_local',
        python_callable=remove_local_file,
    )

    # Fim da Pipeline
    end_of_data_pipeline = DummyOperator(task_id='end_of_data_pipeline', dag=dag)

Assim, nós temos nossa DAG pronta, ela possui 2 tarefas Dummy, ou seja apenas para representação de início e fim do fluxo, e 3 tarefas que executam funcionalidades em código Python. Cada tarefa criado também possui um ID único para identificação, é importante esses IDs também serem únicos pois de outro modo o Airflow vai retornar um erro informando que foi identificado a criação de ciclos.

Para finalizar, nós precisamos criar a ligação entre as tarefas e definir a ordem de execução, o Airflow permite algumas formas de declaração dessa ordem, para esse tutorial vamos seguir o formato mais simples.

start_of_data_pipeline >> movie_review_web_scraping_stage >> movie_review_to_s3_stage >> movie_review_remove_local >> end_of_data_pipeline

Nas nossas tasks PythonOperator, nós realizamos a chamada de 3 funções python: collect_adoro_cinema_data, local_to_s3 e remove_local_file. A primeira encapsula o Web Scraper e a escrita dos dados coletados em um arquivo .csv numa pasta temporária, já a segunda e a terceira são as funções responsáveis por integrar com o S3 e enviar os arquivos, e então remover o arquivo .csv do disco.

import os
import glob
from airflow.hooks.S3_hook import S3Hook

def local_to_s3(bucket_name, filepath='./dags/data/*.csv'):
    s3 = S3Hook()

    for f in glob.glob(filepath):
        key = 'movie_review/'+f.split('/')[-1]
        s3.load_file(filename=f, bucket_name=bucket_name,
                    replace=True, key=key)

def remove_local_file(filepath='./dags/data/*.csv'):
    files = glob.glob(filepath)
    for f in files:
        os.remove(f)

Executando a DAG pelo Painel

Na tela inicial do Airflow nós iremos ver a listagem de todas as DAGs identificadas na aplicação, para executá-las basta arrastar o botão OFF para ON.

Nós podemos entrar no modo de visualização Grafo e verificar as nossas tasks e suas relações, elas vão alterando as cores de acordo com o status de execução, ao terminar a execução com sucesso suas bordas ficam verdes. No painel você também consegue identificar os tipos de Operators usados no fluxo.

O Airflow disponibiliza muitas formas de visualizar o fluxo executando, você pode acompanhar os graus de sucesso e falha por execução no modo Tree View ou pode acompanhar outros tipos de gráficos mostrando informações da execução. Você também tem acesso ao log de execução de cada tarefas, e acesso ao código da DAG, é um painel bastante completo para realizar monitoramento das tafegas criadas.

Por fim, ao final da execução das tarefas temos um arquivo .csv gerado com as informações de filmes obtidas no AdoroCinema e enviadas para o bucket S3 configurado.

Conclusão

Por fim, temos a primeira etapa do nosso projeto concluída, agora temos o S3 informações a respeito de filmes coletados na internet a partir do site AdoroCinema. Para a próxima etapa, nós iremos ler esse arquivo do S3 e realizar um pequeno processamento desses dados usando Apache Spark, a partir daí poderemos disponibilizar o resultado desses processamento em um Data Warehouse na Amazon Redshift.

Espero que tenham gostado, qualquer dúvida pode deixar nos comentários. Até a próxima!

Conhecendo o ecossistema Spark

Mayara Machado — Tue, 06 Oct 2020 00:45:36 +0000

Spark é uma ferramenta para processamento distribuído e o maior projeto open source voltado para o processamento de dados do momento. Ele faz parte do ecossitema Hadoop, ou seja, pode ser utilizado rodando em cima da estrutura Hadoop e também pode utilizar a estrutura do HDFS e do MapReduce para otimizar consultas em grandes quantidades de dados.

Além disso seu uso é muito interessante pois ele permite também realizar a aplicação de algoritmos de Machine Learning em cima de jobs MapReduce, realizar processamento de grafos e stream de dados. Vale ressaltar que o Spark também pode ser usado no modo stand alone e processar dados das mais variadas fontes, como banco de dados como Cassandra e o HBase.

Ecossistema Spark

É composto pelo módulo responsável pela realização de Streaming, Spark SQL e Dataframes, MLlib e o GraphX. É possível utilizar com Java, Scala, Python e R.

Spark SQL e Dataframes

Como comentei logo acima, é possível estabelecer conexão com dados diretamenta de uma base de dados e processalos a partir daí, para isso o Spark conta com um módulo estruturado para tornar esse processamento muito mais eficiente. É possível realizar consultas SQL utilizando a estrutura do Hive e acelerando bastante o tempo de processamento em grandes quantidades de dados, uma integração muito poderosa e que torna o Spark tão especial para o processamento de big data.

Além disso o Spark possui seu próprio módulo para manipulação de Dataframes, que nada mais são do que estruturas que se assemelham a planilhas e permitem a manipulação de dados durante análise e exploração dos mesmos. Para aqueles mais acostumados com o módulo de dataframe do Pandas, pode estranhar algumas diferenças em como cada um desses frameworks possibilitam a manipulação, mas nada que não seja maleável.

Spark Streaming

Algumas soluções necessitam possuir a habilidade de processar e analisar dados em tempo real, se diferenciando da estrutura para consumo em batch de uma base de dados. Para isso, o Spark possui um módulo inteiro voltado para ingestão de dados em stream, permitindo integrações com ferramentas como o Kafka e Flume. Dessa forma você pode utilizar o poder de processamento e aplicação de algoritmos de Machine Learning, a ingestão de dados em tempo real e aproveitar que o Spark possui uma boa tolerância a falhas .

MLlib

É a biblioteca de Machine Learning do ecossistema Spark. Ele possui em sua biblioeteca a aplicação dos algoritmos de classificação, regressão, clusterização, filtro colaborativo, redução de dimensionalidade.

É também um cluster para computação de baixa latência, e permite o processamento de um grande volume de dados, sendo 100x mais rápido que o MapReduce, pois o Spark utiliza os dados em memória enquanto o MapReduce utiliza em disco. Além de ter essa otimização, não devemos esquecer que o Spark roda seus algoritmos em cima do MapReduce, portanto ele consegue aproveitar de ambas as partes.

Já existem vários algoritmos implementados, tantos de machine learning como também de estatística mais simples. Extração de caracteríristicas, transformação de dados como por exemplo de coluna para vetores para uso em algoritmos. Funções de otimização, como métodos do gradiente.

GraphX

É o módulo reservado para criação, manipulação e tratamento de grafos e computação paralela voltada para grafos. O Spark já contém muito dos algoritmos mais comuns que utilizam grafos, como Page Rank, Connected component, Strongly connected components e outros mais.

PySpark

O Spark é escrito majoritariamente em Scala, mas permite integração utilizando outras linguagens de programação como o Python. O PySpark é uma API para dar suporte a utilização do Spark utilizando python.

É possível desfrutar dos módulos MLib, PySparkSQL e Graphframes. Onde é possível usurfruir dos benefícios de perfomance e tolerância a falhas do Spark no processamento e manipulação de dados SQL, de arquivos, em grafos e mais. O PySpark junta os benefícios do Spark às vantagens já existente na linguagem python para processamento de dados.

Conclusão

Nesse artigo eu trouxe um pouco sobre o que é o Spark, quais os módulos presentes nele e quais as possibilidades de uso. Mais a frente compartilharei alguns exemplos de uso de cada um desses módulos apresentados nesse texto.

Post originalmente postado no meu blog.

Analisando tweets sobre Black is King

Mayara Machado — Tue, 18 Aug 2020 21:28:55 +0000

Nesse texto explico os resultados obtidos da minha análise, mas você pode ver aqui como eu construí o código de coleta e análise dos sentimentos.

Eu sou uma grande fã do trabalho da Beyoncé e recentemente ela lançou seu álbum visual Black Is King. Infelizmente a plataforma de streaming onde o álbum foi lançado não está disponível no Brasil, logo eu ainda não pude ter acesso ao resultado produzido pela Beyoncé e sua equipe.

Mas há uma coisa eu posso fazer, e seria acompanhar os comentários nas redes sociais sobre o que as pessoas estão achando sobre. E se tem algo que não faltou foram comentários nas redes sociais sobre a obra da queen B, sendo eles alguns positivos e outros não tão positivos assim.

Então, movida pela curiosa e pela vontade de praticar um pouco a coleta de dados, resolvi realizar uma apanhado de comentários feitos sobre o álbum nos últimos 7 dias por usuários do Twitter falantes da língua inglesa. A escolha pela língua inglesa se deu, pois há poucos países onde essa plataforma de streaming Disney Plus está disponível e parte deles há falantes da língua inglesa.

Objetivos

Então eu queria realizar alguns processos:

Coletar uma amostra de tweets que tinham referenciado Black is King nos últimos 7 dias;
Identificar quais os países de origem desses tweets;
Identificar quais as palavras utilizados com maior frequência;
Qual o sentimento predominante dos tweets obtidos;
A distribuição dos tweets ao longo dos dias;

Para realizar essas tarefas eu irei utilizar a linguagem Python, juntamente a biblioteca Tweepy para realizar a coleta dos tweets, a biblioteca Pandas para fazer o tratamento dos dados, TextBlob para realizar uma rápida análise da polaridade dos tweets, e as bibliotecas Matplotlib e Folium para plotar os gráficos.

Obter os dados

Para fazer a obtenção, eu utilizei uma combinação de strings e hashtags com o título de Black is King em combinação com o nome da Beyoncé, o intuito foi coletar tweets que fizeram referência a esses dois tópicos. Eu escolhi fazer a coleta de uma amostra de 5000 tweets, o que ainda é uma quantidade pequena quando pensamos na quantidade de tweets que são compartilhados diariamente ao redor do mundo.

keyword = ("'Black is King' OR 'black is king' OR 'Beyonce' OR 'beyonce' OR #blackisking OR '#BlackIsKing' OR 'black is king beyonce'")

Resultados

Intervalo de tempo

Apesar de estipular o intervalo de tempo desejado, não necessariamente implicou em receber uma quantidade distribuída de tweets pelos dias solicitados. Podemos ver que grande parte dos tweets obtidos foram do dia 09 de agosto, e sendo muito pouco a quantidade de tweets dos dias mais próximos ao do desenvolvimento desse projeto, dia 12 de agosto.

Origem dos Tweets

Os países que concentraram a maioria dos tweets coletados foram os Estados Unidos e a África do Sul. Além deles, também foram coletados tweets de países da Oceania, Ásia e até do Brasil. Suponho que boa parte destes tweets estejam lamentando por ainda não poderem ter acesso, uma vez que a Disney Plus ainda está com uma lista bem restrita de cobertura de países.

Atualmente a Disney Plus possui disponibilidade nos seguintes países: Alemanha, Austrália, Áustria, Canadá, Espanha, Estados Unidos, França, Índia, Irlanda, Itália, Nova Zelândia, Países Baixos, Porto Rico, Reino Unido, Suiça e Japão.

Além disso, Black is King é uma obra musical e visual que conta com um mix de influências da cultura Africana e Afro-Americana. Beyoncé compartilha tela com cantoras como Busiswa da Africa do Sul, Salatiel de Camarões, Yemi Alade e Mr Eazi da Nigéria, sempre trazendo elementos na dança, nas roupas e também nos ambientes.

Algo interessante que puder observar na minha amostra foi ver a particição dos países da África ativos sobre o álbum. Mesmo dentro da minha amostra podemos ver a África do Sul como um dos pólos com maiores concentração de tweets, e também podemos ver outros menos intensos mas ainda sim presente como Camarões e Nigéria. Sendo esses três países, os países de onde originam alguns participantes do álbum visual.

Nuvem de palavras

As palavras Black, King, Beyonce são as que mais foram reproduzidas, o que é algo a se esperar uma vez que são as palavras referentes ao nome da obra e da artista da obra. Além dessas palavras, podemos visualizar algumas menções à filha da Beyoncé, Blue Ivy, e ao termo daughter filha em inglês. O termo em si pode estar fazendo referência tanto às aparições de Blue Ivy, como também a referência às filhas da artista canadense Ayesha Curry, seu nome também aparece na nuvem de palavras.

No sábado 08 de agosto Ayesha Curry publicou um vídeo de suas filhas dançando ao som de Beyoncé e reproduzindo a coreografia presente no album. Esse video se tornou viral no tarde de sábado, e podemos ver algumas menções a ele na nuvem de palavras.

Outra palavra a aparecer é Dance, fato este que não é de se espantar uma vez que Beyoncé é conhecida por suas coreografias muito bem desenvolvidas e perfomances impecáveis junto a sua equipe de dançarinos. Além disso também temos menções ao dançarino nigeriano Papi Ojo, que realiza uma perfomance ao lado da Beyoncé e é conhecido por perfomances com Rihanna e Janet Jackson.

Polaridade dos tweets

Como suspeitei inicialmente, mais da metade dos tweets continham textos com teor mais negativo. O lançamento de Black is King levantou muitos debates na internet sobre a forma de abordar elementos da cultura africana, é possível encontrar opiniões divididas sobre esse tema. Além disso, como dito mais acima, a Disney Plus ainda conta com uma distribuição bem limitada, o que faz com que fãs da Beyoncé preciserem esperar para poder assistirem. Também é possível encontrar comentários lamentando essa situação.

Conclusão

Foi uma experiência muito interessante poder ver de forma geral quais são os fatos que rodeiam o lançamento do álbum, ainda continuo com a expectativa de poder assistir e emitir a minha própria opinião nas redes sociais. Até lá, vamos ficar apenas com as análises do que tem sido dito lá no exterior. Até a próxima!

Coletando dados do Twitter com Python

Mayara Machado — Sat, 15 Aug 2020 21:38:38 +0000

Eu iniciei um projeto para realizar a coleta de alguns tweets utilizando a biblioteca Tweepy, tratamento de dados utilizando Python e análise de polaridade dos textos escritos nos tweets. Irei relatar um pouco do processo, do código e dos resultados que obtive na série de dois textos, inicialmente iria ser somente um texto mas no decorrer do processo da escrita achei que seria mais interessante dividir essas em etapas.

Dessa forma, temos esse primeiro momento onde eu mostro os códigos usados para coleta dos tweets e o código para tratar os dados. No segundo momento irei mostrar os resultados e algumas visualizações dos dados obtidos. Então vamos lá!

Coleta e limpeza dos tweets

A gente precisa ter, primeiramente, uma conta ativa no ambiente do twitter para desenvolvedores. Uma vez que sua solicitação é aceita e você já possue sua conta de desenvolvimento ativa, você precisa ir no Dashboard de sua conta e criar um App. Ao criar um App vinculado a sua conta, o Twitter irá gerar credenciais que serão usadas para realizar as chamadas à API deles, então guardem bem essas credencias e não compartilhem na internet.

Coleta dos tweets

Utilizando as credenciais que obtivemos, a gente já pode estabelecer a conexão através da Tweetpy.

import tweetpy

CONSUMER_KEY = YOUR_CONSUMER_KEY
CONSUMER_SECRET = YOUR_CONSUMER_SECRET
ACCESS_TOKEN = YOUR_ACCESS_TOKEN
ACCESS_TOKEN_SECRET = YOUR_ACCESS_TOKEN_SECRET

auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

conToken = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=5, retry_delay=10)

Agora já com a conexão estabelecida, nós iremos escrever o código para obter os tweets. Para a proposta em questão, o interessante seria buscar tweets públicos através de uma hashtag ou palavra chave mesmo. O Twitter possui alguns endpoints disponíveis, nós podemos realizar coletas diretamente em perfis de usuários, no nosso próprio perfil ou utilizar o endpoint de pesquisa.

Caso esteja na dúvida de qual endpoint é o mais interessante para o seu projeto, eu recomento que seja realizado uma leitura na documentação da API do Twitter. O Tweepy já possui a implementação para realizar a conexão com esse endpoint, para isso nós precisamos selecionar apenas alguns parâmetros que serão utilizados pelo Tweepy no momento da busca.

tweets_iter = tweepy.Cursor(conToken.search,
                            q=KEYWORD, tweet_mode='extended',
                            rpp=5000, result_type='mixed',
                            since=datetime(2020,7,31,0,0,0).date(),
                            lang='en', include_entities=True).items(count)

Vamos dar uma olhada nos parâmetros utilizados.

tweet_mode='extended' : passamos esse parâmetro para explicitar que nós queremos receber o texto completo do tweet, caso esse parâmetro não seja passado os tweets retornados terão até no máximo 140 caracteres de texto.
rpp=5000 : é a quantidade limite que nós queremos receber de tweets, em algumas requisições por questões de timeout, dependendo da quantidade que você utilizar, é possível que seja retornado menos que a quantidade solicitada.
result_type='mixed': É possível receber 3 tipos de tweets, os recentes, os que tiveram maiores engajamentos como curtidas e retweets, ou uma mistura de ambos. Para definir isso basta especificar o result_type dentre ('mixed', 'recent', 'popular'), caso esse campo não seja passado o default é sempre o mixed. Decidi explicitar aqui nesse caso qual era a minha intenção.
since=datetime(2020,7,31,0,0,0).date(): É possível também delimitar o tempo com os parâmetros since e until, no meu caso eu só delimitei a data a partir pois queria apenas a partir de uma data em específico.

Limpeza e preparo dos tweets

Após obtenção de um objeto Iterator no código anterior, nós vamos iterar para acessar cada tweet obtido e realizar a limpeza. Inicialmente precisamos tratar os textos dos tweets, é nele que vamos realizar a extração de itens que não são necessariamente interessantes para a nossa análise.

Para essa remoção, eu utilizei regex simples que removem links que podem estar associados ao tweet, removem as menções a outros usuários. Além disso, a requisição retorna muitos dados sobre o tweet e não precisamos de tudo isso. Nesse exemplo eu deixei algumas colunas que podem ser interessantes para análises futuras.

import re

def clean_tweet(self, tweets_text):

    clean_text = re.sub(r'RT+', '', tweets_text) 
    clean_text = re.sub(r'@\S+', '', clean_text)  
    clean_text = re.sub(r'https?\S+', '', clean_text) 
    clean_text = clean_text.replace("\n", " ")

    return clean_text

def prepare_tweets_list(self, tweets_iter):

  tweets_data_list = []
  for tweet in tweets_iter:
    if not 'retweeted_status' in dir(tweet):
      tweet_text = clean_tweet(tweet.full_text)
      tweets_data = {
          'len' : len(tweet_text),
          'ID' : tweet.id,
          'User' : tweet.user.screen_name,
          'UserName' : tweet.user.name,
          'UserLocation' : tweet.user.location,
          'TweetText' : tweet_text,
          'Language' : tweet.user.lang,
          'Date' : tweet.created_at,
          'Source': tweet.source,
          'Likes' : tweet.favorite_count,
          'Retweets' : tweet.retweet_count,
          'Coordinates' : tweet.coordinates,
          'Place' : tweet.place 
      }

      tweets_data_list.append(tweets_data)

  return tweets_data_list

Polaridade dos tweets

Por fim, nós também podemos utilizar a biblioteca TextBlob para realizar a análise de polaridade nos tweets que obtivemos. O TextBlob recebe uma string, que pode ser uma palavra ou uma frase completa, e retorna um inteiro no intervalo entre 0 e 1. Esse inteiro diz respeito a polaridade encontrada pela biblioteca.

Para melhorar a visualização e entendimento, eu realizei a substituição do valor da polaridade por uma string indicando em qual espectro aquele tweet se encaixava. Dessa forma valores menores que zero receberam Negative, valores maiores que zero receberam Positive. As vezes a biblioteca não consegue determinar uma polaridade a partir da string e retorna o valor zero, para esses casos eu considerei como Neutral.

from textblob import TextBlob

def sentiment_polarity(self, tweets_text_list):
      tweets_sentiments_list = []

      for tweet in tweets_text_list:
        polarity = TextBlob(tweet).sentiment.polarity
        if polarity > 0:
          tweets_sentiments_list.append('Positive')
        elif polarity < 0:
          tweets_sentiments_list.append('Negative')
        else:
          tweets_sentiments_list.append('Neutral')

      return tweets_sentiments_list

Código final

Por fim, nós podemos organizar todo esse código escrito até então em uma única classe, a classe responsável pelos tweets que realiza coleta, tratamento de tweets.

import re
import tweepy
from datetime import datetime
from textblob import TextBlob

CONSUMER_KEY = YOUR_CONSUMER_KEY
CONSUMER_SECRET = YOUR_CONSUMER_SECRET
ACCESS_TOKEN = YOUR_ACCESS_TOKEN
ACCESS_TOKEN_SECRET = YOUR_ACCESS_TOKEN_SECRET

keyword = ("'Black is King' OR 'black is king' OR 'Beyonce' OR 'beyonce' OR #blackisking OR '#BlackIsKing' OR 'black is king beyonce'")
count = 5000

class TweetAnalyzer():

  def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret):
    '''
      Conectar com o tweepy
    '''
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)

    self.conToken = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=5, retry_delay=10)

  def __clean_tweet(self, tweets_text):
    '''
    Tweet cleansing.
    '''

    clean_text = re.sub(r'RT+', '', tweets_text) 
    clean_text = re.sub(r'@\S+', '', clean_text)  
    clean_text = re.sub(r'https?\S+', '', clean_text) 
    clean_text = clean_text.replace("\n", " ")

    return clean_text

  def search_by_keyword(self, keyword, count=10, result_type='mixed', lang='en', tweet_mode='extended'):
    '''
      Search for the twitters thar has commented the keyword subject.
    '''
    tweets_iter = tweepy.Cursor(self.conToken.search,
                          q=keyword, tweet_mode=tweet_mode,
                          rpp=count, result_type=result_type,
                          since=datetime(2020,7,31,0,0,0).date(),
                          lang=lang, include_entities=True).items(count)

    return tweets_iter

  def prepare_tweets_list(self, tweets_iter):
    '''
      Transforming the data to DataFrame.
    '''

    tweets_data_list = []
    for tweet in tweets_iter:
      if not 'retweeted_status' in dir(tweet):
        tweet_text = self.__clean_tweet(tweet.full_text)
        tweets_data = {
            'len' : len(tweet_text),
            'ID' : tweet.id,
            'User' : tweet.user.screen_name,
            'UserName' : tweet.user.name,
            'UserLocation' : tweet.user.location,
            'TweetText' : tweet_text,
            'Language' : tweet.user.lang,
            'Date' : tweet.created_at,
            'Source': tweet.source,
            'Likes' : tweet.favorite_count,
            'Retweets' : tweet.retweet_count,
            'Coordinates' : tweet.coordinates,
            'Place' : tweet.place 
        }

        tweets_data_list.append(tweets_data)

    return tweets_data_list

  def sentiment_polarity(self, tweets_text_list):
      tweets_sentiments_list = []

      for tweet in tweets_text_list:
        polarity = TextBlob(tweet).sentiment.polarity
        if polarity > 0:
          tweets_sentiments_list.append('Positive')
        elif polarity < 0:
          tweets_sentiments_list.append('Negative')
        else:
          tweets_sentiments_list.append('Neutral')

      return tweets_sentiments_list

Agora basta realizar as chamadas das classes para desfrutar das funções.

analyzer = TweetAnalyzer(consumer_key = CONSUMER_KEY, consumer_secret = CONSUMER_SECRET, 
access_token = ACCESS_TOKEN, access_token_secret=ACCESS_TOKEN_SECRET)

# Para realizar as buscas usando a keyword e quantidade predefinida e converter em uma lista.
tweets_iter = analyzer.search_by_keyword(keyword, count)
tweets_list = analyzer.prepare_tweets_list(tweets_iter)

# Uso de um Dataframe para melhor manipulação
tweets_df = pd.DataFrame(tweets_list)

# Chamar a função para análise de sentimentos
tweets_df['Sentiment'] = analyzer.sentiment_polarity(tweets_df['TweetText'])

Uma vez que temos uma estrutura Dataframe com essas informações, nós podemos explorar outras tópicos também como abordei neste post aqui. Além disso, você pode ver o notebook com código desenvolvido nesse repositório.

Considerações finais e próximos passos

No texto de hoje eu demonstrei como podemos realizar a coleta de tweets, tratamento dos tweets e análise deles utilizando Python, Tweepy e TextBlob. Essas bibliotecas são muito interessantes para quem está iniciando com a coleta e processamento de tweets, são muito práticas e permitem uma introdução interessante para atividades envolvidas no processo de coleta e análise de dados.

No próximo texto irei compartilhar quais foram os resultados obtidos através dessa coleta. Até lá, pessoal!

AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?

Mayara Machado — Mon, 22 Jun 2020 14:03:48 +0000

Esse texto faz parte de uma série sobre a utilização da AWS SQS e SNS em uma aplicação python, sugiro a leitura dos textos anteriores para maior entendimento do processo.

AWS SQS/SNS | Parte 1 | Implementando Mensageria Pub/Sub com Python
AWS SQS/SNS | Parte 2 | Publicando uma mensagem via SNS
AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python
AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?

Introdução

Esta é a parte final de uma série de 4 textos onde eu falo sobre como fazer uso das filas e tópicos disponibilizados pelos serviços SQS e SNS da AWS. Hoje iremos conhecer uma pouco sobre filas que servem para envio de mensagens mortas, iremos ver quais são algumas das possíveis causas de uma mensagem morta e como podemos preparar nosso serviço SQS para lidar com essas mensagens nos casos de erros. Ao final, iremos ver como podemos implementar uma Dead Letter Queue utilizando python e a lib Boto3 que é a SDK da AWS para python.

O que são Dead Letter Queue ou Fila de Mensagens Mortas

De modo geral, quando um serviço está processando uma mensagem da fila e tem esse processamento interrompido por motivos de erro ou até mesmo de instabilidade do serviço, esta mensagem se torna uma mensagem morta. Durante essa ocorrência é muito possível que o serviço perca a mensagem, ou até mesmo que ela seja corrompida e portanto perca-se o acompanhamento de quais mensagens conseguiram de fato serem consumidas e processadas. Nesse cenário, surgem as Dead Letter Queue que ,a grosso modo, são filas que armazenam as mensagens mortas.

Quando uma mensagem morre durante algum erro no processamento, a fila principal é capaz de redirecionar esta mensagem para a Dead Letter Queue vinculada a fila principal. As filas principais não precisam obrigatoriamente ter uma fila de mensagens mortas, ainda que seja uma configuração recomendada, e as filas de mensagens mortas não são criadas automaticamente pela AWS SQS. Para atribuirmos uma Dead Letter Queue a uma fila principal, devemos criar uma política de redirecionamento de mensagens na fila principal e atribuir o vínculo desta a uma segunda fila que irá agir como a fila de mensagens mortas para a fila principal.

Existem outras causas que podem resultar no envio de mensagens para uma Dead Letter Queue, uma delas ocorre quando a fila principal recebe uma mensagem mais vezes do que o limite definido no atributo maxReceiveCount da fila principal. Por exemplo, em filas normais é possível que mais de uma cópia da mensagem seja entregue na fila, nesse caso se temos o maxReceiveCount da fila principal definido como 3 e recebermos uma mensagem 4 vezes, e suas cópias não forem excluídas, essa mensagem será enviada para a Dead Letter Queue configurada.

O grande benefício de uma Dead Letter Queue é poder obter conhecimento sobre o comportamento inexperado da aplicação ao tentar processar a mensagem. É possível definir um alarme para informar quando uma mensagem morta chega na fila, além disso, é possível depurar a mensagem para identificar problemas no envio e/ou recebimento de mensagens, assim como também é possível com auxílio de log identificar qual exceção está causando o redirecionamento daquela mensagem.

Condições para ter uma Dead Letter Queue

É possível ter várias filas principais e que todas elas apontem para uma mesma fila de mensagens mortas, desse modo, a quantidade de para das filas irá depender bastante da definição da arquitetura do seu projeto. Porém você deve se atentar para duas questões, a fila de mensagens mortas de uma fila tipo FIFO também deve ser uma fila FIFO e as filas principais e suas respectivas filas de mensagens mortas devem estar na mesma região, assim como serem criadas com a mesma conta.

No primeiro caso, devemos lembrar que a ordem de envio é muito importante em uma fila do tipo FIFO, e uma fila Dead Letter Queue do tipo padrão não irá manter a ordem das mensagens mesmo que a fila principal de onde se origina a mensagem seja uma do tipo FIFO. Para o segundo caso , é uma determinação da AWS e portanto é necessário prestar atenção a esse detalhe na hora de monstar a estrutura de filas do seu projeto.

Por fim, outro detalhe a ter em mente é que todas as mensagens tem um tempo de expiração nas filas. Quando criamos uma fila é possível configurar o tempo padrão da vida de uma mensagem na fila, por default a AWS configura o período de permanência da mensagem por 4 dias, porém esse valor é totalmente configurável. Quando uma mensagem é redirecionada para uma fila de mensagens morta, ela não terá o contador do seu tempo de permanência reiniciado para a nova fila. Por exemplo, se uma mensagem é encaminhada para a fila de mensagens morta no seu terceiro dia de vida, e esta fila de mensagens morta possui uma definição de retenção de mensagens de 4 dia, a fila irá excluir essa mensagem após 1 dia, pois seu tempo de vida chegou ao limite do definido no tempo de retenção da Dead Letter Queue.

Implementando

Para implementarmos uma Dead Letter Queue, utilizaremos o conhecimento já adquirido anteriormente que é criar uma fila padrão, no entanto, o ponto de atenção na hora de criar essa fila é justamente o tempo de retenção da mensagem a ser definido. No exemplo desenvolvido, eu defini 6 dias em segundos (o Boto3 solicita que esses atributos sejam definidos em segundos) pois o tempo de renteção da minha fila principal é de 4 dias.

Para atribuir uma fila como Dead Letter Queue a uma fila padrão, devemos criar uma política de redirecionamento chamada RedrivePolicy e atribuir o código ARN da fila que será a nossa Dead Letter Queue. No código de exemplo eu montei a código ARN com base nas minhas informações pessoais da conta AWS, só exemplificando para vocês um outro método de obter o código ARN de uma fila.

import boto3, json

class DeadLetterQueue:

    def __init__(self, main_queue_url):
        self.main_queue_url = main_queue_url
        self.sqs_client = boto3.client(
                            'sqs', 
                            region_name='us-east-1'
                        )

    def __create_dead_letter_queue(self, queue_name):
        '''
            Create a dead letter.
        '''
        dead_letter_queue = self.sqs_client.create_queue(
             QueueName = ,
             Attributes={
                    "MessageRetentionPeriod" : "518400"  # 6 dias
                }
            )

        return dead_letter_queue


    def add_dead_letter_queue(self, queue_name, region, aws_id):

             # criar a fila para ser a Dead Letter Queue
             self.__create_dead_letter_queue(queue_name)

             # politica de redirecionamento para a fila recém criada
        redrive_policy = {
            "deadLetterTargetArn" : (f"arn:aws:sqs:{region}:{aws_id}:{queue_name}"),
            "maxReceiveCount" : "10"
        }

             # atualização da fila principal para atribuição da política de redirecionamento
        self.sqs_client.set_queue_attributes(
            QueueUrl= self.main_queue_url,
            Attributes={
            'RedrivePolicy' : json.dumps(redrive_policy)
            }
        )

        return True

Conclusão

Bom pessoal, com isso nós encerramos a série sobre como iniciar a utilizar as filas e tópicos disponibilizadas pelos serviços SQS e SNS da Amazon Web Services. Foi muito bom interessante ter tido essa experiência, e esses serviços se mostraram muito interessantes para uso como solução de comunicação para serviços em arquiteturas de Microserviços ou até mesmo uma arquitetura de Serviços. Obrigada por terem acompanhado até aqui, e vejo vocês na próxima.

Lembrando que o código completo poderá ser visualizado aqui.

AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python

Mayara Machado — Thu, 18 Jun 2020 21:18:07 +0000

Esse texto faz parte de uma série sobre a utilização da AWS SQS e SNS em uma aplicação python, sugiro a leitura dos textos anteriores para maior entendimento do processo.

AWS SQS/SNS | Parte 1 | Implementando Mensageria Pub/Sub com Python
AWS SQS/SNS | Parte 2 | Publicando uma mensagem via SNS
AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python
AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?

Introdução

Na parte 3 dessa série iremos ver como consumir as mensagens publicadas via SNS mostradas no texto anterior. Nessa parte do projeto eu utilizei Django e Django Rest Framework para estruturar o projeto, utilizei o SDK da AWS em python que é a Boto3 e utilizei o Celery juntamento ao Redis para criar agendamento de tarefas.

Lembrando que o código desenvolvido nesse projeto poderá ser encontrado aqui.

Implementação do projeto

Consumir a mensagem via SQS

Primeiro iremos criar a função que irá realizar a chamada de execução do consumo da fila SQS para onde disparamos a mensagem através do tópico.

def poll_queue_for_messages(self, queue_url, max_messages_number=10):
        return self.sqs_client.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=max_messages_number
        )

O comportamento padrão dessa função é de realizar o modo Short Poll, em outras palavras ela retorna uma amostra das mensagens contidas na fila ao invés de trazer todas as mensagens encontradas. Desse modo a quantidade de mensagens disponíveis na fila tem implicação direta na quantidade de mensagens obtidas por essa função, podendo até retornar vazio caso a quantidade da fila seja extremamente pequena, por esse motivo é importante realizar tentativas periodicamente para obter o real estado da fila.

É possível passar algumas especifacações nessa requisição, para maiores conhecimento recomendo fortemente a leitua da documentação do boto3 que é incrivelmente bem descrita.

Com a função que realiza a conexão com AWS feita, vamos para a task periódica que irá chamar a função anterior. Ela basicamente irá verificar se houve retorno e caso tenha tido iterar sobre ele e obter a mensagem para realizar o processamento daquela mensagem. Importante lembrar que devemos excluir a mensagem uma vez que ela tenha sido devidamente processada, pois caso ela persista poderá ocasionar erros como processamento duplicado e isso é algo que não queremos deixar ocorrer, imagina o problema de uma cobrança realizada duplicada? Nem pensar.

Outro problema que pode ocorrer é algum erro durante o processamento ou até mesmo a indisponibilidade do serviço que não permita que aquela mensagem seja completamente processada, para essas ocasiões não devemos permitir que a aplicação remova a mensagem da fila, um recurso que ainda não utilizamos aqui mas que é bastante útil para a recuperação das mensagens nesses casos é a Dead Letter Queue que abordaremos no próximo texto da série.

Mas voltando para o cenário que imaginamos o serviço ecommerce estará constantemente enviando mensagens para a nossa fila. Por isso o nosso serviço payment gateway precisa ficar fazendo verificações constantes na fila para que ele possa atualizar os dados da sua base conforme os dados enviados pelo serviço ecommerce enquanto processa novos pedidos. Além disso, como a função poll_queue_for_messages não retorna todas as mensagens a cada chamada, devemos fazer chamadas repetidas para capturar mensagens ainda não processadas.

Então para consumir as mensagens de forma periódica utilizei o Celery, com ele eu pude configurar para que a função de consumo da fila seja executada de minuto em minuto. Seria possível implementar esse agendamento com a AWS CloudWatch, mas como não era o foco desse projeto utilizei o Celery por facilidade de execução. Para ter o código funcionando bastou apenas realizar a instação da lib para uso no Django, e configurar o agendamento da task na base de dados usando o celery beat.

@task()
def verify_transaction_queue():
    billing_domain = BillingDomainService()
    sqs_connection = SQSConnection()

    queue_messages = sqs_connection.poll_queue_for_messages(settings.TRANSACTION_URL_QUEUE)
    if 'Messages' in queue_messages and len(queue_messages['Messages']) >= 1:
        for message in queue_messages['Messages']:
            message_body = message['Body']
            receipt_handle = message['ReceiptHandle']

            success = billing_domain.process_purchase(message_body)
            if success:
                sqs_connection.delete_message_from_queue(settings.TRANSACTION_URL_QUEUE, receipt_handle)

Conclusão e implementações futuras

No texto de hoje vimos como realizar o consumo de mensagens disponibilizadas nas filas SQS. O próximo texto será o último dessa série sobre SQS e SNS, e nele veremos o que são as Dead Letter Queues. Aguardo você lá!

AWS SQS/SNS | Parte 2 | Publicando uma mensagem via SNS

Mayara Machado — Sun, 14 Jun 2020 01:34:25 +0000

Este texto faz parte de uma série, veja também os outros textos dessa série:

Implementando Mensageria Pub/Sub com Python, AWS SNS e SQS | Parte 1
Publicando uma mensagem utilizando AWS SNS | Parte 2
AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python
AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?

Introdução

Dando continuidade na série sobre o uso da AWS SQS e SNS, hoje implementaremos a funcionalidade para publicação de mensagens utilizando SNS, e como utilizar o recurso de filtros dos serviços SNS. De modo mais introdutório, resolvi fazer duas aplicações REST utilizando Django e Django Rest Framework, que serão os serviços de ecommerce e payment_gateway que não possuem comunicação direta e realizam comunicação utilizando tópicos e filas.

No geral essas aplicações representam operações e processamentos que podem ocorrer em serviços mais completos, os serviços são para simular API's de ecommerce. Quero ressaltar que este artigo não tem o propósito de explicar o funcionamento do Django Rest Framework, por esse motivo o código desenvolvido nesse projeto foi o mais simples possível e apenas funcional o suficiente para a utilização das funcionalidades de fila necessária.

Pois bem, vamos começar.

Implementação do projeto

Nesse projeto exemplo, imaginei um cenário onde temos um serviço responsável pelo processamento da API ecommerce, nesse serviço teremos a rota GET de listagem de produtos disponíveis e qual o vendedor daquele produto, também teremos uma rota POST para criação de um pedido de compra. Na outra ponta temos o serviço responsável pelo processamento do pagamento, o que eu chamei de payment gateway, esse serviço irá consumir a quantidade de pedidos processados pelo serviço ecommerce, e irá através de uma rota GET exibir a quantidade arrecadada de pedido para cada vendedor.

O código desenvolvido nesse projeto poderá ser encontrado aqui.

Funcionalidades do serviço

Quando o serviço recebe a solicitação de pedido de compra, a aplicação realiza os processamentos que precisa para a sua execução. Uma vez que o processo foi completado, a aplicação irá disparar uma mensagem para o tópico SNS.

def create(self, purchase_data):
        purchase = Purchase()

        products_pk = [product["id"] for product in purchase_data['products']]
        products_list = list(self.product_domain.get_all(query_params={'pk__in': products_pk}))
        purchase.total_price = self.__calculate_total_price(products_list)

        # salvar o objeto
        purchase = self.repository.create(purchase)

        # salvar os relacionamento ManyToMany
        purchase.product.add(*products_list)
        self.repository.update_m2m(purchase)

        # disparar mensagem para tópico SNS
        self.__notificate_transaction_sns(purchase)

        return purchase

A mensagem que a aplicação irá enviar vai ser um json contendo o identificador do vendedor, a data da compra, o valor total que aquela compra para aquele lojista e um identificador indicando a qual compra esses dados se referem.

def __notificate_transaction_sns(self, purchase, seller_id):
        products = purchase.product.all()
        message = {
                    "seller" : str(seller_id),
                    "purchase_date" :  purchase.created_at.strftime("%Y-%m-%d %H:%M:%S %Z"),
                    "purchase_id" : str(purchase.pk),
                    "total_price" : str(purchase.total_price)
        }

        # Chama a classe de domínio que faz conexão com a AWS SNS
        self.sns_connection.publish_message_to_subscribers(settings.TRANSACTION_TOPIC_SNS, json.dumps(message))

Envio da mensagem via SNS

Publicar uma mensagem em um tópico SNS é muito simples, para uma configuração mínima você precisa passar apenas o código ARN do tópico e a mensagem em questão. A mensagem pode estar em qualquer padrão definido para o projeto, nesse exemplo estou utilizando o envio de um json como mostrado anteriormente.

Vale lembrar que o código ARN do tópico foi obtido durante a criação do mesmo, o código de criação do tópico e da instanciação do cliente SNS está disponível na Parte 1 dessa série.

def publish_message_to_subscribers(self, topic_arn, message, message_attributes={}):
        publish = self.sns_client.publish(
                                            TopicArn=topic_arn,
                                            Message=message,
                                            MessageAttributes = message_attributes
                                        )
        return publish

Esse código é tudo o que você precisa para poder enviar uma mensagem para todos os incritos do tópico, lembrando também que é possível ter contas de email e números de telefone como inscritos de um tópico, desse modo todos eles receberiam igualmente a mensagem disparada sem haver a necessidade de adicionar um parâmetro a mais no envio. Como visto na parte 1 dessa série, apenas adicionei uma fila SQS como um inscrito desse tópico, logo ela foi quem recebeu a mensagem que foi disparada por esse trecho de código e agora nós iremos consumi-la.

Filtrando o envio de mensagens

Mas lembram que eu falei anteriormente que um tópico pode ter n inscritos e que eles podem ser do tipo SQS, email ou SMS? Imagine que no nosso tópico ao invés de uma única Queue, a gente tivesse duas Queue inscrita e aptas para receberem mensagens, e vamos imaginar também que na verdade gostaríamos nesse cenário de mandar as mensagens para uma Queue em específica. Isso é possível através da utilização dos filtros SNS, mas como funciona?

Toda vez que é disparado o envio de mensagens através da SNS, por padrão o serviço já verifica se existe uma política de filtro existente para algum inscrito, caso tenha ela verifica se o filtro descrito nos atributos da mensagem bate com algum dos filtros encontrados naquele tópico. Nos casos onde não há filtro, a SNS dispara a mensagem para todos os seus inscritos sem ter nenhuma diferenciação. A configuração do filtro é atribuído no momento da inscrição, desse modo, voltando ao código publicado no texto anterior iremos pegar a váriavel payment_subscription e configurar essa inscrição.

sns_cliente.set_subscription_attributes(
    SubscriptionArn = payment_subscription['SubscriptionArn'],
    AttributeName ='FilterPolicy', 
    AttributeValue ='{"event_type": ["order_placed"]}'
)

No código acima, a gente chama novamente o cliente sns instanciado utilizando o boto3 e chamamos a função para atualizar os atributos da inscrição. Adicionaremos um novo atributo para essa inscrição em específico, e a gente garante isso através do código Arn da inscrição da fila PaymentQueue, que é a fila que vai ser consumida pelo serviço PaymentGateway do meu exemplo. O atributo em questão é uma política de filtro, onde adicionamento um dicionário contendo a chave sinalizando que é um tipo de evento e o seu valor é um array contendo o nome dos filtros aplicados.

Vale ressaltar que esse array pode conter ter 1 ou mais especificações de filtro, e uma mensagem será enviada para esta inscrição caso o atributo da mensagem seja correspondente com ao menos 1 filtro atribuído. A comparação realizada pela AWS é booleana, ou seja, ou é exatamente igual ou não, dessa forma os filtros são case sensitve.

Para enviar uma mensagem com um filtro em específico, nós iremos reescrever a função de notificate_transaction_sns para atribuirmos a especificação do filtro nos atributos da mensagem.

def __notificate_transaction_sns(self, purchase, seller_id):
        products = purchase.product.all()
        message = {
                    "seller" : str(seller_id),
                    "purchase_date" :  purchase.created_at.strftime("%Y-%m-%d %H:%M:%S %Z"),
                    "purchase_id" : str(purchase.pk),
                    "total_price" : str(purchase.total_price)
        }

        message_attributes = {
            'event_type': {
                'DataType': 'String',
                'StringValue': 'order_placed'
            }
        }

        self.sns_connection.publish_message_to_subscribers(settings.TRANSACTION_TOPIC_SNS, json.dumps(message), message_attributes)

E voilá, a mensagem será enviada apenas para o inscrição que possua o filtro order_placed atribuída.

Conclusão e implementações futuras

Na parte 2 dessa série nós vemos como publicar mensagens através do serviço SNS, vimos como pode ser absurdamente prático e vimos também como utilizar os filtros ao nosso favor. Nos próximos textos da série Implementando AWS SQS e SNS iremos ver como consumir as mensagens que enviamos hoje, e vamos descobrir também sobre o conceito das Dead Letter Queue. Vejo você lá!

AWS SQS/SNS | Parte 1 | Implementando Mensageria Pub/Sub com Python

Mayara Machado — Mon, 08 Jun 2020 00:28:49 +0000

Este texto faz parte de uma série, veja também os outros textos dessa série:

AWS SQS/SNS | Parte 1 | Implementando Mensageria Pub/Sub com Python
AWS SQS/SNS | Parte 2 | Publicando uma mensagem via SNS
AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python
AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?

Introdução

Eu decidi me desafiar e aprender algo novo. Nesse projeto eu quis estudar e explorar uma tecnologia que eu ainda não utilizava, então decidi me desafiar a implementar um pequeno projeto utilizando os conceitos aprendidos. Para narrar minha jornada resolvi escrever uma série de textos compartilhando o que eu tenho aprendido, ao longo dessa série irei implementar o padrão Publish/Subscribe utilizando dois serviços Rest que são desacoplados. O link para o repositório onde o projeto está em desenvolvimento está aqui.

As tecnologias escolhidas foram a Amazon Simple Queue Service (AWS SQS) e a Amazon Simple Notification Service (AWS SNS), a serem utilizadas em APIs construídas com Django Rest Framework e utilizando a lib Boto3 que é uma Amazon Web Services SDK para python.

Com a tecnologia decidida e uma ideia do que eu iria desenvolver, agora é hora de começar o projeto.

Sobre AWS SQS e SNS

Mas não poderia começar sem mergulhar antes em alguns conhecimento teóricos sobre a SQS e a SNS, a amazon contém muito material de documentação disponível o que facilitou bastante para um entendimento inicial. Mas para quem não está tão familiarizado com essas tecnologias da AWS calma que eu vou dar uma pincelada no que se trata esses serviços.

O que seriam a Amazon SNS e a Amazon SQS? De forma geral a SNS é o serviço da Amazon para publicação de notificações/mensagens para grandes números de endpoints. Ele é muito útil para aplicações de microserviços e/ou sistemas distribuídos, pois ajudam a garantir o desacoplamento entre os serviços ao mesmo tempo em que permitem que se estabeleça uma comunicação entre eles.

Os canais para envio de mensagens criados com a SNS são chamados de Tópicos e cada tópico pode ter n inscrições, uma inscrição representa o endpoint que receberá as mensagens que são enviadas naquele tópico. Um endpoint no SNS pode ser um usuário final através de um email ou sms, assim como também pode ser outro serviço da AWS como o Lambda ou AWS SQS.

A AWS SQS vem para completar esse sistema de mensageria, como o próprio nome já indica a SQS é um serviço de filas que também ajuda a garantir o desacoplamento. Esse serviço permite que seja enviado, armazenado e recebido mensagens de forma assíncrona entre aplicações, sem que haja perda de mensagens ou que seja necessário ter o sistema disponível.

O serviço SQS implementa filas normais que suportam quantidade quase que ilimitada de transações por segundo, mas em contrapartida não tem garantia de que serão entregues por ordem de chegada, e as vezes é possível que as mensagens sejam entregues mais de uma vez. A outra implementação possível são as filas FIFO que seguem o princípio First in First out, são filas onde as mensagens seguem estritamente a ordem nos seus envios e recebimentos, assim como também há a garantia de que as mensagens serão entregues uma única vez e estarão disponíveis aguardando o seu processamento e exclusão. Porém as filas FIFO tem uma taxa limitada de mensagens por segundo, suportando em média 300 operação de mensagem (envio, recebimento ou exclusão) por segundo.

No fim a decisão de que tipo de fila você irá implementar irá depender inteiramente de qual a necessidade do seu projeto, as vezes seu projeto precisa que muitas mensagens sejam disparadas por segundo como em aplicações de cartão de crédito. Mas as vezes você precisa que seja garantido uma determinada ordem, como em um aplicativo de compras que para finalizar o pedido você precisa antes ter cadastrado o seu endereço. É importante fazer o levantamento dos seus requisitos para poder definir qual tipo de fila é mais adequada pro seu projeto, nesse projeto exemplo irei usar as filas comum.

Padrão Publish/Subscribe

Com esses dois serviços é possível estabelecer o padrão de Publish-Subscribe, essa estrutura permite que se estabeleça uma comunicação totalmente assíncrona entre dois serviços independentes, por possibilitar esse nível de desacoplamento é que essa estrutura é bastante utilizada em arquiteturas de microserviço ou serveless.

De modo geral, existe um serviço B que aguarda o recebimento de notificação que serão enviadas por um serviço A, mas que esses dois serviços não tem nenhuma ligação direta. Em diversas situações o serviço B, por ser completamente independente, não necessariamente precisaria saber de quem é que se origina a notificação que ele recebeu, deveria saber apenas que ele recebeu uma mensagem e deverá seguir o seu processamento a partir dela.

Nesse caso, o serviço A apenas dispararia a mensagem para um tópico que ele tem acesso, esse tópico dispararia essa mensagem para todos os endpoints que estão inscritos nele, que no nosso exemplo seria uma fila SQS, ao receber a mensagem na fila o serviço B iria verificar quais as mensagens foram disponibilizadas na fila a qual tem acesso.

No diagrama de exemplo abaixo mostra como aconteceria esse fluxo, podemos ver inclusive que é possível enviar uma mensagem para inscritos em específicos de um tópico, desse modo é possível com o serviço SNS disparar mensagens tanto para todos os inscritos de um determinado tópico como também é possível filtrar o envio dessa mensagem. Futuramente farei uma implementação do uso desses filtros e compartilharei aqui com vocês, para esse início iremos começar simples com apenas 1 tópico SNS enviando mensagens para 1 fila SQS.

Configuração do Tópico e da Fila

Você pode utilizar o console da AWS tanto para criação quanto para configuração e acesso dos tópicos e das filas, mas a Amazon também disponibiliza SDK's para criação e gerenciamente desses serviços via código. Nesse projeto eu irei utilizar o Boto3, que é um SDK da AWS implementado para linguagem python, tanto para criação e configuração dos tópicos e de filas, como para processamento e envio de mensagens nos código mostrados nos futuros textos dessa série.

Precisamos importar o boto3 e instanciar os clientes

Lembrando que aqui você deverá substituir e utilizar as suas credenciais da AWS.

import boto3

sns_client = boto3.client(
            'sns', 
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name='us-east-1')

sqs_client = boto3.client(
            'sqs', 
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name='us-east-1')

Criando um tópico

Para criar um tópico basta dar-lhe um nome, é possível realizar configurações mais específicas utilizando o boto3, mas por agora manteremos simples.

topic_arn = sns_client.create_topic(
    Name='PurchaseEvents'
)['TopicArn']

Criando uma fila

Criar filas é igualmente simples, porém existem algumas configurações que ao decorrer do projeto veremos que será necessário fazer.

queue = sqs_client.create_queue(
    QueueName='PaymentQueue'
)
queue_url, queue_arn = queue['QueueUrl'], queue['QueueArn']

A primeira dela é criar a permissão de que o tópico que a gente criou logo acima possa enviar mensagens para esta fila, para isso devemos declarar a seguinte política de acesso.

queue_sns_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
            "Effect": "Allow",
            "Principal": "*",
            "Action": "SQS:SendMessage",
            "Resource": queue_arn,
            "Condition": {
                "StringEquals": {
                "aws:SourceArn": topic_arn
                }
            }
            }
        ]
        }

sqs_client.set_queue_attributes(
    QueueUrl= queue_url
    Attributes={
         'Policy' : json.dumps(queue_sns_policy)
    }
)

Inscrevendo a fila no nosso tópico

Agora que temos o nosso tópico e a nossa fila SQS que já está configurada para receber mensagens do tópico, falta apenas inscrever a fila para que quando uma mensagem for enviada para o nosso tópico ele dispare para nossa fila.

payment_subscription = sns_cliente.subscribe(
    TopicArn = topic_arn,
    Protocol = 'sqs',
    Endpoint = queue_arn
)

Código completo

import boto3

# Instanciando o cliente SNS e SQS utilizando as credenciais da AWS
# substitua pelas suas credenciais.
sns_client = boto3.client(
            'sns', 
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name='us-east-1')

sqs_client = boto3.client(
            'sqs', 
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name='us-east-1')

# Criação de um tópico, salvando o Arn do tópico criado para uso
# futuro.
topic_arn = sns_client.create_topic(
    Name='PurchaseEvents'
)

# Criação de uma fila, salvando o Arn da fila criada para uso futuro.
queue = sqs_client.create_queue(
    QueueName='PaymentQueue'
)
queue_url, queue_arn = queue['QueueUrl'], queue['QueueArn']

# Importante lembrar que a fila precisa ter uma política de acesso que
# permita o recebimento das mensagens advindas do seu tópico, por isso 
# iremos atualizar a fila para aceitar essa política.
queue_sns_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
            "Effect": "Allow",
            "Principal": "*",
            "Action": "SQS:SendMessage",
            "Resource": queue_arn,
            "Condition": {
                "StringEquals": {
                "aws:SourceArn": topic_arn
                }
            }
            }
        ]
        }

# Atualizar a política de acesso da fila.
sqs_client.set_queue_attributes(
    QueueUrl= queue_url
    Attributes={
         'Policy' : json.dumps(queue_sns_policy)
    }
)

# Inscrevendo a fila no tópico recém criado.
payment_subscription = sns_cliente.subscribe(
    TopicArn = topic_arn,
    Protocol = 'sqs',
    Endpoint = queue_arn
)

Conclusão e implementações futuras

E assim finalizamos a parte 1 desse projeto para explorar as funcionalidades da AWS SQS e SNS. No próximo texto irei mostrar um pouco da implementação da API em DRF e como realizar a integração com o tópico e fila que a criamos agora.