Orquestrando scripts python utilizando o pacote/biblioteca LUIGI

Em um outro post, falei de uma biblioteca chamada Pandas que serve para fazer a transformação de dados em Python. Hoje, vamos falar de uma alternativa de orquestrar Pipelines. Vejamos a biblioteca Luigi.

Vamos imaginar que você tenha dividido o seu trabalho da seguinte maneira:

$ python extracao_dos_dados.py

$ python limpeza_dos_dados.py

$ python juncao_dos_dados.py

$ python fazer_coisas_com_os_dados.py

Estas atividades são muito comuns em projetos de dados. Cada script realiza uma atividade, tudo "organizadinho". Esse conjunto de atividades é conhecido, no jargão da engenharia de dados como Pipeline. Daí toda vez que você for executar a pipeline como um todo, terá que executar cada um dos scripts. Quando estamos falando de um ou dois conjuntos de dados beleza, mas sabemos que não é assim que funcionam as coisas. Para cada conjunto de dados teríamos 4 scripts para realizar o processo mencionado acima. A medida que as pipelines se multiplicam, a situação fica muito complicada.

Daí um colega sugere de botar todas essas atividades num script só e depois uma rápida refatoração, o script faz_tudo.py pode ficar assim:

if __name__ == '__main__':

extracao_dos_dados()

limpeza_dos_dados()

juncao_dos_dados()

fazer_coisas_com_os_dados()

   

Isso é bastante simples de executar: $ python faz_tudo.py

Poderíamos também colocar tudo em um script bash, que chamaria esses scripts todos, em sequência, mas as deficiências serão mais ou menos as mesmas.

Ao avançar para um pipeline pronto para produção, há mais alguns aspectos a serem considerados além do código de execução de tudo. Em particular, o tratamento de erros deve ser levado em consideração:

try:

    extracao_dos_dados()

except ExtracaoDeDadosError as e:

    # handle this

Mas se encadearmos todas as tarefas individuais, acabamos com uma árvore de Natal de try/except:

try:

    extracao_dos_dados()

    try:

        limpeza_dos_dados()

        try:

            # Rapaz, aonde isso vai parar!

        except EvenMoreErrors:

            # ...

    except limpeza_dos_dadosError as e:

        # tratando limpeza_dos_dadosError

except extracao_dos_dadosError as e:

    # tratando extracao_dos_dadosError

Outro aspecto importante a considerar é como retomar um pipeline. Por exemplo, se as primeiras tarefas forem concluídas, mas ocorrer um erro no meio do caminho, como executaremos novamente o pipeline sem executar novamente as etapas iniciais bem-sucedidas? Poderíamos tratar com if.

# testando se uma tarefa já foi executada com sucesso.

if not i_got_the_data_already():

    # if not, run it

    try:

       limpeza_dos_dados()

    except limpeza_dos_dadosError as e:...

Percebe o quanto pode ficar complicado esse controle das pipelines, caso a pipeline seja implementada desta forma?

Para resolver essa situação de uma maneira mais elegante, eis que surgem diversas ferramentas para tratar esse problema entre elas está o pacote Luigi.

Luigi é uma ferramenta Python para gerenciamento de fluxo de trabalho. Ele foi desenvolvido pela Spotify, para ajudar a construir pipelines de dados complexos de trabalhos em lote. 

Alguns dos recursos úteis do Luigi incluem:

• Gerenciamento de dependência
• Pontos de verificação/recuperação de falhas
• Integração / parametrização CLI
• Visualização do gráfico de dependência

Existem dois conceitos principais para entender como podemos aplicar o Luigi ao nosso próprio pipeline de dados: Tarefas e Destinos. Uma tarefa é uma unidade de trabalho, projetada estendendo a classe luigi.Task e substituindo alguns métodos básicos. A saída de uma tarefa é um destino, que pode ser um arquivo no sistema de arquivos local, um arquivo no S3 da Amazon, ou ainda, algum dado em um banco de dados..

As dependências são definidas em termos de entradas e saídas, ou seja, se tarefa2 depende da tarefa1, significa que a saída da tarefa1 será a entrada da tarefa2.

Então vamos deixar de bla bla bla, partindo para a parte prática.

Para instalar o Luigi: $ pip install luigi

""" Trata-se de scripts que contem classes que extendem a

a classe luigi task

"""

import luigi

 

 #métodos que precisam ser implmentados

