Introdução Airflow provê diferentes formas de se trabalhar com fluxos automatizados e uma das formas é a possibilidade de acesso a APIs externas utilizando operadores HTTP e extraindo os dados necessários. Mãos na massa Neste tutorial iremos criar uma DAG na qual acessará uma API externa e fará a extração dos dados diretamente para um arquivo local. Se for sua primeira vez utilizando Airflow, recomendo acessar este link para entender mais sobre o Airflow e como montar um ambiente. Criando a DAG Para este tutorial, criaremos uma DAG que será trigada a cada 1 hora (schedule_interval="0 * * * *") e acessará uma API externa extraindo alguns dados diretamente para um arquivo JSON local. Neste cenário usaremos o operador SimpleHttpOperator onde provê uma API com capacidade de executar requisições para APIs externas. Perceba que utilizamos dois operadores dentro de uma mesma DAG. O operador SimpleHttpOperator que provê formas de acesso a APIs externas que através do campo method definimos métodos HTTPs (GET, POST, PUT, DELETE). O campo endpoint permite especificar o endpoint da API, que no caso é products e por fim, o parâmetro http_conn_id, onde é necessário passar o identificador da conexão que será definida a seguir através da UI do Airflow. Conforme imagem abaixo, acesse o menu Admin > Connections Preencha os dados conforme imagem abaixo e salve em seguida. Sobre o operador PythonOperator, estamos utilizando apenas para executar uma função Python chamada _write_response utilizando XComs onde através da task_id da task write_response, é possível recuperar o resultado do response e utilizar em qualquer parte do código. Neste cenário estamos usando o resultado recuperado da API para escrever no arquivo. XCom é um mecanismo de comunicação entre diferente tasks que torna o Airflow bastante flexível. Muitas das vezes as tasks podem ser executadas em diferentes máquinas e com o uso de XComs, possibilita a comunicação e a troca de informações entre Tasks. Por fim, definimos a execução das tasks e suas dependências, veja que usamos o operador >>, que é basicamente definir a ordem da execução entre as tasks. No nosso caso, o acesso a API e extração deve ser executado antes da escrita no arquivo extract_data >> write_response. Após a execução da DAG, é possível acessar o arquivo que foi gerado com o resultado da extração, basta acessar via terminal um dos workers que no caso vamos ter somente um. Execute o seguinte comando abaixo para listar os containers: docker ps Uma listagem similar a esta abaixo será mostrada. Perceba que uma das linhas na coluna NAMES refere-se ao worker, no caso coffee_and_tips_airflow-worker_1. Continuando no terminal, digite o seguinte comando para ter acesso ao diretório do Airflow onde arquivo extract_data.json se encontra. docker exec -it coffee_and_tips_airflow-worker_1 /bin/bash Pronto, agora é só abrir o arquivo e conferir o conteúdo. Conclusão Mais uma vez vimos o poder do Airflow para processos automatizados que requer acessos e integrações de APIs externas de forma fácil e com poucas linhas de código. Neste exemplo exploramos o uso de XComs que visa flexibilizar a troca de mensagens entre tasks que podem ser executadas em diferentes máquinas em um ambiente distribuído. Espero que tenha curtido!

Introdução Java Future é uma das várias formas de se trabalhar com a linguagem de forma assíncrona provendo um contexto multi-Thread em que é possível executar tarefas em paralelo sem gerar um bloqueio no processo. No exemplo abaixo faremos uma simulação de envio de e-mail fictício em que mesmo durante o envio, o processo não será bloqueado, ou seja, não será necessário esperar o termino do envio para que as demais funcionalidades ou mecanismos voltem a operar. Classe EmailService Entendendo a classe EmailService A classe acima representa de forma fictícia o envio de e-mails em massa, a ideia de se usar o loop simulando o envio é justamente para atrasar o processo em si. Por fim, ao final do envio, o método sendEmailBatch(int numberOfEmailsToBeSent) retorna uma String contendo uma mensagem referente ao fim do processo. Classe EmailServiceAsync Entendendo a classe EmailServiceAsync A classe EmailServiceAsync representa o mecanismo assíncrono em si, nela temos o método sendEmailBatchAsync(int numberOfEmailsToBeSent) no qual será o responsável por tornar o processo de envio de e-mail fictício assíncrono. O processo assíncrono é gerenciado pelo uso do ExecutorService no qual facilita o gerenciamento de tarefas de forma assíncrona nas quais são atribuídas a um pool de threads. No caso, a chamada ao método sendEmailBatch(int numberOfEmailsToBeSent) se resume a uma tarefa (task) no qual será atribuída a uma Thread definida em Executors.newFixedThreadPool(1). Por fim, o método retorna uma Future que é literalmente uma promessa de que aquela tarefa em alguma hora será finalizada, representando um processo assíncrono. Classe EmailServiceAsyncRun Entendendo a classe EmailServiceAsyncRun É nesta classe onde iremos testar o processo assíncrono usando Future. Vamos recapitular, na classe EmailService, criamos um método chamado sendEmailBatch(int numberOfEmailsToBeSent) no qual estamos simulando através do for o envio de e-mail fictício e printando uma mensagem de envio que usaremos para testar a concorrência. Na classe EmailServiceAsync, o método sendEmailBatchAsync(int numberOfEmailsToBeSent) cria um ExecutorService que fará o gerenciamento das tasks juntamente com o pool de threads, que neste caso, estamos criando só uma Thread definido em Executors.newFixedThreadPool(1) e retornará uma Future. Agora na classe EmailServiceAsyncRun, é onde de fato testamos o processo, vamos entender por partes: Instanciamos um objeto do tipo EmailServiceAsync Criamos um objeto do tipo Future e atribuímos ao retorno do método emailAsync.sendEmailBatchAsync(500) . A ideia do argumento 500 é apenas para controlar a iteração do For, atrasando o processo para ser finalizado. Até poderíamos usar Thread.sleep() como alternativa e definir um tempo de delay que também funcionaria bem. Perceba que estamos utilizando para controlar o controle de iteração while, o método futureReturn.isDone(), ou seja, este método permite que o processo não seja bloqueado enquanto o fluxo de e-mail seja executado. Neste caso, qualquer processo em que deseja implementar para concorrer enquanto o envio é feito, pode ser criado dentro do while, como por exemplo, um fluxo de atualização de tabelas de clientes ou qualquer outro processo. Na linha 20, através do método futureReturn.get(), estamos imprimindo o resultado do envio dos e-mails. E por fim, finalizamos o executorService e suas tasks através do método executorService.shutdown(). Executando o processo Perceba claramente que existem dois processos distintos sendo executados, o processo de envio de email "Sending email Nº 498.." e o processo de atualização de uma tabela de cliente. Trabalhando com processos blocantes O uso do Future é bastante utilizado para casos de uso onde precisamos bloquear um processo, ou seja, a Thread atual será bloqueada até que o processo sendo executado por Future termine. Para isso, basta invocar diretamente o método futureReturn.get() sem usar qualquer controle de iteração como foi usado no exemplo anterior. Um ponto importante é que este tipo de abordagem pode fazer com que recursos sejam desperdiçados devido ao bloqueio da Thread atual. Conclusão O uso de Future é bem promissor quando precisamos adicionar processos assíncronos em nosso código da maneira mais simples ou até mesmo utilizar para bloqueio de processos. É uma API enxuta com uma certa limitação de recursos mas que funciona bem para alguns cenários. Espero que tenha curtido!

