Desenvolvendo uma pipeline de dados com Apache Airflow, API Sap, Google Storage e BigQuery para extração de funcionários Sap via API
Este artigo foi escrito pelo nosso Data Engineer Diego Oliveira. Foi postado diretamente na sua página do LinkedIn e, agora, estamos repostando aqui em nosso blog do Medium. Sigam a página do Diego para mais conteúdos como este!
O Apache Airflow é uma plataforma desenvolvida pela comunidade para criar, agendar e monitorar fluxos de trabalho, tudo feito programaticamente. Com ela, os pipelines do Airflow são definidos em Python, permitindo a geração dinâmica de pipeline, sem sair da sintaxe que já conhecemos 🐍.
Saber desenvolver pipeline de dados com o Apache Airflow é um requisito mais do que essencial caso você almeje uma carreira em engenharia de dados. Portanto, caso queira saber mais sobre essa poderosa ferramenta, continue lendo 😉.
Neste artigo irei ensinar como desenvolver uma pipeline de dados para extrair os dados de funcionários do sistema Sap. Depois da extração será aplicado algumas transformações bem simples nos dados para que, por fim, possam ser salvos em um dos maiores DataLakes do mercado o BigQuery.
Desenvolvimento da pipeline de dados
Antes de tudo, crie um arquivo chamado funcionarios_sap_flow_dag.py no subdiretório dags:
Importando Bibliotecas
Depois disso, adicione o seguinte código no arquivo:
Com isso já temos a estrutura inicial de nosso DAG. A biblioteca requests e pandas são necessárias para as tarefas de extração e transformação dos dados e o operador PythonOperator é importante para executar as funções desenvolvidas.
No DAG temos três tarefas principais, onde:
- Extract: irá fazer a extração dos funcionários da empresa cadastrado no Sap;
- Transform: irá selecionar apenas os funcionários ativos na empresa, e ordená-los de forma decrescente pelo campo name;
- Load: por fim, esta task irá salvar os dados transformados no Storage e BigQuery.
Como pôde perceber, tanto a task transform quanto a task load dependem dos dados extraídos ou transformados pela task anterior. Para que a transição dos dados seja feita entre as tasks, o Airflow usa um mecanismo interno chamado XComs, uma abreviação para Comunicação Cruzada.
Extração de dados
A extração de funcionários na API do Sap é feita através das seguintes etapas:
- Faz uma chamada GET no endpoint /api/v2/EmpJob, com o parâmetro format=json. O resultado será um json com o campo result, semelhante a este:
- Depois, acessamos o campo results e fazemos uma chamada GET para cada uma das URLS. O resultado será uma lista com todos os funcionários extraídos da API;
Parabéns! A task para extração da API está concluída 🎉.
Transformação dos dados
As etapas da task transform são as seguintes:
- Pega os dados extraídos pela task anterior, isto é, a task extract;
- Cria um DataFrame do Pandas com os dados e seleciona os funcionários ativos, descarrtando as demais;
- Ordena os dados do DataFrame pela coluna name, de forma decrescente;
Agora, mãos à obra, de novo! Adicione o seguinte código dentro da função transform dentro de nosso DAG.
Carregamento dos dados
Por fim, só o que nos resta é desenvolver a task load. As etapas desta task são:
- Pega os dados transformados pela task anterior;
- Salva os dados do DataFrame no storage da cloud GCP, no formato CSV;
- Persistir os dados do DataFrame no BigQuery
O código final de nosso DAG ficou assim:
Considerações finais
Como vimos, o Apache Airflow é uma poderosa ferramenta para a orquestração de tarefas de uma pipeline de dados. Não esqueça de conferir as outras inúmeras funcionalidades do Airflow 😉.
Neste artigo ensinei como desenvolver um DAG para extração de dados de funcionários do SAP via API. Foi aplicado algumas transformações bem simples nos dados antes de serem salvos no Google Storage e Big Query.
Se você gostou desse artigo, não esqueça de curtir e compartilhar nas redes sociais 💚.
Até a próxima!
Referências:
Documentação Airflow:
Documentação Success Factors:
Texto escrito por Diego Oliveira, Data Engineer. https://www.linkedin.com/pulse/desenvolvendo-uma-pipeline-de-dados-com-apache-airflow-diego-oliveira/?trackingId=tlpdJ70sYP28ozCex7eJfg%3D%3D