class Tarefa1(luigi.Task):

 #o método requires recebe as tarefas das quais ela depende.

 # neste caso a tarefa 1 não depende de nenhuma outra tarefa

    def requires(self):

        return []

 #o método output é a saída da tarefa que pode ou não virar entrada de

 #outra tarefa.

    def output(self):

        return luigi.LocalTarget("numbers_up_to_10.txt")

 #o método run é a implementação da tarefa o que ela faz.

    def run(self):

        with self.output().open('w') as f:

            for i in range(1, 11):

                f.write("{}\n".format(i))

 

class Tarefa2(luigi.Task):

    #veja que a tarefa2 depende da tarefa 1, ou seja, a tarefa 1 tem

    #que rodar primeiro para depois a tarefa dois rodar.

    def requires(self):

        return [Tarefa1()]

 

    def output(self):

        return luigi.LocalTarget("squares.txt")

 

    def run(self):

        with self.input()[0].open() as fin, self.output().open('w') as fout:

            for line in fin:

                n = int(line.strip())

                out = n * n

                fout.write("{}:{}\n".format(n, out))

                 

if __name__ == '__main__':

    luigi.run()

# para executar $ luigi -m run_luigi.py SquaredNumbers --local-scheduler

Este código apresenta duas tarefas: tarefa1 , que grava o número de 1 a 10 em um arquivo chamado numbers_up_to_10.txt , um número por linha, e tarefa2 , que lê esse arquivo e gera uma lista de pares number-square em squares.txt , também um par por linha.

Salve esse código em um arquivo chamado dag_luigi e execute o mesmo através da linha de comando.

$ luigi -m dag_luigi.py tarefa2 --local-scheduler

O primeiro argumento que estamos passando para o arquivo é o nome da última tarefa no pipeline que queremos executar. O segundo argumento simplesmente diz ao Luigi para usar um agendador local .

O Luigi cuidará de verificar as dependências entre tarefas, se a entrada de tarefa2 não estiver lá, então ele executará a tarefa tarefa1 primeiro, depois continuará com a execução.

Para criar uma tarefa, simplesmente precisamos criar uma classe que estenderá a classe luigi.Task e sobrescrever alguns métodos. Em particular:

• require() deve retornar a lista de dependências para uma determinada tarefa — em outras palavras, uma lista de tarefas
• output() deve retornar o destino da tarefa (por exemplo, um arquivo local, ou um a inserção dos dados em um Banco).
• run() deve conter a lógica para executar

Luigi verificará os valores de retorno de require() e output() e construirá o gráfico de dependência entre as tarefas.

Para ver o gráfico de dependências e acompanhar a execução das tarefas,  o Luigi disponibiliza uma interface web que pode ser acessada através do endereço localhost:8082. Para disponibilizar essa interface rode no pronpt de comando a seguinte instrução em foreground $luigid ou em background $luigid --background

Interface web do luigi

Nem tudo são as mil maravilhas, o agendamento da execução da DAG é feita através do velho e bom Cron.

Existem outras alternativas e o apache Airflow é uma delas.

Bom pessoal, era isso. Obviamente tem mais coisa para ser estudada sobre essa biblioteca veja a documentação dela em https://luigi.readthedocs.io/en/stable/index.html

Datalake (Uma versão BIG DATA do DW clássico)

Houve um tempo em que era possível gerar relatórios gerenciais com dados extraídos das bases transacionais que subsidiam os sistemas de informação. Com o aumento do volume de dados, constatou-se que as consultas envolvidas na produção desses relatórios estavam comprometendo o desempenho dos sistemas.

No sentindo de resolver essa questão eis que surgem as bases de dados analíticas com dados extraídos de bases transacionais, modelados de forma dimensional de modo a otimizar a performance das consultas efetuadas. Surge aí o conceito de Data Warehouse,

Agora vamos imaginar um cenário com uma grande VARIEDADE de dados (estruturados, semiestruturados e não estruturados).

Quando se fala em BIG DATA, verifica-se que o mercado e as ferramentas até então usadas para trabalhar com dados, precisaram ser adaptadas para processar um volume e uma variedade maior de dados. E o quesito análise de dados não poderia ser deixado de lado. 

Há quem diga que um DATALAKE é uma versão moderna do DW clássico surgido nos anos 90. Assim como o Data warehouse, um Datalake é um armazém de dados corporativo com a diferença de que neste último, os dados contidos podem ser estruturados, semiestruturados, e não estruturados. Além disso, esse armazém de dados é hospedado em um ambiente distribuído (dados hospedados em um sistema de arquivos distribuído), escalável e tolerante a falhas.