Introdução Airflow tem sido uma das principais ferramentas de orquestração do mercado e muito falada no mundo Modern Data Stack, por ser uma ferramenta capaz de orquestrar workloads de dados através de ETLs ou ELTs. Mas na verdade, o Airflow não se resume somente a isso, ele pode ser aplicado em diversos casos de usos do dia a dia de um Engenheiro de Dados ou Software. Neste Tutorial sobre Airflow para iniciantes, iremos apresentar o Airflow da maneira mais simples, sem a necessidade de saber ou criar ETLs. Mas o que é o Airflow de fato? O Apache Airflow é uma plataforma de orquestração de fluxo de trabalho amplamente utilizada para agendar, monitorar e gerenciar pipelines de dados. Ele possui vários componentes que trabalham juntos para fornecer suas funcionalidades. Componentes do Airflow DAG O DAG (Directed Acyclic Graph) é o principal componente e a representação de fluxo de trabalho no Airflow. É composto por tarefas (tasks) e dependências entre elas. As tarefas são definidas como operadores (operators), como PythonOperator, BashOperator, SQLOperator e entre outros. O DAG define a ordem de execução das tarefas e as relações de dependências. Webserver O componente do Webserver fornece uma interface da web para interagir com o Airflow. Ele permite que você visualize, gerencie e monitore seus fluxos de trabalho, tarefas, DAGs e logs. O Webserver também permite a autenticação de usuários e o controle de acesso com base em funções. Scheduler O Scheduler é responsável por agendar a execução das tarefas de acordo com a definição do DAG. Ele verifica periodicamente se há tarefas pendentes para execução e atribui recursos disponíveis para a execução das tarefas no momento apropriado. O Scheduler também lida com a recuperação de falhas e o agendamento de retries de tarefas. Executor O Executor é responsável por executar as tarefas definidas nos DAGs. Existem diferentes tipos de executores disponíveis no Airflow, como LocalExecutor, CeleryExecutor, KubernetesExecutor e etc. Cada executor tem suas próprias configurações e comportamentos de execução. Metadatabase O Metadatabase é um banco de dados onde o Airflow armazena metadados sobre as tarefas, DAGs, execuções, agendamentos, entre outros. Ele é usado para rastrear o status das tarefas, registrar o histórico de execução e fornecer informações para o monitoramento e a visualização do fluxo de trabalho. É possível utilizar diversos outros bancos de dados para registrar o histórico como MySQL, Postgres e dentre outros. Workers Os Workers são os nós de execução em um ambiente distribuído. Eles recebem tarefas atribuídas pelo Scheduler e as executam. Os Workers podem ser dimensionados horizontalmente para lidar com pipelines de dados maiores ou para distribuir a carga de trabalho para vários recursos. Plugins Os Plugins são extensões do Airflow que permitem adicionar novos recursos e funcionalidades ao sistema. Eles podem incluir novos operadores, hooks, sensores, conexões com sistemas externos e entre outros. Os Plugins fornecem uma maneira de personalizar e estender as capacidades do Airflow para atender às necessidades específicas de um fluxo de trabalho. Operadores Operadores são basicamente a composição de um DAG. Entenda um operador como um bloco de código com responsabilidade própria. Pelo Airflow ser um orquestrador e executar um fluxo de trabalho, podemos ter diferentes tarefas a serem executadas como por exemplo, acessar uma API, enviar um e-mail, acessar uma tabela em um banco de dados e efetuar uma operação, executar um código Python ou até um comando Bash. Para cada uma das tarefas acima, devemos usar um operador. A seguir, iremos abordar alguns dos principais operadores: BashOperator O BashOperator permite executar comandos Bash ou scripts diretamente no sistema operacional onde o Airflow está sendo executado. É útil para tarefas que envolvem a execução de scripts de shell, utilitários ou qualquer ação que possa ser realizada no terminal. Resumindo, quando precisamos abrir o terminal do nosso sistema e executar algum comando para manipular arquivos ou algo relacionado ao próprio sistema, porém dentro de um DAG, é este o operador a ser usado. PythonOperator O PythonOperator permite que você execute funções Python como tarefas no Airflow. Você pode escrever suas próprias funções Python personalizadas e usar o PythonOperator para chamar essas funções como parte do fluxo de trabalho. DummyOperator O DummyOperator é uma tarefa "falsa" que não realiza nenhuma ação. É útil para criar dependências e fluxos de trabalho complexos sem a necessidade de executar nenhuma ação real. Sensor Os Sensores são usados para aguardar a ocorrência de algum evento externo antes de continuar o fluxo de trabalho, pode funcionar como um listener (ouvinte). Por exemplo, o HttpSensor que é um tipo de Sensor, pode validar se uma API externa esteja ativa, caso esteja, o fluxo continua sendo executado. Não necessariamente é um operador HTTP que deveria retornar algo, mas somente um tipo de ouvinte. HttpOperator Diferente de um Sensor, o HttpOperator é usado para realizar solicitações HTTP, como GET, POST, PUT, DELETE e etc. Neste caso, permite interagir de forma mais completa com APIs internas ou externas. SqlOperator SqlOperator é o operador responsável por executar operações de DML e DDL em um banco de dados, ou seja, desde manipulações de dados como SELECTS, INSERTS, UPDATES até criação de tabelas, alteração e etc. Executores Os executores são responsáveis por executar as tarefas definidas em um fluxo de trabalho (DAG). Eles gerenciam a alocação e a execução das tarefas em tempo de execução, garantindo que cada tarefa seja executada de forma eficiente e confiável. O Airflow oferece diferentes tipos de executores, cada um com características e funcionalidades distintas, permitindo que você escolha o mais adequado para suas necessidades específicas. A seguir, abordaremos alguns dos principais executores: LocalExecutor O LocalExecutor é o executor padrão no Apache Airflow. Ele é projetado para ser usado em ambientes de desenvolvimento e teste onde a escalabilidade não é uma preocupação. O LocalExecutor executa as tarefas em threads separados no mesmo processo do Airflow. Essa abordagem é simples e eficiente para pipelines menores ou para execuções em um único nó. CeleryExecutor Se você precisa de um executor para ambientes distribuídos e com alta escala, o CeleryExecutor é uma excelente opção. Ele utiliza o Celery, uma biblioteca de tarefas em fila, para distribuir as tarefas em nós de execução separados. Essa abordagem torna o Airflow adequado para executar pipelines em clusters de servidores, permitindo dimensionar horizontalmente conforme a demanda. KubernetesExecutor Para ambientes que utilizam Kubernetes como plataforma de orquestração de containers, o KubernetesExecutor é uma escolha natural. Ele aproveita a capacidade de orquestração do Kubernetes para executar tarefas em pods separados, o que pode resultar em melhor isolamento de recursos e mais facilidade na execução de tarefas em containers. DaskExecutor Se o seu fluxo de trabalho requer processamento paralelo e distribuído, o DaskExecutor pode ser a opção certa. Ele utiliza a biblioteca Dask para realizar a computação paralela em um cluster de recursos. Essa abordagem é ideal para tarefas que podem ser divididas em sub-tasks independentes, permitindo maior aproveitamento dos recursos disponíveis. Linguagem de programação Airflow suporta unicamente como linguem de programação, o Python. Para ser sincero, não é um limitador para quem não conhece bem a linguagem. Na prática, o processo de criar DAGs é padrão, o que pode mudar de acordo com as suas necessidades, será lidar com diferentes tipos de operadores, podendo ou não usar Python. Mãos na massa Montando o ambiente Para este tutorial iremos usar o Docker que nos ajudará a provisionar o nosso ambiente sem a necessidade de instalar o Airflow. Caso não tenha o Docker instalado, recomendo seguir as recomendações deste link e após instalado, volte para seguirmos o tutorial. Baixando o projeto Para facilitar, clone o projeto do seguinte repositório e siga os passos para fazer o deploy do Airflow. Passos para o deploy Com o docker instalado e após ter baixado o projeto conforme o item anterior, acesse o diretório onde se encontra o projeto e abra o terminal, execute o seguinte comando docker: docker-compose up O comando acima irá iniciar os containers do docker onde estão presentes os serviços do próprio Airflow, postgres e entre outras que iremos utilizar. Se tiver curiosidade de como estes serviços estão mapeados, abra o arquivo docker-compose.yaml do projeto e lá você terá mais detalhes. Enfim, após a execução do comando acima e os containers já iniciados, acesse via browser o seguinte endereço http://localhost:8080/ Uma tela como abaixo será aberta, basta digitar airflow para o usuário e senha e acesse a UI do Airflow. Criando a DAG Criando um simples Hello World Para este exemplo, iremos criar uma simples DAG onde irá printar o clássico "Hello World". No projeto que você baixou, acesse a pasta /dags e crie o seguinte arquivo python chamado hello_world.py . O Código acima é um simples exemplo de uma DAG escrita em Python. Percebe-se que começamos importando algumas funções, incluindo a própria DAG, função relacionada a data e o operador Python. Em seguida, criamos uma função Python que irá printar no console "Hello World" chamada print_hello. Esta função será chamada pela DAG mais a frente. A declaração de uma DAG inicia-se com with DAG(..) passando alguns argumentos como: dag_id: Identificador da DAG no contexto do Airflow start_date: A data definida é apenas um ponto de referência e não necessariamente a data do início de execução e nem da criação da DAG. Normalmente as execuções são feitas a uma data posterior a definida neste parâmetro, e tem uma importância quando precisamos calcular execuções entre o início e o que foi definido no parâmetro schedule_interval. schedule_interval: Neste parâmetro definimos a periodicidade em que a DAG será executada. É possível definir diferentes formas de execuções através de expressões CRON ou através de Strings já definidas como @daily, @hourly, @once, @weekly e etc. No caso do exemplo, o fluxo será executado apenas uma vez. catchup: Este parâmetro controla execuções retroativas, ou seja, caso esteja definido como True, o Airflow irá executar o período retroativo a partir da data definida no start_date até a data atual. No exemplo anterior definimos como False por não ter a necessidade da execução retroativa. Após o preenchimento dos argumentos, criamos a task hello_task dentro da própria DAG através do operador PythonOperator no qual provê formas de executar funções python dentro de uma DAG. Perceba que declaramos um identificador através do task_id e no argumento python_callable no qual é nativa do operador PythonOperator, passamos a função python print_hello criada anteriormente. Por último, invoque a task hello_task. Dessa forma, a DAG entenderá que esta será a task a ser executada. Caso você já tenha feito o deploy, a DAG aparecerá no Airflow em pouco tempo para ser executada conforme na imagem abaixo: Após a DAG criada, ative-a e a execute clicando em Trigger DAG conforme a imagem acima. Clique na task hello_operator (centro) e em seguida uma janela será aberta conforme imagem abaixo: Clique no botão Log para ver mais detalhes da execução: Perceba como é simples a criação de uma DAG, basta pensar nas diferentes possibilidades e cenários de aplicabilidade. A seguir, faremos mais exemplos um pouco mais complexos explorando várias outros cenários. Conclusão Com base no simples exemplo mostrado, Airflow apresentou uma abordagem flexível e simples para o controle de fluxos automatizados, desde a criação de DAGs até a navegação do seu componente web. Como citei no início, a sua utilização não se limita somente na orquestração de ETLs como é feito no mercado em geral, mas também na possibilidade do seu uso em tarefas que requer qualquer necessidade em controlar fluxos que possuem dependências entre seus componentes dentro de um contexto escalável ou não. Repositório GitHub Espero que tenha curtido!

