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

Análise de Dados

 Ao assistir Netflix, nos são apresentados os filmes que estão em "alta" e ainda, nos deparamos com recomendações baseadas nos filmes que costumamos assistir.

O mesmo acontece ao acessarmos o Mercado Livre e o Youtube. Nos são sugeridos videos e produtos similares aos que foram pesquisados em visitas passadas.

Fica parecendo que estes sites são videntes que advinham o que se passa pelas nossas cabeças, ou ainda, que seus desenvolvedores fazem magia negra ou até mesmo tem pacto com o Pai do Fogo, não mesmo?

Uma das maiores redes de varejo dos Estados Unidos descobriu que a venda de fraldas descartáveis estava associada à de cerveja. Em geral, os compradores eram homens, que saíam à noite para comprar fraldas e aproveitavam para levar algumas latinhas para casa. Os produtos foram postos lado a lado. Resultado: a venda de fraldas e cervejas disparou. A pergunta que não quer calar é: Como essa empresa chegou a essa conclusão com informações que aparentemente são desconexas? 

Magia negra? Simpatia? Pacto com o Demo? Não! Trata-se de análise de dados.

Análise de dados é o processo de aplicação de técnicas estatísticas e lógicas para avaliar informações obtidas a partir de determinados processos. O principal objetivo da prática é extrair informações úteis a partir dos dados. A partir destas informações, é possível tomar decisões mais assertivas e orientadas para resultados. 

Não é de hoje que análise de dados é uma atividade bastante desempenhada por pessoas e empresas. Quem nunca utilizou o Excel para analisar e contabilizar dados, gerar insights e apoiar nossas decisões? 

Mas como analisar dados? Acredito que a melhor forma de explicar isso é através de exemplos.

Vamos supor que um plano de saúde encomende uma análise de perfil de seus associados. A Operadora utiliza um sistema de gestão que possui o cadastro de todos os associados contendo a data de nascimento, sexo, tipo de plano, procedimentos médicos efetuados, etc.

Uma forma de analisar os dados é se basear em fatos, utilizando procedimentos e técnicas de estatística descritiva.

Algumas medidas que são normalmente usadas para descrever um conjunto de dados são medidas de tendência central. Medidas de tendência central incluem média, mediana e moda. Medidas de variabilidade incluem desvio padrão, variância, o valor máximo e mínimo, obliquidade e curtose.

Depois de promover a agregação em conjunto de dados podemos saber a média da idades dos associados do plano de saúde. O percentual de associados do sexo Masculino ou feminino, quais procedimentos médicos são mais realizados e por aí vai.

Damos a ser tipo de análise o nome de ANÁLISE DESCRITIVA.

Após a produção dessas informações podemos prescrever ações para chegarmos a novos resultados. Vamos supor que a incidência de consultas com clínico geral esteja muito alta. Poderíamos propor a aplicação de programas do tipo medicina em família de modo reduzir essa incidência. Esse tipo de análise é conhecimento com ANÁLISE PRESCRITIVA.

Verifique que os exemplos apresentados são basicamente quantitativos. A estatística serve não só como instrumento quantificador, como também para montar modelos e fórmulas que possam antecipar resultados (PROBABILIDADE). É aí que entra a ANÁLISE PREDITIVA aonde são desenvolvidos modelos de aprendizado de máquina capazes de prever que a abertura de clínicas próprias por parte da Operadora de plano de Saúde tende a reduzir o custo da Operadora com consultas com clínico geral.

Resumindo analisar dados é verificar as possibilidades que esses dados trazem, tirando conclusões sobre eles e antecipar tendências.

Mas isso aí é só a pontinha do Iceberg. Bora pro próximo Post!