Outra diferença entre Data Warehouse  e Datalake é a ordem de como as coisas acontecem. No caso do DW, as atividades do processo de trabalho é a extração de dados de banco de dados transacionais, transformação desses dados para um formato dimensional e em seguida, a carga desses dados em tabelas fato e tabelas dimensão, presentes no DW. No caso do Datalake, primeira é feita a ingestão  dos dados(extração), em seguida ocorre a carga desses dados no lake para a partir daí se faça a transformação destes dados. Esse processo é conhecido como ELT.

A estruturas que recebem os dados varia de projeto, mas basicamente é previsto uma área para receber os dados brutos, outra área para realização de transformações sobre os dados e por fim, uma área para manter um histórico dos dados. 

A ingestão dos dados para essas estruturas pode ser feita em lote (batch), aonde acumula-se os dados por um tempo ou ainda a  medida que os dados são gerados, sendo capturados em tempo real(streaming).  O Spark Streaming ou ainda um serviço de mensageria (Kafka ou RabbitMQ) podem ser utilizadas como ferramenta para apoiar o trabalho.

A medida que esses dados são trabalhados vão sendo armazenados nas devidas camadas. utilizando formatos diversos de BIG DATA( Parquet, ORC, AVRO) visando economia de espaço em disco e velocidade de consulta. Para apoiar esse trabalho, geralmente utiliza-se uma engine de processamento de BIG DATA (Spark, Hadoop, Flink, entre outras) e alguma ferramenta de orquestração das atividades envolvidas( apache airflow, Oozie, etc).

No caso do DW os dados são consultados na bases dimensionais do mesmo. E no caso do DL como disseminar a informação? Podemos indexar esses dados numa ferramenta de busca como o Elastic, armazenar os dados em um banco de dados NoSQL, ou ainda, jogar os dados tratados sob o formato de arquivo, numa ferramenta do tipo DMS (CKAN).

Com os dados tratados, é possível disponibilizar a informação através de ferramentas de visualização do tipo Self Service BI (Tableau, Qlik, PowerBI, apache superset, KIBANA entre outras), mas aí é assunto para outro post. 

Inté!

Processamento de dados em larga escala Spark

Introdução

Chegamos a um ponto que os tradicionais bancos relacionais já não está atendendo as demandas de análise de dados. Por que digo isso? Com a popularização da Internet, advento da IOT (Internet das coisas), o volume de dados produzidos, seja de forma estruturada, não estruturada ou semi estruturada cresceu exponencialmente. Com isso foram surgindo tecnologias para suprir essas necessidades do mercado, capazes de processar grandes volumes de dados de forma paralelizada, descentralizada e distribuída e o Spark é uma alternativa criada pela a indústria para realizar esse tipo de processamento.

A solução mais conhecida e amada pelos profissionais de engenharia de dados era o Hadoop, contudo as operações realizadas através dessa solução dependem muito de operações de IO, representando um dos grandes gargalos do desempenho dos bancos de dados. O Spark resolve isso evitando essas operações, mantendo os conjuntos de dados em memória principal.

O que é?

O SPARK é um framework de código aberto para computação distribuída. Trata-se de um mecanismo que possibilita o processamento (análise) de grande volume de dados. Podemos entender a computação distribuída como um sistema que interliga vários computadores, conseguindo-se assim um grande poder de processamento. Esse conjunto de computadores é conhecido como CLUSTER e cada computador que faz parte desse conjunto é chamado de NÓ. Esse poder de processamento se torna mais eficiente se cada nó faz parte do trabalho. No Spark utiliza-se a estratégia DIVIDIR PARA CONQUISTAR.

Exemplificando, vamos supor que você tenha a tarefa de contar o número de M & M contidas em um pote. Você poderia contar sozinho (que daria muito trabalho) ou pedir ajuda de uns amigos para ajudar na contagem facilitando o trabalho. A atividade seria dividir os M & M's contidos no pote entre os amigos. Cada amigo contaria os M & M's que estão com eles e no final, você somaria as quantidades contabilizadas por você e pelos seus amigos para chegar ao resultado final. Essa é a estratégia adotada pelo Spark.

O Spark faz a divisão do conjuntos de dados, distribuindo os mesmos em estruturas conhecidas como  RDD's(Resilient, Distributed, Dataset).que ficam em memória principal e são imutáveis. A cada novo processamento, novos RDD's são criados de modo a refinar o processamento para no final, o resultado pro processamento ser gravado em disco.