Introdução A Java Reflection é uma poderosa API que permite a um programa Java examinar e manipular informações sobre suas próprias classes em tempo de execução. Com a Reflection, é possível obter informações sobre campos, métodos e construtores de uma classe, além de acessar e modificar esses elementos mesmo que eles sejam privados. Neste post vamos escrever alguns códigos Java explorando algumas das facilidades de se usar Reflection e quando aplicar em seus projetos. Classe Bank Criaremos uma simples classe chamada Bank, onde iremos criar alguns campos, métodos e construtores para que sejam explorados pelo uso do Reflection. Acessando os campos da classe Bank Com a classe Bank criada, vamos explorar via Reflection a listagem de todos campos da classe através do método getDeclaredFields da classe Class. Perceba que através do método estático Class.forName, passamos como parâmetro uma string com o nome da classe em que desejamos explorar via Reflection. Saída Field name: code Field type: class java.lang.Integer ************ Field name: nameOfBank Field type: class java.lang.String ************ Field name: amountOfDepositedMoney Field type: class java.lang.Double ************ Field name: totalOfCustomers Field type: class java.lang.Integer ************ Acessando os métodos da classe Bank Através do método getDeclaredMethods, podemos recuperar todos os métodos da classe Bank. Saída Method name: doDeposit Method type: class java.lang.String ************ Method name: doWithDraw Method type: class java.lang.String ************ Method name: getReceipt Method type: class java.lang.String ************ Criando objetos Com o uso do Reflection para criar objetos, é necessário cria-los através de um construtor. Neste caso, devemos primeiro invocar um construtor para que seja criado o objeto. O detalhe é que para recuperar este construtor, devemos nos atentar aos tipos dos parâmetros que compõem o construtor e a ordem que eles são declarados. Isso torna flexível recuperar diferentes construtores com quantidade de parâmetros e tipos diferentes. Perceba abaixo que foi necessário criar um array do tipo Class atribuindo diferentes tipos de acordo com a composição do construtor que iremos usar para criar o nosso objeto. Neste cenário, será necessário invocar o método class.getConstructor(argType) passando o array criado anteriormente como argumento. Dessa forma, teremos um objeto construtor que será usado na criação do nosso objeto. Por fim, criamos um novo array do tipo Object atribuindo os valores que irão compor o nosso objeto obedecendo a ordem definida no construtor e em seguida, basta invocar o método constructor.newInstance(argumentsValue) passando o array como parâmetro retornando o objeto em que desejamos criar. Saída Bank{code=1, nameOfBank='Bank of America', amountOfDepositedMoney=1.5, totalOfCustomers=2500} Invocando métodos Para invocar um método através de Reflection é bem simples como é mostrado no código abaixo. Perceba que é necessário passar como parâmetro no método cls.getMethod("doDeposit", argumentsType) o nome explícito do método, no caso o "doDeposit" e no segundo parâmetro, um array representando o tipo do dado usado no parâmetro do método doDeposit(double amount), no caso um parâmetro do tipo double. Por fim, invocar o método method.invoke passando como primeiro parâmetro o objeto referenciando a classe, no caso um objeto do tipo Bank. E como segundo parâmetro, o próprio valor do parâmetro que será executado no método. Saída 145.85 of money has been deposited Conclusão O uso de Reflection é uma boa estratégia quando necessita-se de flexibilidade em explorar diferentes classes e seus métodos sem a necessidade de instanciar objetos. Normalmente usa-se Reflection em componentes específicos de uma arquitetura mas não impede de ser usado em cenários variados. Pelo exemplos mostrados acima, percebe-se cenários infinitos de sua aplicação e as vantagens do seu uso. Espero que tenha curtido!

Introdução Liberado no Java 14 como preview, mais especificamente na JEP 395, Record Class é uma alternativa para se trabalhar com Classes em Java. Record Class foi uma sacada bem interessante desenhado para eliminar a verbosidade de quando se precisa criar uma classe e seus componentes, como por exemplo: Construtores canônicos Métodos de acesso público Implementar os métodos equals e hashCode Implementar o método toString Usando Record Classes não é mais necessário declarar os itens acima, ajudando o desenvolvedor a ser mais focado em outras frentes. Vamos entender melhor na prática. Vamos criar uma classe Java chamada User e adicionar alguns campos e métodos. Perceba que para uma simples classe com 4 campos, criamos um construtor, métodos de acesso público, implementamos os métodos equals e hashCode e por fim, o método toString. Funciona bem, porém poderíamos evitar a complexidade e criar um código menos verboso. Nesse caso, podemos utilizar Record Classes no lugar da classe User acima. User Record Class É notável a diferença entre Record e uma Classe Java tradicional. Perceba que não foi necessário declarar os campos, criar os métodos de acessos ou implementar nenhum outro método. Em uma Record Class quando criada, implicitamente os métodos de acesso público são criados, as implementações dos métodos equals, hashCode e toString também são criados de forma automática sendo não necessário implementa-los de forma explícita. E por fim, os campos de referências ou componentes, são criados como private final com o mesmos nomes. Saída Desvantagens Record Class se comporta como uma classe Java comum, mas a diferença é que não se pode trabalhar com herança, ou seja, não se pode estender (extends) outra classe, somente implementar uma ou mais interfaces. Outro ponto é que não é possível criar variáveis de instâncias não estáticas (static); Conclusão final Record Classes é uma ótima abordagem para quem procura um código menos verboso ou que precisa de agilidade em implementar modelos. Apesar da limitação de não poder estender outras Record Classes, é um limitador que não prejudica o seu uso em geral. Espero que tenha curtido!

O que é o Change Data Feed? Change Data Feed é uma feature do Delta Lake a partir da versão 2.0.0 que permite rastrear a níveis de linhas em tabelas Delta, mudanças como operações de DML (Merge, Delete ou Update), versões do dados e o timestamp de quando aconteceu a mudança. O processo mapeia operações de Merge, Delete e Update mantendo o histórico de alterações a nível de linha, ou seja, cada evento efetuado em um registro, o Delta através do Change Data Feed consegue registrar como uma espécie de auditoria. É claro que é possível utilizar para diferentes casos de uso, é extenso as possibilidades. Como funciona na prática Iremos simular uma pipe aplicando Change Data Feed para tabelas Delta, pois não é um recurso default. Após a criação da tabela Delta, iremos executar algumas operações visando explorar mais sobre o poder do Change Data Feed. Iremos trabalhar com seguinte Dataset: Criando a Sessão Spark e configurando alguns parâmetros do Delta A partir de agora, criaremos o código em partes para um fácil entendimento. O código abaixo estamos criando o método responsável por manter a sessão do Spark e configurando alguns parâmetros para o funcionamento do Delta. Carregando o Dataset Vamos carregar o Dataset e criar uma view temporário para ser usada na nossa pipeline mais adiante. Criando a tabela Delta Agora faremos a criação da tabela Delta já configurando Change Data Feed nas propriedades da tabela e toda a parte de metadata será baseada no Dataset anteriormente apresentado. Perceba que estamos usando na propriedade o seguinte parâmetro delta.enableChangeDataFeed = true para a ativação do Change Data Feed. Executando Merge dos dados Agora executaremos uma simples operação Merge para que o Change Data Feed possa registrar como mudança na nossa tabela. Veja que o Merge utiliza na nossa view global_temp.raw_product anteriormente criada para o upsert dos dados. Auditando a tabela Agora que o Merge foi executado, vamos executar uma leitura na nossa tabela para entender o que aconteceu e como o Change Data Feed atuou. Perceba que estamos passando os seguinte parâmetros: 1. readChangeFeed em que é necessário para o uso do Change Data Feed. 2. startingVersion é o parâmetro responsável por restringir a partir de qual versão inicial queremos que seja apresentada. Resultado após execução: Veja que além da colunas definidas na criação da tabela, temos 3 novas colunas gerenciadas pelo Change Data Feed. 1. _change_type: Coluna contendo valores de acordo com cada operação efetuada como insert, update_preimage , update_postimage, delete 2. _commit_version: Versão da mudança 3. _commit_timestamp: Timestamp representando a data da mudança No resultado acima, o resultado do upsert foi um simples insert, por não conter todas as condições possíveis de um update. Deletando um registro Nessa etapa faremos um simples delete em um registro da tabela, apenas para validarmos como o Change Data Feed irá se comportar. Auditando a tabela (novamente) Perceba abaixo que após a deleção do registro 6, agora temos um novo registro criado como delete na tabela e seu incremento de versão para 2. Outro ponto é que o registro original foi mantido, porém com a versão antiga. Atualizando um registro Agora como último teste, iremos atualizar um registro para entender novamente o comportamento do Change Data Feed. Auditando a tabela (última vez) Agora como último teste, executamos um simples update em um registro para entender como será o comportamento. Perceba que 2 novos valores foram adicionados/atualizados na coluna _change_type. O valor update_postimage é o valor após o update executado e dessa vez, para o registro antigo, foi mantido a mesma versão do novo na coluna _commit_version, pois este mesmo registro foi atualizado de acordo com coluna _change_type para update_preimage, ou seja, valor antes da mudança. Conclusão O Change Data Feed é um ótimo recurso para entender o comportamento da sua pipeline de dados e também uma forma de auditar registros visando um melhor entendimento das operações ali executadas. Segundo o próprio time do Delta, é um recurso que mantido não gera nenhum overhead significativo, ou seja, talvez seja uma feature que pode ser adotada de forma integral em sua estratégia de dados pois possui vários benefícios como foi mostrado neste post. Repositório GitHub Espero que tenha curtido!

Introdução A ordenação de dados é uma tarefa comum em muitos projetos de desenvolvimento de software. Quando trabalhamos com coleções de objetos em Java, uma abordagem poderosa e flexível para realizar a ordenação é o uso da interface Comparator.comparing em conjunto com Streams. Neste post, vamos mostrar que usando Comparator.comparing para ordenar Java Stream pode tornar a ordenação elegante e eficiente. O que é a interface Comparator.comparing? A interface Comparator.comparing é uma funcionalidade introduzida no Java 8 como parte do pacote java.util.Comparator. Ela fornece um método estático chamado comparing, que permite especificar uma função de chave (chave de ordenação) para comparar objetos. Essa função é usada para extrair um valor de um objeto e compará-lo com base nesse valor durante a ordenação. Flexibilidade na ordenação com Comparator.comparing Uma das principais vantagens da interface Comparator.comparing é sua flexibilidade. Com ela, podemos realizar ordenações em diferentes campos de um objeto, permitindo a criação de lógicas de ordenação complexas de forma simples e concisa. Perceba no código abaixo que simplemente no método sorted(), passamos como argumento a interface Comparator.comparing que por sua vez, passou o campo city como argumento utilizando method reference (People::getCity) executando a ordenação por este campo. Saída da execução Monica John Mary Anthony Seth Ordenação de múltiplos critérios Muitas vezes, é necessário realizar ordenações com base em múltiplos critérios. Isso é facilmente alcançado com a interface Comparator.comparing. Basta encadear vários métodos comparing, cada um especificando um critério diferente. O Java se encarregará de realizar a ordenação de acordo com a sequência especificada. Por exemplo, podemos ordenar a mesma lista por city e, em seguida, por name: Comparator.comparing(People::getCity).thenComparing(People::getName). Ordenação ascendente e descendente Outra vantagem importante da interface Comparator.comparing é a capacidade de realizar ordenações tanto em ordem ascendente quanto descendente. Para isso, basta encadear o método reversed() como no código abaixo: Saída da execução Seth Mary John Anthony Monica Eficiência e simplicidade Ao usar a interface Comparator.comparing em conjunto com Streams, a ordenação se torna mais eficiente e elegante. A combinação desses recursos permite escrever código limpo, de fácil leitura e manutenção. Além disso, o Java otimiza internamente a ordenação usando algoritmos eficientes, resultando em um desempenho satisfatório mesmo para grandes conjuntos de dados. Conclusão final A interface Comparator.comparing é uma ferramenta poderosa para realizar a ordenação de Streams em Java. Sua flexibilidade, capacidade de ordenação ascendente e descendente, suporte para múltiplos critérios e eficiência na execução tornam-na uma opção valiosa para qualquer desenvolvedor Java. Ao aproveitar essa interface, podemos obter um código mais conciso, menos verboso e eficiente, facilitando a manipulação de objetos em uma Stream. Espero que tenha curtido!