Spark Core é o mecanismo de execução geral da plataforma Spark sobre o qual todas as outras funcionalidades são construídas. O Spark suporta desenvolvimento em várias linguagens como Java, Scala, Python, R e SQL. Além disso existem vários módulos dedicados a diferentes tipos de aplicação, como por exemplo SparkSQL que dá suporte escrever em linguagem SQL em vez de usar a api do Spark. Tem uma módulo chamado Spark Streaming que dá suporte a processamento de dados em tempo real. Tem a biblioteca de machine learning chamada SparkML com diversos modelos para serem utilizados, além de uma mecanismo para computação Gráfica chamada GraphX. Todo esse ecossistema contribuiu para o seu crescimento acelerado e ajudou o Spark a ser um dos sistemas de processamento massivo paralelo mais usados na área de Big Data.

No caso do Python, para interagir com o Spark é utilizada uma  biblioteca chamada PySpark.

Glossário do Spark

RDD (Resilient Distributed Datasets)- É a principal estrutura de dados do Spark. Um RDD é uma coleção de elementos particionada e imutável , o que significa que contém valores, e esses elementos são particionados para serem utilizados em sistemas distribuídos e não podem ser alterados. Não possuem uma estrutura de colunas e sim de linhas e exibidos como se fossem uma lista.

Spark Dataframe - Tem todas as características do RDD porém os dados estão organizados em colunas como uma tabela em um banco de dados. Foi estruturado para tornar mais fácil o processamento de grandes quantidade de dados. Apesar de estruturalmente serem diferentes, para quem conhece a estrutura de um dataframe da biblioteca de Pandas, a manipulação se torna mais fácil e por este motivo é a melhor escolha para começar a usar o PySpark. Existem algumas mudanças sutis entre o Spark dataframe e o Spark dataset que falaremos a seguir, mas o importante a saber quando estiver usando o Dataframe , estará aproveitando o formato interno otimizado pelo Spark e este formato aplica ganho de eficiência em todas as linguagens usadas pelas API's.

Spark Dataset - Eles são semelhantes aos DataFrames, mas são fortemente tipados, o que significa que o tipo é especificado na criação do DataSet e não é inferido o tipo de registros armazenados nele. Os Datasets estão disponíveis apenas para o Java Virtual Machine baseado nas linguagens Java e Scala e nele você define o tipo de dados que serão inseridos no Dataset.

Transformações- são as operações que podemos fazer em um DataFrame no Spark. É importante observar que as transformações criam novos RDDs porque, lembre-se, os RDDs são imutáveis, portanto, não podem ser alterados de depois de criados. Então as Transformações pegam um RDD como uma entrada e executam alguma operação e gera um ou mais RDDs. 

Como o Spark trabalha com Lazy evaluation , conforme um compilador verifica cada transformação, ele não cria nenhum RDDs novo, mas sim uma cadeia de RDDs e os resultados dessas transformações que são armazenados no DAG ,que só serão avaliadas uma vez quando uma uma Ação é chamado. Essa cadeia de RDDs ou “filhos”, todos conectados logicamente ao RDD “pai” , é chamado de gráfico de linhagem

Ações - uma ação é qualquer operação RDD que não produz um RDD como saída. Alguns exemplos de ações comuns são fazer uma contagem dos dados, máximo ou mínimo, retornar o primeiro elemento de um RDD, etc. Como foi mencionado antes, uma ação é a mensagem para o compilador avaliar o gráfico de linhagem e retorna o valor especificado pela ação.

Gráfico de linhagem - Um gráfico de linhagem descreve o que é chamado de “plano de execução lógico”. O que isso significa é que o compilador começa com os primeiros RDDs que não dependem de nenhum outro RDD e segue uma cadeia lógica de transformações até terminar com o RDD em que uma ação é chamada. Esse recurso é principalmente o que impulsiona a tolerância a falhas do Spark. Se um nó falhar por algum motivo, todas as informações sobre o que esse nó deveria estar fazendo são armazenadas no gráfico de linhagem, que pode ser replicado em outro lugar.

 Como o Spark executa uma tarefa

Você que é engenheiro de dados acabou de codificar um programa para ser executado no Spark. 

Ao colocar seu programa para executar, ele será executado como conjunto independente de processos em um cluster, coordenados por um programa principal (chamado de driver ).

Especificamente, para ser executado em um cluster, o driver pode se conectar a vários tipos de gerenciadores de cluster (seja o gerenciador de cluster do Spark, Mesos ou YARN), que alocam recursos para o seu programa. 

Uma vez conectado, o Spark adquire executores em nós no cluster, que na verdade são processos  ficam aguardando receberem uma ordem do driver para executar cálculos e armazenam dados para seu programa. Em seguida, o Spark envia o código do seu programa para os executores. Finalmente, o driver envia tarefas para os executores executarem.

Cada programa obtém seus próprios processos executores, que permanecem ativos durante todo o programa e executam tarefas em vários threads.

O driver deve escutar e aceitar conexões de entrada de seus executores ao longo de sua vida útil e como o driver agenda tarefas no cluster, ele deve ser executado próximo aos nós de trabalho, de preferência na mesma rede local.

É importante frisar que o Spark é um dos engines de big data mantidos pela a Apache Foundation Existem outros! Hadoop, Flink, Storm, Samza.

Gostou do post? Falaremos de outras soluções de processo aqui no blog.

O Ecossistema Hadoop

Até pouco tempo atrás, empresas trabalhavam com sistemas integrados operando em bancos de dados relacionais. Alguns desses sistemas, produziam relatórios com base nesses bancos. Outros, produziam relatórios a partir de armazéns de dados analíticos conhecidos como Datawarehouses.

Hoje o cenário mudou bastante. Estamos vivendo na era da informação em um cenário de BIG DATA Praticamente tudo gera dado(redes sociais, dispositivos do tipo "Internet das coisas", aviões, carros, telefones, entre outros). As vezes, os dados são estruturados, algumas vezes semiestruturados e  muitas vezes, os dados não são estruturados.

Por conta deste novo cenário, o volume de dados e a velocidade de produção desses dados aumentou exponencialmente e consequentemente, a complexidade para mantê-los aumentou proporcionalmente. 

A arquitetura clássica empregada nas soluções de processamento desses dados já não atendia mais a demanda. Verificou-se que toda normalização e segurança providas pelos SGBD's relacionais comprometia do desempenho da análise de dados.

Era necessário processar dados em tempo real, processar as informações de forma distribuída, além de possibilitar a análise de dados não estruturados.

No sentido de atender a essas necessidades, eis que surge um conjunto de ferramentas que fazem parte de um ecossistema conhecido como HADOOP.

O Hadoop foi inspirado na publicação de MapReduce , GoogleFS e BigTable of Google . Ele foi criado por Doug Cutting e faz parte dos projetos da Apache Software Foundation desde 2009.

Esse ecossistema é uma estrutura de código aberto livre escrita em Java destinada a facilitar a criação de aplicativos distribuídos (armazenamento e processamento de dados) e escalonáveis ​​(escalonáveis) , permitindo que os aplicativos trabalhem com milhares de nós e petabytes de dados. Cada nó é, portanto, composto de máquinas padrão agrupadas em um cluster. Todos os módulos do Hadoop são projetados com a ideia de que as falhas de hardware são frequentes e, portanto, devem ser tratadas automaticamente pela estrutura.

O núcleo do Hadoop consiste em uma parte de armazenamento: HDFS ( Hadoop Distributed File System ) e uma parte de processamento chamada MapReduce. O Hadoop divide os arquivos em grandes blocos e os distribui pelos nós do cluster. Para processar os dados, ele transfere o código para cada nó e cada nó processa os dados que possui. Isso torna possível processar todos os dados de forma mais rápida e eficiente do que em uma arquitetura de supercomputador mais tradicional, que conta com um sistema de arquivos paralelo onde cálculos e dados são distribuídos em redes de alta velocidade.

O HDFS é o software responsável pela gestão dos computadores do cluster, definindo como os arquivos serão distribuídos pelos nós que o compõe.

Componentes do HADOOP

Map Reduce

Para manejar grandes volumes de dados e extrair o máximo do big data, o Hadoop conta com um algoritmo, também introduzido pelo Google, chamado Map Reduce que facilita a distribuição e execução de uma tarefa, paralelamente. Sua missão, é basicamente, dividir uma tarefa em várias e processar essas tarefas em máquinas diferentes.

Utilizando-me do ditado popular, ele divide para conquistar e executa os processos nas máquinas do cluster, evitando assim um tráfego intenso de rede.

A forma como o MapReduce funciona pode ser dividida em três fases, com uma quarta fase como opção.

Mapper: Nesta primeira fase, a lógica condicional filtra os dados em todos os nós em pares de valores-chave. A “chave” refere-se ao endereço de deslocamento para cada registro e o “valor” contém todo o conteúdo do registro.

Shuffle: Durante a segunda fase, os valores de saída do mapeamento são classificados e consolidados. Os valores são agrupados de acordo com chaves semelhantes e os valores duplicados são descartados. A saída da fase shuffle também é organizada em pares de valores-chave, mas desta vez os valores indicam um intervalo em vez do conteúdo em um registro.