Neste post iremos aprender a como criar um app serverless usando AWS SAM. AWS SAM (Serverless Application Model) é uma extensão do AWS CloudFormation, especificamente projetado para desenvolvimento e implantação de aplicativos sem servidor, o famoso serverless como AWS Lambda, API Gateway, DynamoDB, entre outros recursos da AWS. Nível de abstração AWS SAM é uma ferramenta de nível de aplicativo, focada principalmente na criação e implantação de aplicativos serverless na AWS. Ele fornece abstrações de nível mais alto para facilitar o desenvolvimento e a implantação de aplicativos serverless, com foco nos serviços da AWS necessários para suportar esse tipo de arquitetura, ou seja, todo o foco é AWS e não outra nuvem. AWS SAM possui toda uma forma de gerar o código do projeto localmente e possibilita gerar testes, Build e Deploy através de SAM CLI. Como instalar AWS SAM Acesse este link e siga os passos de acordo com cada sistema operacional. Como criar um projeto serverless Após instalado, através de um terminal, gere seu projeto localmente gerando os arquivo necessários para então, efetuar o deploy da aplicação. Primeiramente, acesse a pasta em que deseja gerar seu recurso serverless e em seguida abra o terminal. Digite no terminal o seguinte comando para iniciar o SAM: sam init Após digitar, um prompt com algumas opções aparecerá para que você preencha as informações do seu projeto. Acima temos 2 opções para gerar o nosso template inicial, vamos digitar 1 para gerarmos a opção 1 - AWS Quick Start Templates. Após digitarmos, uma nova lista será mostrada com algumas opções de templates. Perceba que cada opção se resume a um recurso como, Lambda, tabela Dynamo e até um simples API utilizando API Gateway. Para este cenário, vamos criar uma tabela Dynamo, neste caso, digite a opção 13 e aperte o enter. Após digitar, algumas questões serão feitas, apenas digite y para prosseguir até que uma nova tela sobre as informações do projeto seja ofertada como abaixo. Digite o nome do projeto em que deseja e aperte o enter. No nosso caso digitei o seguinte nome para o projeto dynamo-table-using-aws-sam como na imagem abaixo. Após digitar o nome do projeto, o template e arquivos contendo o código base estará disponível e pronto para o deploy. Acesse a pasta e veja que um arquivo chamado template.yaml foi criado contendo informações dos recursos que serão criados. É bem similar a um template do CloudFormation, porém mais resumido. Abra o arquivo e perceba que vários recursos auxiliares foram mapeados no template, como o próprio Dynamo, um Lambda e uma API Gateway. Foram criados também, alguns códigos bases relacionados ao Lambda e alguns testes unitários que possibilitam invocações locais. Como fazer o deploy Agora que o nosso template e o código base foi gerado, é hora de criarmos a tabela Dynamo na AWS, basta seguir os próximos passos. Acesse novamente o terminal e digite o seguinte comando: sam deploy --guided Após executar este comando, as seguintes opções serão mostradas no prompt do terminal para preenchimento: Para o campo Stack Name, digite um valor que será o identificador daquela stack no qual será usado pelo CloudFormation para criar os recursos necessários. Na dúvida, siga o que o foi digitado conforme imagem acima, no caso dynamo-stack. Após preencher todos os campos, um resumo do que será criado será apresentado conforme imagem abaixo: Por fim, mais uma última pergunta será feita sobre o desejo de efetivar o deploy, basta digitar y para confirmar. Após confirmar a operação, o progresso da criação dos recursos será apresentado no terminal até a finalização do processo. Deploy finalizado, perceba novamente os recursos que foram criados. Agora é só acessar o console da AWS e conferir a tabela criada no Dynamo. Deletando os recursos Caso seja necessário, você pode deletar os recursos via SAM CLI, basta executar o comando abaixo: sam delete dynamo-stack O argumento dynamo-stack refere-se ao identificador que digitamos anteriormente no campo Stack Name, lembra? Utilize o mesmo para deletar toda a stack criada. Após digitar o comando acima, é só confirmar os próximos passos. É bem simples a forma de como criar um recurso serverless com AWS SAM, existem algumas vantagens e desvantagens e isso tudo vai depender da sua estratégia. Espero que tenha curtido!

O que são as classes de armazenamento do Amazon S3? O Amazon S3 (Simple Storage Service) provê uma forma estratégica em organizar os objetos em diferentes camadas, onde cada camada possui suas particularidades que iremos detalhar mais a frente. As classes de armazenamento tem como características oferecer diferentes níveis de durabilidade, disponibilidade, desempenho e custos. Para isso, deve entender bem qual estratégia utilizar para manter os objetos visando o melhor custo benefício. A seguir, iremos detalhar cada classes, descrevendo suas vantagens de desvantagens. S3 Standard A classe de armazenamento S3 Standard é a opção padrão e mais utilizada do Amazon S3. Ela é projetada para oferecer alta durabilidade, disponibilidade e desempenho para objetos frequentemente acessados. Vantagens S3 Standard é a classe mais comum utilizada no armazenamento e acesso a objetos com maior frequência, pois é a camada que oferece baixa latência e isso permite ser usado por diferentes casos de usos onde o acesso dinâmico aos objetos é imprescindível. Outra vantagem é a durabilidade de 99.999999999%, ou seja, as chances dos objetos serem corrompidos ou até perdidos é muito baixa. Quanto a disponibilidade, esta classe fornece um SLA de 99,99%, significa que os objetos possuem uma alta disponibilidade para o acesso. Desvantagens S3 Standard possui algumas desvantagens comparado as outas classes. Uma delas é o alto custo de armazenamento para objetos pouco acessados. Por isso é importante definir políticas de ciclo de vida para lidar com objetos pouco acessados. No caso, existe a classe S3 Standard-Infrequent Access, que seria a mais apropriada para este contexto. Falaremos em breve sobre esta classe. Outra desvantagem está relacionada ao acesso a objetos recém-criados. Mesmo que esta classe possui como umas das principais características a baixa latência. Objetos recém-criados podem não estar disponíveis de imediato em todas as regiões, podendo levar um tempo para a disponibilidade dos objetos para algumas regiões, gerando uma alta latência S3 Intelligent-Tiering A classe de armazenamento S3 Intelligent-Tiering fornece um mecanismo em que possibilita mover objetos com base no padrão de uso de forma automática para camadas mais adequadas, visando baixo custos de armazenamento. Vantagens O próprio conceito já diz tudo sobre uma das vantagens de se usar S3 Intelligent-Tiering. Esta classe é capaz de gerenciar objetos com base no padrão de uso, ou seja, para aqueles objetos que são poucos acessados, a própria classe move para classes mais adequada visando menor custos de armazenamento. S3 Intelligent-Tiering monitora e move objetos de forma automática para camadas mais adequadas de acordo com o padrão de uso, geralmente esta integração funciona para 3 tipos de camadas. Uma camada otimizada para objetos de acessos frequentes, uma camada otimizada para objetos pouco acessados, que segundo a AWS gera uma economia de até 40%. E uma última camada direcionada a objetos que são raramente acessados, gerando uma economia de armazenamento em torno de 68%. Outro ponto de vantagem é que não existe cobrança por acesso aos dados utilizando S3-Intelligent-Tiering. Somente cobranças pelo armazenamento e transferência. Desvantagens Possível aumento na latência para objetos acessados pela primeira vez. O motivo é de que ao mover os objetos para camadas mais adequadas, existe a possibilidade do aumento de latência para estes objetos que são pouco acessados. S3 Standard-Infrequent Access (S3 Standard-IA) Classe adequada para armazenar objetos com acessos menos frequentes mas que precisam estar disponíveis para acessos rápidos, mantendo uma baixa latência. É uma típica classe para armazenar dados a longo prazo. Vantagens O custo de armazenamento é mais baixo se comparado a classe S3 Standard, mantendo as mesmas características de durabilidade. Sobre a disponibilidade dos dados, possui as mesmas características que a classe S3 Intelligent-Tiering, de 99.9% de SLA. E também, permite acesso rápido aos dados oferecendo uma alta taxa de vazão (throughput). A taxa mínima de armazenamento é cobrada mensalmente, diferente de classes como S3-Standard e S3-Intelligent Tiering. Desvantagens Não que necessariamente seja uma desvantagem, mas se comparado a outras classes pode não fazer tanto sentido, no caso, seria o custo de acesso aos dados. O acesso é cobrado por gigabyte acessado, ou seja, dependendo da frequência de acesso e volume acessado, seria melhor manter os dados em uma camada como S3 Standard. Tudo vai depender da sua estratégia. S3 One Zone-Infrequent Access (S3 One Zone-IA) Classe de armazenamento ideal para objetos que são acessados com pouca frequência e que estarão disponíveis em apenas uma zona (Availability Zone). A própria AWS sugere esta classe para cópias secundárias de backup de dados. Vantagens O custo é mais baixo comparada as demais classes de armazenamento, pois os dados serão armazenados em apenas uma zona, tornando o custo operacional baixo. Desvantagens Diferente das demais classes de armazenamento, onde o armazenamento dos objetos estão disponíveis em no mínimo 3 zonas de disponibilidade (AZ). A S3 One Zone-Infrequent Access disponibiliza os dados em somente 1 zona, fazendo com que não exista nenhum tipo de redundância, ou seja, existe uma possível chance na perda dos dados, caso haja alguma falha naquela determinada zona. S3 Glacier Instant Retrieval S3 Glacier Instant Retrieval faz parte da família Glacier, que tem como caraterística o baixo custo de armazenamento para objetos raramente acessados. É uma classe de armazenamento ideal para arquivamento de dados que necessita de acesso imediato. Vantagens O custo de armazenamento é baixo. Possui a mesma disponibilidade, comparado as classes S3 Intelligent-Tiering e S3 Standard-IA. Fornece redundância, ou seja, os dados são replicados para no mínimo 3 zonas de disponibilidades (AZ). Desvantagens Embora ofereça recuperação imediata dos dados mantendo a mesma vazão (throughput) do que classes como S3 Standard e S3 Standard-IA, o custo se torna alto quando é necessário a recuperação destes dados com uma grande frequência em um curto período. S3 Glacier Flexible Retrieval S3 Glacier Flexible Retrieval é a antiga classe de armazenamento chamada somente de S3 Glacier, possui uma característica em armazenar objetos com longa vida de duração, como qualquer outra classe da família Glacier. Esta classe é ideal para objetos que são acessado de 1 a 2 vezes ao ano e que requer recuperação de forma assíncrona, ou seja, sem acesso imediato. Vantagens Esta classe é ideal para manter objetos que não requer recuperação imediata, tornando uma vantagem quanto a custos. No caso, dados como backup, nos quais a recuperação é bem rara, esta classe não oferece custo de recuperação pela ideia da frequência de acessar estes dados ser bem perto de zero. Desvantagens Tempo de recuperação pode ser lento para alguns cenários. Como característica da própria classe, S3 Glacier Flexible Retrieval pode não atender quando seja necessário o acesso imediato aos dados. S3 Glacier Deep Archive Classe de armazenamento com o menor custo entre as classes da família Glacier. Ideal para armazenar dados que podem ser acessados de 1 a 2 vezes ao ano. A AWS sugere o uso desta classe para cenários onde temos que manter dados entre 8 a 10 anos visando atender regulamentos relacionados a compliance ou qualquer outro tipo de regra relacionada a retenção dos dados para longos períodos. Vantagens O menor custo entre as classes do mesmo segmento e com disponibilidade de 99.99%. Classe disponível em no mínimo 3 zonas de disponibilidade (AZ) e ideal para dados que requer longos períodos de retenção. Desvantagens Tempo de recuperação longo, ou seja, caso precise de uma recuperação rápida, este SLA não atenderá. Por ter uma característica em que o dado deve ser pouco acessado, o custo de recuperação pode ser mais alto, dependendo da frequência de acessos. Espero que tenha curtido!