Reducer: Na terceira fase, a saída da fase Shuffle consolidada é agregada, com todos os valores adicionados às suas chaves correspondentes. Isso é então combinado em um único diretório de saída.

Combiner: A execução desta fase pode otimizar o desempenho do trabalho do MapReduce, fazendo com que os trabalhos fluam mais rapidamente. Ele faz isso pegando as saídas do mapper e examinando-as no nível do nó em busca de duplicatas, que são combinadas em um único par k-v, reduzindo assim o trabalho realizado pela fase shuffle.

HDFS – Hadoop Distribuited File System

Para possibilitar a aplicação do algorítmo de Map Reduce, foi necessária a criação de um de uma novo sistema de arquivos conhecido com HDFS. O HDFS é o componente que armazena dados em formato de arquivos no Hadoop.

A sua arquitetura possui dois componentes principais que são o name node e o data node.

Name node:  é o nó mestre e há apenas um por cluster. Sua tarefa principal é gerenciar os arquivos e os blocos armazenados em cada cluster. Ele basicamente mapeia a localização, faz a divisão dos arquivos em blocos e controla a localização de suas réplicas.  

Data Nodes: são nós escravos e podem ter vários por cluster. Sua missão é recuperar os dados quando requisitado, de onde eles estiverem. São eles que fazem o armazenamento efetivo dos dados e podem conter inúmeros blocos de diferentes arquivos. Eles estão sempre reportando ao name node quais blocos estão guardando e todas alterações que foram efetuadas neles. 

O orquestrador YARN

O YARN ou Yet Another Resource Negotiator gerencia os recursos no cluster e as aplicações no Hadoop. Ele permite que os dados armazenados no HDFS possam ser processados em vários engines como processamento em lote, streaming e etc.

Segundo a Wikipedia, ele se divide basicamente em duas funcionalidades de gerenciamento de recursos e agendamento/monitoramento de jobs em daemons separados.

Em resumo, ele coordena como as aplicações são executadas.

HBASE um banco de dados NoSQL

O HBase é um banco de dados NoSQL, open-source com estrutura colunar.

Roda com o HDFS e pode trabalhar com diversos formatos de dados, permitindo o processamento em tempo real e randômico de leitura/gravação nos dados.

A função principal dele é hospedar grandes tabelas – bilhões de linhas x milhões de colunas – sobre clusters de hardware comum, muito semelhante ao HDFS.

Simplificando a análise de dados com  o Apache Pig

Esse componente do ecossistema Hadoop originou-se de um projeto desenvolvido pelo Yahoo, por volta de 2006 por conta da necessidade de ser ter um mecanismo para execução de jobs MapReduce de maneira ad-hoc, ou seja, de forma mais simplificada para os usuários.

O Pig foi desenvolvido para analisar grandes conjuntos de dados e simplificar a escrita de funções map reduce ou Spark. Ele está estruturado em dois componentes: Pig Latin e Pig engine.

Pig Latin é uma linguagem de script semelhante ao SQL e o Pig Engine é o motor de execução do que é desenvolvido em Pig Latin. Internamente, o script desenvolvido em Pig Latin é convertido em funções MapReduce, tornando a vida dos programadores que não conhecem Java muito mais fácil.

Hive o data warehouse do Hadoop

O Hive é um sistema de data warehouse distribuído desenvolvido pelo facebook.

Ele permite de maneira simples, ler, gravar e gerenciar grandes arquivos no HDFS. Tem sua própria linguagem de consulta chamda HQL – Hive Querie Language – que é muito similar à linguagem SQL.

Ela simplifica a escrita de funções MapReduce usando a linguagem HQL.

Importando e exportando dados de bases relacionais com Sqoop

Sabemos que muitos dados estão armazenados em bancos de dados relacionais. Como eles já estão há muito tempo no mercado, são uma importante fonte de dados. e é aqui que o Sqoop desempenha um papel importante para trazer esses dados dos bancos estruturados para dentro do HDFS.

Todos comandos escritos internamente no Sqoop são convertidos em tarefas MapReduce que são carregadas no HDFS, tem compatibilidade com a maioria dos bancos relacionais (Oracle, SQL Server, MySQL, PostGree e etc.) e pode exportar dados do HDFS para esses bancos.

 Coletando dados em tempo real (streaming) com o Flume

O Flume é software de código aberto, confiável e distribuído utilizado para coletar, agregar e mover grandes quantidades de dados para o HDFS. Segundo a Wikipedia, possui uma arquitetura simples e flexível baseada em fluxo de dados de streaming. Ela é bastante robusta e tolerante a falhas pois possui vários mecanismos de failover e recuperação caso seja necessário.

Ele pode coletar os dados em real time ou batch.

Mensageria com o Kafka

Existem inúmeras aplicações gerando dados e muitas outras consumindo-os, porém conectá-las individualmente é uma tarefa bastante difícil. Quando essa necessidade aparece é que entra o Kafka.

Ele fica entre as aplicações que geram dados (producers/produtores) e as que consomem os dados (consumers/consumidores).

A plataforma, segundo a Wikipedia, tem por objetivo entregar uma plataforma integrada e de baixa latência para tratamento de dados em tempo real.

O Kafka tem processamento distribuído, replicação e tolerância a falhas nativo. Ele pode lidar com streaming de dados e permite a análise de dados em tempo real. Ele foi originalmente desenvolvido pelo Linkedin e teve seu código aberto no início de 2011.

 Scheduller com o Oozie

O Oozie é um agendador de workflows que permite aos usuários fazerem o agendamento dos jobs desenvolvidos em várias plataformas, como MapReduce, Pig, Hive e etc.

Usando o Oozie é possível criar um job que possa chamar de maneira orquestrada, outros jobs ou pipelines de dados, seja de maneira sequencial ou paralela para executar uma determinada tarefa.

É um produto open-source, confiável e escalável que auxilia muito as tarefas de quem usa o ecossistema Hadoop.

Zookeeper

Como já vimos manter um ambiente Hadoop é bastante desafiador. Sincronizar, coordenar e manter as configurações de um ambiente de cluster Hadoop exige bastante esforço, para resolver esse problema entra em cena o Zookeeper.

Ele é um software open-source, distribuído, e com um serviço centralizado para manter as informações de configuração, naming , sincronização distribuída e um grupo de serviços para todo o cluster.

Processando dados em memória principal com o Spark

O Spark é um framework de código aberto, alternativo ao Hadoop, desenvolvido em Scala que também oferece suporte a diversas aplicações escritas em Java, Python(Pyspark) e etc.

Comparado com o MapReduce, ele executa o processamento em memória aumentando consideravelmente a velocidade de execução desses processos.

Outro ponto importante a ser considerado é que além dele executar processos em batch igual o Hadoop, ele também consegue trabalhar em tempo real.

Além disso, o Spark tem seu próprio ecossistema:

Spark Core: é o core da aplicação, o principal mecanismo de execução do Spark e outras API´s construídas sobre ele.

Spark SQL API: permite fazer queries/consultas em dados estruturados em data frames ou tabelas Hive.

Streaming API: permite o Spark lidar com dados em tempo real. Ele se integra facilmente com uma variedade de fontes de dados como Kafka, Flume, Twitter e etc.

MLlib: é uma biblioteca de machine learning escalável que permite executar algoritimos de ciência de dados aproveitando as funcionalidades do Spark ao mesmo tempo, sem perda drástica de performance.

GraphX: é um engine de computação gráfica que permite os usuários interagir, construir e transformar dados em gráficos e vem com uma biblioteca de algorítimos comuns.

Etapas de um processo de Big Data com Hadoop

• Flume, Kafka, Sqoop são utilizados para fazer a ingestão de dados no HDFS.
• O HDFS é a unidade de armazenamento do Hadoop. Até os dados importados do HBase também são armazenados no HDFS.
• MapReduce e Spark são usados para processar os dados no HDFS e executar várias outras tarefas.
•  Pig, Hive e Spark são utilizados para analisar dados.
• Oozie ajuda a agendar as tarefas e por funcionar com várias plataformas, é utilizado ao longo das etapas.
• Zookeeper sincroniza todos os nós do cluster e é utilizado em todas as etapas.

Ficou grande, mas considero um bom resumo sobre tecnologias que foram criadas para trabalhar com grande volume de informações.

Inté!

Análise de dados na prática com o Pandas

 Lembra daquele trabalho que a gente fazia para promover a transformação e a higienização de dados no Excel?  Não tínhamos problema em trabalhar dessa forma enquanto o volume de dados era pequeno. Porém na era do Big Data planilhar e analisar os dados no excel virou um problema. O pacote Pandas surgiu para resolver esse problema. 

A biblioteca Pandas nos permite fazer análises exploratórias em conjunto de dados. Ela dá ao Python, a capacidade de trabalhar com dados tipo planilha, permitindo carregar, manipular e combinar dados.