Em alguns momentos precisamos criar um novo projeto por várias razões: estudo, trabalho ou testes. Existem uma série de ferramentas que nos ajudam nesse processo, nesse tutorial vou mostrar como podemos criar um novo projeto Spring Boot diretamente do Intellij IDEA. Para este tutorial estou utilizando a última versão do Intellij IDEA (2023.1). Criando o Projeto: A primeira etapa é criar o projeto, para isso podemos ir no seguinte caminho dentro da IDE: File > New > Project Logo após, você deve selecionar a opção Spring Initializr e preencher as informações do projeto: Nesta janela você pode preencher: Name: nome do seu projeto Location: local onde o projeto será salvo Language: linguagem de programação do projeto Type: tipo de gerenciador de pacotes que será utilizado Gradle ou Maven Group: nome da base de pacotes do projeto Artifact: nome do artefato Package name: nome da base de pacotes que será organizado o projeto Jdk: caminho para a jdk a ser utilizada no projeto Packaging: tipo de pacote que será gerado do projeto ao buildar, geralmente se utilizar o Jar Após preencher tudo cliente em NEXT. Na próxima tela você poderá selecionar quais frameworks serão utilizados, como por exemplo Spring Web: Selecione os frameworks que você deseja para seu projeto, em seguida clique em CREATE. Após isso seu projeto estará pronto. Pode começar a codar ! Links: Intellij IDEA - https://www.jetbrains.com/idea/

Quando estamos desenvolvendo microserviços, um dos itens que buscamos é ter serviços alto contidos. Dessa forma um dos pontos que costumam ficar de fora em vários casos é o banco de dados. Um framework que ajuda a versionar o banco dados através de scripts é o Flyway. O Flyway nos ajuda a tratar todas as mudanças que serão realizadas no banco de dados através de um projeto spring-boot. Ele faz uso de scripts SQL e alguns metadados para gerenciar o estado e as mudanças do banco de dados. A vantagem de utilizar tal framework é que qualquer um que esteja desenvolvendo no projeto vai ter o mesmo estado do banco de dados, geralmente uma cópia do banco de dados de desenvolvimento ou produção. Nesse artigo vou mostrar como configurar o Flyway em um projeto Spring Boot. Criando o projeto Para criar o projeto vamos utilizar o site oficial do Spring. Primeiramente acesse o site:: https://start.spring.io/ Quando a página estiver carregada, configure as informações conforme a imagem abaixo: Quando estiver finalizado basta clicar em GENERATE para baixar o projeto configurado. Assim que realizar o download, você pode importar para sua IDE preferida. Eu vou utilizar o Intellij Idea (https://www.jetbrains.com/idea/). Entendendo o Flyway Se você abrir o arquivo pom.xml, você irá ver a dependência do flyway-core: Dentro da estrutura do projeto você encontrará a pasta db.migration, nela vamos salvar os scripts do banco de dados. Quando o projeto iniciar uma das tarefas durante o processo de inicialização é verificar se há algum novo script no projeto, havendo qualquer script ele irá aplicar no banco de dados. Para criar nosso novo script, precisamos seguir alguns padrões no nome do arquivo. Esse padrão precisa ter um número que será incrementado a cada novo script. Dessa forma ajudamos o flyway a entender que se trata de um script não executado. Para simplificar esse tutorial vamos criar um simples script que irá seguir o padrão V1, V2, V3 para incrementar o arquivo. No final o nome é o seguinte: V1__create_base_tables.sql Criando nosso primeiro script Crie um novo arquivo chamado V1_create_base_tables.sql dentro da pasta db.migration e adicione o conteúdo abaixo: Configurando o banco de dados Para simplificar ainda mais o tutorial, eu utilizei o banco de dados H2 (banco em memória) para mostrar o funcionamento do flyway. Para configurar esse banco no nosso projeto, abra o arquivo pom.xml e adicione a seguinte dependência: Logo em seguida vamos colocar os dados para conexão no banco de dados. Dentro do arquivo application.properties adicione as seguintes configurações: Após isso podemos executar o projeto, nos logs vamos ver algo parecido com isso durante a execução: 2023-04-07 14:12:29.896 INFO 8012 --- [ main] o.f.c.i.database.base.BaseDatabaseType : Database: jdbc:h2:mem:testdb (H2 2.1) 2023-04-07 14:12:30.039 INFO 8012 --- [ main] o.f.core.internal.command.DbValidate : Successfully validated 1 migration (execution time 00:00.037s) 2023-04-07 14:12:30.055 INFO 8012 --- [ main] o.f.c.i.s.JdbcTableSchemaHistory : Creating Schema History table "PUBLIC"."flyway_schema_history" ... 2023-04-07 14:12:30.132 INFO 8012 --- [ main] o.f.core.internal.command.DbMigrate : Current version of schema "PUBLIC": << Empty Schema >> 2023-04-07 14:12:30.143 INFO 8012 --- [ main] o.f.core.internal.command.DbMigrate : Migrating schema "PUBLIC" to version "1 - create base tables" 2023-04-07 14:12:30.177 INFO 8012 --- [ main] o.f.core.internal.command.DbMigrate : Successfully applied 1 migration to schema "PUBLIC", now at version v1 (execution time 00:00.057s) 2023-04-07 14:12:30.477 INFO 8012 --- [ main] o.hibernate.jpa.internal.util.LogHelper : HHH000204: Processing PersistenceUnitInfo [name: default] Este log mostra que a execução da atualização do banco aconteceu com sucesso. Quando um novo script for adicionado, como por exemplo V2__new_tables.sql, o flyway irá executar somente o novo script V2. Observação: neste caso estamos utilizando um banco de dados em memória, toda vez que a aplicação parar, todos os dados contidos nela serão liberados. Portanto sempre iremos iniciar o banco do zero. Para os próximos tutoriais, vamos explorar cenários parecidos com casos reais. Conclusão O versionamento do banco de dados trás algumas vantagens como dar aos desenvolvedores uma cópia exata do banco de dados. Quando há alguma modificação a própria aplicação irá atualizar e gerenciar todas as mudanças, podendo aplicar tanto em desenvolvimento e produção. Referências Para mais detalhes você pode checar a documentação oficial do Spring: https://docs.spring.io/spring-boot/docs/current/reference/html/howto.html#howto.data-initialization.migration-tool.flyway Criando migrations: https://flywaydb.org/documentation/tutorials/baselineMigrations H2 database http://www.h2database.com/html/tutorial.html