A ideia aqui é apresentar um trabalho prático, de modo a assimilar conhecimento sobre o Pandas, sendo necessário apenas conhecimento básico das estruturas de dados do Python.

1 - Vamos criar um arquivo pandas.csv

2- importar o pacote pandas

#importando a biblioteca pandas

import pandas as pd

3- Carregar o Dataset sob o qual faremos a análise exploratória. No pandas, um dataset é conhecido como Dataframe.

#Carregando o dataset retirando linhas com problema e 

#utilizando como separador ponto e vírgula

df = pd.read_csv("Gapminder.csv",error_bad_lines=False, sep=';')

3- Visualizando os dados

#visualizando as 5 primeiras linhas

df.head(10) # ao passar um valor inteiro, você determina a quantidade

de linhas retornadas

5- Renomeando colunas

#Renomeando colunas

df.rename(columns=

{"country":"País","continent":"Continente","year":"Ano",

"lifeExp":"Expectativa de vida","pop":"População","gdpPercap":"PIB"})

6- Retornando o número de linhas e colunas

#retornando o número de linhas e colunas

df.shape

(3312, 6)

7- Retornando o nome das colunas

#Retornando o nome das colunas 

df.columns

Index(['country', 'continent', 'year', 'lifeExp', 'pop', 'gdpPercap'], dtype='object')

8- Retornando o nome das colunas

#retornando os tipos de dados das colunas

df.dtypes

country object continent object year int64 lifeExp float64 pop int64 gdpPercap float64 dtype: object

9- Retornando as últimas linhas

#retornando as últimas linhas

df.tail()

10- Retornando dados estatísticos

#Retornando dados estatísticos

df.describe()

11 - Pegando valores únicos na Coluna (Distinct)

# pegando valores únicos na coluna Continent

df['continent'].unique()

array(['Asia', 'Europe', 'Africa', 'Americas', nan, 'FSU', 'Oceania'], dtype=object)

12 - Filtrando dados (where)

#filtrando dados

Oceania = df.loc[df["continent"] == "Oceania"]

Oceania.head()

13 - Pegando a média de um campo

#média da expectativa de vida

df.groupby("year")["lifeExp"].mean()

14 - Agrupando dados (paises por continente count())

#agrupando dados por continente

df.groupby("continent")["country"].nunique()

15 - Soma e Média dos valores de um campo.

#Soma dos valores de um campo

df["gdpPercap"].sum()

#Média de valores de um campo

df["gdpPercap"].mean()

Neste exemplo, trabalhamos com arquivos .csv, mas é possível também trabalhar com planilhas Excel, dados provenientes de bancos de dados, entre outros.

Até a próxima!

O que é Engenharia de Dados

         Até pouco tempo atrás, nos deparávamos com um cenário onde os sistemas eram monolíticos, persistindo e consultando informações em bancos de dados relacionais. 

        Os mais avançadinhos criavam bancos de dados dimensionais visando performance e facilidade na busca das informações.

        Com o tempo, esse cenário foi mudando. Com o advento da Internet e Internet das coisas, hoje praticamente tudo gera informação, nem sempre de forma estruturada. O formato varia de acordo com o dispositivo que gera a informação. Exemplo: Câmeras que geram fotos, roteadores que geram logs em formato txt, uma peça de um automóvel que indica ao fabricante a necessidade de reparo, entre outros.

        Daí surge algumas questões: Como e onde guardar esses dados? Como produzir informação com os dados gerados se estes, nem sempre são estruturados e possui diversos formatos. Seria o DBA o responsável por resolver essas questões? É nesse cenário que surge a Engenharia de Dados, visando promover a governança das estruturas que irão receber dados desta natureza.

     Em um outro post falamos de ETL, aonde extraímos dados de bancos relacionais, realizamos transformações e geramos informações de acordo com a necessidade do negócio envolvido, com a finalidade de alimentar um banco de dados dimensional(Um DW estruturado). Mas, quando estamos lidando com dados semiestruturados ou ainda dados não estruturados, a ordem dos fatores se altera. Ocorre outro tipo de processo que é conhecido como ELT. É feita a carga desses dados não estruturados para o que chamamos de Datalake, onde é feita a transformação dos dados.

    Resumindo, um Engenheiro de Dados é o profissional que desenvolve, opera e mantem estruturas de dados complexas e heterogêneas, sendo responsável pela segurança, integridade, disponibilidade e confiabilidade desses dados.