Introdução AWS fornece diferentes maneiras de transformar dados por meio de seus serviços e um dos meus favoritos é o Kinesis Firehose Data Transformation. Essa é uma estratégia para transformar dados aproveitando o serviço de Streaming na entregar dos dados. Para este tutorial, vamos ensinar em como criar um Kinesis Firehose Data Transformation baseado na estratégia abaixo. O Kinesis Firehose enviará dados e, em vez de escrever no S3, invocará um Lambda para transformar esses dados e, em seguida, enviá-los de volta ao Kinesis Firehose, que entregará os mesmos dados ao S3. Criando o projeto Para este post usaremos Java como linguagem e Maven como gerenciador de dependências. Portanto, é necessário gerar um projeto Maven que criará a estrutura do nosso projeto. Se você não sabe como gerar um projeto Maven, recomendo ler o post Gerando um projeto Maven sem IDE em 2 minutos, onde mostro como gerá-lo. Estrutura do projeto Após gerar o projeto Maven e importá-lo para sua IDE, vamos criar os mesmos arquivos e pacotes conforme imagem ao lado, exceto o pom.xml que foi criado pelo gerador maven. Dentro da pasta java, crie um pacote chamado coffee.tips.lambda e crie também uma classe Java chamada Handler.java dentro deste mesmo pacote. Agora, crie um pacote chamado model dentro do pacote coffee.tips e então, crie duas classes java: Record.java Records.java Por fim, crie um novo pacote chamado status e também crie um enum chamada Status. Criando a classe Record Por que precisamos da classe Record? Kinesis Firehose espera um objeto como retorno contendo os campos acima. Isso acontece quando o Kinesis Firehose invoca o Lambda para transformar os dados e o mesmo Lambda deve retornar um objeto contendo esses campos preenchidos. recordId Este valor de campo deve conter o mesmo id vindo do Kinesis result Esse valor de campo controla o resultado do status da transformação. Os valores possíveis são: Ok: Registro transformado com sucesso. Dropped: O registro rejeitado intencionalmente de acordo com sua lógica de processamento. ProcessingFailed: Os dados não puderam ser transformados. data Este campo deve conter os dados após serem convertidos usando o algoritmo de conversão base64. Este modelo deve conter os seguintes parâmetros. Caso contrário, o Kinesis Firehose rejeita e define como falha na transformação. Criando a classe Records A classe Records será nossa classe Java onde será utilizada como retorno na classe Handler.java onde abordaremos mais a frente, contendo uma lista do tipo Record. Criando o Enum Status Decidi criar o Enum acima apenas para criar um código mais elegante, mas é útil quando precisamos mapear valores diferentes para um contexto específico. Criando a classe Handler A classe Handler será nosso controlador para o Lambda. Este Lambda será invocado pelo Kinesis Firehose passando alguns parâmetros contendo os dados a serem transformados. Observe que, para o método handleRequest, um parâmetro chamado input do tipo KinesisFirehoseEvent contém os registros enviados pelo Kinesis Firehose e o mesmo método retornará um objeto do tipo Records contendo uma lista de registros que posteriormente serão enviados de volta ao Kinesis Firerose entregando ao S3. Dentro da iteração usando o Java Stream, criamos algumas condições apenas para explorar como o campo de result funciona. Dependendo da condição, definimos o valor do result como Dropped, o que significa que os dados não serão entregues ao Kinesis Firehose. Caso contrário, para aqueles que foram definidos como Ok, os dados serão enviados para o Kinesis Firehose. Outro detalhe é que você pode alterar os valores durante a execução. Definimos "TECH" como o valor para o campo TICKER_SYMBOL quando o valor do campo SECTOR for TECHNOLOGY. É uma forma de transformar dados aplicando uma certa logica. Por fim, foram criados outros dois métodos apenas para decodificar e codificar os dados como requisito para o bom funcionamento do processamento. Atualizando o arquivo pom.xml Após gerar nosso projeto via Maven, precisamos adicionar algumas dependências e um plugin para empacotar o código e as bibliotecas para implantação. Seguindo o conteúdo do pom.xml abaixo: Criando recursos via Terraform Em vez de criar o Kinesis Firehose, Lambda, policies e roles manualmente por meio do console, vamos criar por meio do Terraform. Se você não conhece muito sobre Terraform, recomendo ver este tutorial Primeiros passos utilizando Terraform na AWS. Dentro da pasta terraform, crie os seguintes arquivos: Conteúdo do arquivo vars.tf O arquivo vars.tf é onde declaramos as variáveis. As variáveis ​​fornecem flexibilidade quando precisamos trabalhar com recursos diferentes. Conteúdo do arquivo vars.tfvars Agora precisamos definir os valores dessas variáveis. Então, vamos criar uma pasta chamada development dentro da pasta terraform. Após a criação da pasta. Crie um arquivo chamado vars.tfvars como na imagem e cole o conteúdo abaixo. Observe o campo bucket, você deve especificar o nome do seu próprio bucket. O nome do bucket deve ser único. Conteúdo do arquivo main.tf Para este arquivo, apenas declaramos o Provider. Provider é o serviço de nuvem que usaremos para criar nossos recursos. Neste caso, estamos usando a AWS como provider e o Terraform fará o download dos pacotes necessários para criar os recursos. Observe que, para o campo region, estamos usando a palavra-chave var para atribuir o valor da região já declarado no arquivo vars.tfvars. Conteúdo do arquivo s3.tf Este arquivo é onde declaramos os recursos relacionados ao S3. Nesse caso, estamos criando apenas o bucket S3. Mas, se você deseja criar mais recursos relacionados ao S3, como policies, roles e etc, pode declará-lo aqui. Conteúdo do arquivo lambda.tf O conteúdo abaixo será responsável pela criação do AWS Lambda e suas roles e policies. Observe que no mesmo arquivo estamos criando um recurso chamado aws_s3_object. É uma estratégia carregar o arquivo Jar diretamente no S3 após o empacotamento. Manter alguns arquivos no S3 é uma forma inteligente quando trabalhamos com arquivos grandes. Entendendo o conteúdo do lambda.tf 1. Declaramos aws_iam_policy_document que descrevem quais ações os recursos que serão atribuídos a essas políticas podem executar. 2. aws_iam_role fornece a função IAM e controlará algumas ações do Lambda. 3. Declaramos o recurso aws_s3_object porque queremos armazenar nosso arquivo Jar que será criado no S3. Assim, durante a fase de implantação, o Terraform obterá o arquivo Jar que será criado na pasta target e fará o upload para o S3. depends_on: O recurso atual somente será criado caso o recurso atribuído neste campo, já estivesse sido criado. bucket: É o nome do bucket onde também irá armazenar o arquivo Jar. key: Nome do arquivo Jar. source: Localização do arquivo de origem. etag: Aciona atualizações quando o valor muda. 4. aws_lambda_function é o recurso responsável por criar o Lambda e precisamos preencher alguns campos como: function_name: nome Lambda. role: Função do Lambda declarada nas etapas anteriores que fornece acesso aos serviços e recursos da AWS. handler: Neste campo você precisa passar o caminho da classe principal. source_code_hash: Este campo é responsável por acionar as atualizações do lambda. s3_bucket: É o nome do bucket onde também irá armazenar o arquivo Jar gerado durante o deploy. s3_key: Nome do arquivo Jar. runtime: Aqui você pode definir a linguagem de programação suportada pelo Lambda. Para este exemplo, java 11. timeout: Tempo limite de execução do Lambda. Lembrando que o timeout máximo é de minutos. 5. aws_iam_policy fornece políticas IAM para os recursos onde definimos algumas ações a serem executadas. Nesse caso, definimos ações como invocação do Lambda e habilitação aos logs no CloudWatch. 6. Para o recurso aws_iam_role_policy_attachment, podemos anexar policies do IAM às roles. Nesse caso, anexamos as policies lambda_iam_role criadas anteriormente. 7. Finalmente, temos o recurso aws_lambda_permission. Precisamos desse recurso para dar permissão ao Kinesis Firehose para invocar o Lambda. Conteúdo do arquivo kinesis.tf Entendendo o conteúdo do kinesis.tf 1. Declaramos o recurso aws_kinesis_firehose_delivery_stream e seus campos, seguindo os detalhes: destination: Esse é o próprio destino, o Kinesis fornece um mecanismo para entregar dados ao S3(extended_s3), Redshift, Elasticsearch (OpenSearch serviço AWS), splunk e http_endpoint. name: Nome Kinesis Firehose. depends_on: O recurso atual somente será criado caso o recurso atribuído neste campo, já estivesse sido criado, ou seja, Kinesis Firehose será criado se o S3 Bucket já existir. extended_s3_configuration: 1. bucket_arn: S3 Bucket utilizando o prefixo arn. 2. role_arn: ARN role. 3. prefix: Pasta do S3 Bucket onde os dados serão armazenados. Você pode especificar o formato de pastas usando as seguintes expressões, "/year=! {timestamp:yyyy}/month=!{timestamp:MM}/". 4. error_output_prefix: Para este campo, você pode definir um caminho para armazenar os resultados da falha do processo. 5. buffer_interval: Buffer do Kinesis Firehose utilizando configurações de intervalo para a entrega do dado. 6. buffer_size: Buffer do Kinesis Firehose utilizando configurações de tamanho de dados para a entrega do dado. 7. compression_format: Existem algumas opções de formato de compactação como ZIP, Snappy, HADOOP SNAPPY e GZIP. Para este tutorial, escolhemos GZIP. processing_configuration:Esse é o bloco onde definimos qual recurso será processado. Para este caso, AWS Lambda. 1. enabled: true para habilitar e false para desabilitar. 2. type: Tipo do processador. No caso, Lambda. 3. parameter_value: Nome da função Lambda com prefixo arn. 2. Declaramos aws_iam_policy_document que descrevem quais ações os recursos que serão atribuídos a essas políticas podem executar. 3. aws_iam_role recurso que fornece a função do IAM para controlar algumas ações do Kinesis. 4. aws_iam_policy fornece IAM policies para os recursos onde definimos algumas ações a serem executadas. Nesse caso, definimos S3 e algumas ações do Lambda. 5. Para o recurso aws_iam_role_policy_attachment, podemos anexar policies do IAM às roles. Nesse caso, anexamos as policies firehose_iam_role e firehose_policies criadas anteriormente. Packaging Criamos nosso projeto Maven, classe Handler com arquivos Java e Terraform para criar nossos recursos na AWS. Agora, vamos executar os seguintes comandos para o deploy do nosso projeto. Primeiro, abra o terminal e verifique se você está no diretório raiz do projeto e execute o seguinte comando maven: mvn package O comando acima empacotará o projeto criando o arquivo Jar a ser implantado e carregado no S3. Para ter certeza, verifique a pasta target e veja se alguns arquivos foram criados, incluindo o arquivo lambda-kinesis-transform-1.0.jar. Executando Terraform Agora, vamos executar alguns comandos do Terraform. Dentro da pasta terraform, execute os seguintes comandos no terminal: terraform init O comando acima iniciará o terraform, baixando as bibliotecas do terraform e também validará os arquivos do terraform. Para o próximo comando, vamos executar o comando plan para verificar quais recursos serão criados. terraform plan -var-file=development/vars.tfvars Após a execução, você verá logs semelhantes no console: Por fim, podemos solicitar a criação dos recursos através do seguinte comando: terraform apply -var-file=development/vars.tfvars Após a execução, você deve confirmar para executar as ações, digite "sim". Agora a provisão foi concluída! Enviando mensagens Bem, agora precisamos enviar algumas mensagens para serem transformadas e vamos enviá-las via console do Kinesis Firehose. Obviamente existem outras formas de enviar, mas para este tutorial vamos enviar pela forma mais fácil. Abra o console do Kinesis Firehose, acesse a opção Delivery Stream conforme a imagem abaixo. Na seção Test with demo data, clique no botão Start sending demo data para iniciar o envio. Após clicar, as mensagens serão enviadas pelo Kinesis Firehose e de acordo com as configurações do buffer, o Kinesis levará 2 minutos para entregar os dados ou caso atinja 1 MIB em quantidade de dados. Vamos dar uma olhada no nosso Lambda e ver as métricas: Clique na guia Monitor e depois, na opção Metrics e observe que o Lambda está sendo invocado e não há erros. Resultado dos Dados Transformados Agora que sabemos que tudo está funcionando bem, vamos dar uma olhada nos dados transformados diretamente no Amazon S3. Vá e acesse o S3 Bucket criado. Observe que muitos arquivos foram criados. Vamos ler um deles e ver os dados transformados. Escolha um arquivo como na imagem abaixo e clique no botão Actions e depois na opção Query with S3 Select. Seguindo as opções selecionadas na imagem abaixo, clique no botão Run SQL query para ver o resultado. Com base na imagem acima, você pode ver que, de acordo com a classe Handler.java, definimos um algoritmo para descartar dados com valor do campo CHANGE menor que zero e para aqueles com valor de campo SECTOR igual a TECHNOLOGY, definiríamos o valor de campo TICKER_SYMBOL como TECH. Este foi um exemplo de como você pode transformar dados usando Kinesis Firehose Data Transformation e Lambda como um componente barato para transformar dados. Parar de enviar mensagens Você pode parar de enviar mensagens antes de destruir os recursos criados via Terraform procurando economizar dinheiro. Então, basta voltar ao console do Kinesis Firehose e clicar no botão Stop sending demo data. Destroy As cobranças da AWS ocorrerão se você não destruir esses recursos. Portanto, recomendo destruí-los evitando algumas cobranças desnecessárias. Para evitá-lo, execute o comando abaixo. terraform destroy -var-file=development/vars.tfvars Lembra que você precisa confirmar essa operação, ok? Conclusão Definitivamente, Kinesis Firehose não é apenas um serviço de Streaming. Existe flexibilidade integrando os serviços da AWS e a possibilidade de entregar dados para diferentes destinos possibilitando a transformação de dados e aplicando a lógica de acordo com o seu caso de uso. Repositório GitHub Espero que tenha curtido!