ETL_open_data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introdução "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nesse tutorial vamos aprender o básico sobre as etapas de uma rotina de ETL(extract, transform and load). A ideia aqui é utilizar um dataset público real para demonstrar como implementar as diferentes etapas dessa rotina. Vamos explorar alguns conceitos básicos da Engenharia de dados, como implementar rotinas para extração de arquivos e como manipular dados tabulares com o Pandas. Lembrando que toda a implementação aqui é focada em *small data*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# O que é ETL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**ETL(Extract, transform and load)** é um processo que consiste em integrar dados de diferentes fontes buscando consolidá-los de uma maneira que facilite o processo de análise. O termo se popularizou com o surgimento das *data warehouses*, que nasceram da necessidade de centralizar diversas fontes de dados para permitir criar análises que ajudassem as empresas na tomada de decisão."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O processo consiste de três etapas:\n",
    "\n",
    "1. *Extract*: extrair dados de uma fonte de informações, seja banco de dados, API, arquivos, etc. Para serem processados posteriormente;\n",
    "2. *Transform*: nessa etapa os dados extraídos passam por uma transformação para atender os requisitos das aplicações cliente. A transormação envolve:\n",
    "    - Limpar e validar os dados para garantir qualidade;\n",
    "    - Transformar o formato dos dados para, por exemplo, facilitar usabilidade e busca;\n",
    "    - Combinar diferentes fontes para compor as informações necessárias;\n",
    "    - Aplicar regras de negócio.\n",
    "3. *Load*: carregar resultado das transformações no sistema de destino, como *date warehouses* ou *data lakes*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Data warehouse**: Somente dados estruturados, geralmente alimenta BI's;\n",
    "\n",
    "**Data lake**: Dados estruturados e não estruturados, utilizado por cientistas de dados."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](./img/etl_diagram.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A fonte de dados "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vamos trabalhar com a fonte de dados abertos da [Comissão de Valores Mobiliários](http://www.cvm.gov.br/) contendo a cotação diária dos fundos de investimentos negociados no mercado brasileiro. Essa fonte é atualizada diariamente com os dados de fechamento do dia anterior e as cotações são agrupadas por arquivos correspondentes a cada mês do ano.\n",
    "\n",
    "Esse é um cenário bem comum em fontes de dados abertas, uma série de arquivos no formato CSV agrupando informações de acordo com a data, então a solução que vamos implementar durante o tutorial é reaproveitável para outras fontes de dados.\n",
    "\n",
    "Primeiro é importante analisar a fonte de dados, entender como ela está estruturada, quais campos compõem o dataset e como nós podemos automatizar a coleta dos dados.\n",
    "\n",
    "A fonte de dados contendo a cotação diária dos fundos pode ser acessada através do portal de dados abertos pelo link:\n",
    "\n",
    "http://www.dados.gov.br/dataset/fi-doc-inf_diario"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Como automatizar o download "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Podemos observar que no site existe um link para todos os datasets dos últimos 12 meses. Primeiro precisamos analisar se existe um padrão nas urls de download dos arquivos, para isso vamos copiar alguns endereços e compará-los."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copiei três links e vamos compará-los a seguir:\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_201907.csv\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_201910.csv\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_202003.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como podemos observar existe um padrão claro nos links:\n",
    "\n",
    "`dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_YYYYMM.csv`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import requests\n",
    "from tqdm import tqdm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extrair "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O processo de extração consiste nesse caso em fazer o download de todos os arquivos da janela de tempo q nos interessa. As etapas pra nós atingirmos esse objetivo são:\n",
    "\n",
    "1. Automatizar a geração do nome dos arquivos: já que a única coisa que varia nos links é a data dos arquivos precisamos automatizar a geração dessas datas de acordo com a janela de tempo do nosso interesse;\n",
    "\n",
    "2. Requisitar o arquivo: precisamos enviar uma requisição para o portal de dados abertos do arquivo que queremos fazer o download;\n",
    "\n",
    "3. Salvar arquivo: o portal de dados abertos vai nos enviar o arquivo requisitado e precisamos salvá-lo no nosso computador."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Gera lista de datas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = pd.date_range(start='2019-01', end='2020-01', freq='M')[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "date_list = generate_date('2020-01', '2020-03')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Requisita e salva os arquivos "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. Chamar método `get`\n",
    "2. Checar se a requisição teve sucesso olhando o atributo `status_code`\n",
    "3. Salvar o conteúdo no computar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = requests.get('http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_202003.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_url = 'http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_{date}.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "files = [base_url.format(date=date) for date in date_list]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "def download_files(file_list):\n",
    "    save_list = []\n",
    "    for file in file_list:\n",
    "        response = requests.get(file)\n",
    "        \n",
    "        if response.status_code != 200:\n",
    "            print('Erro ao baixar o arquivo {}'.format(file))\n",
    "            continue\n",
    "        \n",
    "        file_save_name = 'data/{}'.format(file.split('/')[-1])\n",
    "        \n",
    "        with open(file_save_name, 'wb') as f:\n",
    "            f.write(response.content)\n",
    "        \n",
    "        save_list.append(file_save_name)\n",
    "        \n",
    "    return save_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "saved_files = download_files(files)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['data/inf_diario_fi_202001.csv', 'data/inf_diario_fi_202002.csv']"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "saved_files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Análise"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Aqui vou abrir um parênteses para uma análise rápido do nosso dateframe, é claro que só a parte de exploração dos dados vale um tutorial completo, então não vou explorar muito esses aspecto aqui, mas de qualquer forma é importante ter noção de alguns pontos básicos quando se trabalha com uma fonte de dados:\n",
    "\n",
    "- Qual o tipo de cada coluna\n",
    "- Quantos valores nulos que existem\n",
    "- Como estão formatados\n",
    "\n",
    "Essa etapa não faz parte do ETL, na verdade essa exploração inicial normalmente é feita antes de construir a *pipeline* para entender a fonte de dados utilizada."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Leitura de um arquivo\n",
    "df = pd.read_csv('data/inf_diario_fi_202001.csv', sep=';')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(369894, 8)"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Formato do dataframe (linhas, colunas)\n",
    "df.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 369894 entries, 0 to 369893\n",
      "Data columns (total 8 columns):\n",
      " #   Column         Non-Null Count   Dtype  \n",
      "---  ------         --------------   -----  \n",
      " 0   CNPJ_FUNDO     369894 non-null  object \n",
      " 1   DT_COMPTC      369894 non-null  object \n",
      " 2   VL_TOTAL       369894 non-null  float64\n",
      " 3   VL_QUOTA       369894 non-null  float64\n",
      " 4   VL_PATRIM_LIQ  369894 non-null  float64\n",
      " 5   CAPTC_DIA      369894 non-null  float64\n",
      " 6   RESG_DIA       369894 non-null  float64\n",
      " 7   NR_COTST       369894 non-null  int64  \n",
      "dtypes: float64(5), int64(1), object(2)\n",
      "memory usage: 22.6+ MB\n"
     ]
    }
   ],
   "source": [
    "# Quais são nossas colunas, seus tipos e se existem valores nulos\n",
    "df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>CNPJ_FUNDO</th>\n",
       "      <th>DT_COMPTC</th>\n",
       "      <th>VL_TOTAL</th>\n",
       "      <th>VL_QUOTA</th>\n",
       "      <th>VL_PATRIM_LIQ</th>\n",
       "      <th>CAPTC_DIA</th>\n",
       "      <th>RESG_DIA</th>\n",
       "      <th>NR_COTST</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-02</td>\n",
       "      <td>1132491.66</td>\n",
       "      <td>27.225023</td>\n",
       "      <td>1123583.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-03</td>\n",
       "      <td>1132685.12</td>\n",
       "      <td>27.224496</td>\n",
       "      <td>1123561.25</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-06</td>\n",
       "      <td>1132881.43</td>\n",
       "      <td>27.225564</td>\n",
       "      <td>1123605.31</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-07</td>\n",
       "      <td>1133076.85</td>\n",
       "      <td>27.226701</td>\n",
       "      <td>1123652.24</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-08</td>\n",
       "      <td>1132948.59</td>\n",
       "      <td>27.227816</td>\n",
       "      <td>1123698.26</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           CNPJ_FUNDO   DT_COMPTC    VL_TOTAL   VL_QUOTA  VL_PATRIM_LIQ  \\\n",
       "0  00.017.024/0001-53  2020-01-02  1132491.66  27.225023     1123583.00   \n",
       "1  00.017.024/0001-53  2020-01-03  1132685.12  27.224496     1123561.25   \n",
       "2  00.017.024/0001-53  2020-01-06  1132881.43  27.225564     1123605.31   \n",
       "3  00.017.024/0001-53  2020-01-07  1133076.85  27.226701     1123652.24   \n",
       "4  00.017.024/0001-53  2020-01-08  1132948.59  27.227816     1123698.26   \n",
       "\n",
       "   CAPTC_DIA  RESG_DIA  NR_COTST  \n",
       "0        0.0       0.0         1  \n",
       "1        0.0       0.0         1  \n",
       "2        0.0       0.0         1  \n",
       "3        0.0       0.0         1  \n",
       "4        0.0       0.0         1  "
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "16977"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(df.CNPJ_FUNDO.unique())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformar "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nessa etapa é necessário consolidar todas as informações extraídas, aplicar os tratamentos e regras de negócio que fazem sentido para a aplicação cliente, defini alguns objetivos para nos orientar durante essa etapa:\n",
    "\n",
    "1. Consolidar todos os arquivos em um *dataframe*;\n",
    "2. Transformar tipo da coluna de data(`DT_COMPTC`) para `datetime`;\n",
    "3. Manter somente fundos com mais de `1000` cotistas;\n",
    "4. Manter somente informações sobre: data, CNPJ do fundo e valor da cota;\n",
    "5. Mudar o formato do dataframe para:\n",
    "\n",
    "|            | 00.017.024/0001-53 | 97.929.213/0001-34 | 00.068.305/0001-35 | ... |\n",
    "|------------|--------------------|--------------------|--------------------|-----|\n",
    "| 2020-01-02 | 27.225023          | 27.112737          | 1.733476e+08       | ... |\n",
    "| 2020-01-03 | 27.224496          | 27.115661          | 6.611408e+07       | ... |\n",
    "| ...        | ...                | ...                | ...                | ... |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Consolida todos os arquivos "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_df_list = [pd.read_csv(file, sep=';') for file in saved_files]\n",
    "raw_df = pd.concat(raw_df_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(675754, 8)"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_df.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Converte coluna para tipo data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_df.DT_COMPTC = pd.to_datetime(raw_df.DT_COMPTC, format='%Y-%m-%d')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "Int64Index: 675754 entries, 0 to 305859\n",
      "Data columns (total 8 columns):\n",
      " #   Column         Non-Null Count   Dtype         \n",
      "---  ------         --------------   -----         \n",
      " 0   CNPJ_FUNDO     675754 non-null  object        \n",
      " 1   DT_COMPTC      675754 non-null  datetime64[ns]\n",
      " 2   VL_TOTAL       675754 non-null  float64       \n",
      " 3   VL_QUOTA       675754 non-null  float64       \n",
      " 4   VL_PATRIM_LIQ  675754 non-null  float64       \n",
      " 5   CAPTC_DIA      675754 non-null  float64       \n",
      " 6   RESG_DIA       675754 non-null  float64       \n",
      " 7   NR_COTST       675754 non-null  int64         \n",
      "dtypes: datetime64[ns](1), float64(5), int64(1), object(1)\n",
      "memory usage: 46.4+ MB\n"
     ]
    }
   ],
   "source": [
    "raw_df.info()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Filtra por quantidade de cotistas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_final_date = raw_df[raw_df.DT_COMPTC == raw_df.DT_COMPTC.max()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "qtd_shareholders = 1000\n",
    "valid_cnpjs = df_final_date.query('NR_COTST >= @qtd_shareholders').CNPJ_FUNDO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_filtered = raw_df.query('CNPJ_FUNDO in @valid_cnpjs')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(44403, 8)"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_filtered.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Altera formato do dataframe "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [],
   "source": [
    "result = pd.DataFrame(data={'DT_COMPTC': df_filtered.DT_COMPTC.sort_values().unique()})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "group_cnpjs = df_filtered.groupby('CNPJ_FUNDO')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1113/1113 [01:54<00:00,  9.74it/s]\n"
     ]
    }
   ],
   "source": [
    "for group in tqdm(group_cnpjs.groups):\n",
    "    df_cnpj = group_cnpjs.get_group(group)\n",
    "    result = pd.merge(result, df_cnpj[['DT_COMPTC', 'VL_QUOTA']], on='DT_COMPTC', how='left')\\\n",
    "               .rename(columns={'VL_QUOTA': group})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [],
   "source": [
    "result.set_index('DT_COMPTC', inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>00.068.305/0001-35</th>\n",
       "      <th>00.071.477/0001-68</th>\n",
       "      <th>00.180.995/0001-10</th>\n",
       "      <th>00.222.725/0001-24</th>\n",
       "      <th>00.222.816/0001-60</th>\n",
       "      <th>00.280.302/0001-60</th>\n",
       "      <th>00.306.278/0001-91</th>\n",
       "      <th>00.322.699/0001-06</th>\n",
       "      <th>00.360.293/0001-18</th>\n",
       "      <th>00.398.561/0001-90</th>\n",
       "      <th>...</th>\n",
       "      <th>67.976.449/0001-60</th>\n",
       "      <th>68.599.141/0001-06</th>\n",
       "      <th>68.623.479/0001-56</th>\n",
       "      <th>68.670.512/0001-07</th>\n",
       "      <th>68.971.183/0001-26</th>\n",
       "      <th>73.899.759/0001-21</th>\n",
       "      <th>88.002.696/0001-36</th>\n",
       "      <th>88.198.056/0001-43</th>\n",
       "      <th>97.519.703/0001-62</th>\n",
       "      <th>97.519.794/0001-36</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>DT_COMPTC</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2020-01-02</th>\n",
       "      <td>27.112737</td>\n",
       "      <td>9.979337</td>\n",
       "      <td>550.381807</td>\n",
       "      <td>3695.421072</td>\n",
       "      <td>7689.303576</td>\n",
       "      <td>4696.049724</td>\n",
       "      <td>18.811236</td>\n",
       "      <td>13.041650</td>\n",
       "      <td>22.378766</td>\n",
       "      <td>227.998183</td>\n",
       "      <td>...</td>\n",
       "      <td>12.634536</td>\n",
       "      <td>4.013766</td>\n",
       "      <td>5.693611</td>\n",
       "      <td>1.864455</td>\n",
       "      <td>1.100212</td>\n",
       "      <td>40.429076</td>\n",
       "      <td>0.711309</td>\n",
       "      <td>6107.57366</td>\n",
       "      <td>22.081756</td>\n",
       "      <td>21.002109</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2020-01-03</th>\n",
       "      <td>27.115661</td>\n",
       "      <td>9.979770</td>\n",
       "      <td>550.468081</td>\n",
       "      <td>3697.159049</td>\n",
       "      <td>7690.674711</td>\n",
       "      <td>4696.806615</td>\n",
       "      <td>18.813887</td>\n",
       "      <td>13.043727</td>\n",
       "      <td>22.501972</td>\n",
       "      <td>225.529173</td>\n",
       "      <td>...</td>\n",
       "      <td>12.541972</td>\n",
       "      <td>4.013966</td>\n",
       "      <td>5.694228</td>\n",
       "      <td>1.853287</td>\n",
       "      <td>1.100384</td>\n",
       "      <td>40.128902</td>\n",
       "      <td>0.709291</td>\n",
       "      <td>6119.45635</td>\n",
       "      <td>22.085839</td>\n",
       "      <td>21.005407</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2020-01-06</th>\n",
       "      <td>27.118216</td>\n",
       "      <td>9.980323</td>\n",
       "      <td>550.475403</td>\n",
       "      <td>3697.272064</td>\n",
       "      <td>7690.802997</td>\n",
       "      <td>4697.423331</td>\n",
       "      <td>18.816628</td>\n",
       "      <td>13.045941</td>\n",
       "      <td>22.544107</td>\n",
       "      <td>223.095944</td>\n",
       "      <td>...</td>\n",
       "      <td>12.452326</td>\n",
       "      <td>4.014164</td>\n",
       "      <td>5.694767</td>\n",
       "      <td>1.816770</td>\n",
       "      <td>1.100555</td>\n",
       "      <td>39.842954</td>\n",
       "      <td>0.703597</td>\n",
       "      <td>6091.10393</td>\n",
       "      <td>22.089976</td>\n",
       "      <td>21.008664</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2020-01-07</th>\n",
       "      <td>27.121996</td>\n",
       "      <td>9.980816</td>\n",
       "      <td>550.688536</td>\n",
       "      <td>3697.922105</td>\n",
       "      <td>7694.073773</td>\n",
       "      <td>4698.291493</td>\n",
       "      <td>18.819392</td>\n",
       "      <td>13.047728</td>\n",
       "      <td>22.593079</td>\n",
       "      <td>223.376967</td>\n",
       "      <td>...</td>\n",
       "      <td>12.428292</td>\n",
       "      <td>4.014379</td>\n",
       "      <td>5.695564</td>\n",
       "      <td>1.816762</td>\n",
       "      <td>1.100703</td>\n",
       "      <td>39.768262</td>\n",
       "      <td>0.702067</td>\n",
       "      <td>6132.30651</td>\n",
       "      <td>22.095126</td>\n",
       "      <td>21.011485</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2020-01-08</th>\n",
       "      <td>27.126159</td>\n",
       "      <td>9.981363</td>\n",
       "      <td>550.899190</td>\n",
       "      <td>3698.652972</td>\n",
       "      <td>7697.353404</td>\n",
       "      <td>4699.138945</td>\n",
       "      <td>18.822258</td>\n",
       "      <td>13.050008</td>\n",
       "      <td>22.501116</td>\n",
       "      <td>223.389476</td>\n",
       "      <td>...</td>\n",
       "      <td>12.382488</td>\n",
       "      <td>4.014584</td>\n",
       "      <td>5.696441</td>\n",
       "      <td>1.814723</td>\n",
       "      <td>1.100884</td>\n",
       "      <td>39.617461</td>\n",
       "      <td>0.698760</td>\n",
       "      <td>6130.89475</td>\n",
       "      <td>22.099806</td>\n",
       "      <td>21.014944</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 1113 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "            00.068.305/0001-35  00.071.477/0001-68  00.180.995/0001-10  \\\n",
       "DT_COMPTC                                                                \n",
       "2020-01-02           27.112737            9.979337          550.381807   \n",
       "2020-01-03           27.115661            9.979770          550.468081   \n",
       "2020-01-06           27.118216            9.980323          550.475403   \n",
       "2020-01-07           27.121996            9.980816          550.688536   \n",
       "2020-01-08           27.126159            9.981363          550.899190   \n",
       "\n",
       "            00.222.725/0001-24  00.222.816/0001-60  00.280.302/0001-60  \\\n",
       "DT_COMPTC                                                                \n",
       "2020-01-02         3695.421072         7689.303576         4696.049724   \n",
       "2020-01-03         3697.159049         7690.674711         4696.806615   \n",
       "2020-01-06         3697.272064         7690.802997         4697.423331   \n",
       "2020-01-07         3697.922105         7694.073773         4698.291493   \n",
       "2020-01-08         3698.652972         7697.353404         4699.138945   \n",
       "\n",
       "            00.306.278/0001-91  00.322.699/0001-06  00.360.293/0001-18  \\\n",
       "DT_COMPTC                                                                \n",
       "2020-01-02           18.811236           13.041650           22.378766   \n",
       "2020-01-03           18.813887           13.043727           22.501972   \n",
       "2020-01-06           18.816628           13.045941           22.544107   \n",
       "2020-01-07           18.819392           13.047728           22.593079   \n",
       "2020-01-08           18.822258           13.050008           22.501116   \n",
       "\n",
       "            00.398.561/0001-90  ...  67.976.449/0001-60  68.599.141/0001-06  \\\n",
       "DT_COMPTC                       ...                                           \n",
       "2020-01-02          227.998183  ...           12.634536            4.013766   \n",
       "2020-01-03          225.529173  ...           12.541972            4.013966   \n",
       "2020-01-06          223.095944  ...           12.452326            4.014164   \n",
       "2020-01-07          223.376967  ...           12.428292            4.014379   \n",
       "2020-01-08          223.389476  ...           12.382488            4.014584   \n",
       "\n",
       "            68.623.479/0001-56  68.670.512/0001-07  68.971.183/0001-26  \\\n",
       "DT_COMPTC                                                                \n",
       "2020-01-02            5.693611            1.864455            1.100212   \n",
       "2020-01-03            5.694228            1.853287            1.100384   \n",
       "2020-01-06            5.694767            1.816770            1.100555   \n",
       "2020-01-07            5.695564            1.816762            1.100703   \n",
       "2020-01-08            5.696441            1.814723            1.100884   \n",
       "\n",
       "            73.899.759/0001-21  88.002.696/0001-36  88.198.056/0001-43  \\\n",
       "DT_COMPTC                                                                \n",
       "2020-01-02           40.429076            0.711309          6107.57366   \n",
       "2020-01-03           40.128902            0.709291          6119.45635   \n",
       "2020-01-06           39.842954            0.703597          6091.10393   \n",
       "2020-01-07           39.768262            0.702067          6132.30651   \n",
       "2020-01-08           39.617461            0.698760          6130.89475   \n",
       "\n",
       "            97.519.703/0001-62  97.519.794/0001-36  \n",
       "DT_COMPTC                                           \n",
       "2020-01-02           22.081756           21.002109  \n",
       "2020-01-03           22.085839           21.005407  \n",
       "2020-01-06           22.089976           21.008664  \n",
       "2020-01-07           22.095126           21.011485  \n",
       "2020-01-08           22.099806           21.014944  \n",
       "\n",
       "[5 rows x 1113 columns]"
      ]
     },
     "execution_count": 96,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Carregar "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para etapa de carregamento vamos exportar o resultado em formato CSV e também vou demostrar como subir esse arquivo no [S3 da Amazon](https://aws.amazon.com/pt/s3/), que é um serviço de armazenamento de objetos na cloud, comumente usado como *data lake*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [],
   "source": [
    "result.to_csv('result.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Bônus**: Essa função utiliza do SDK do AWS para fazer o upload do arquivo resultante em um *bucket* no S3 da AWS, não vou entrar em detalhes sobre a configuração dessa ferramenta, mas se tiver interesse em saber mais a [documentação do boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html#installation) tem um passo a passo sobre como utilizar o SDK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "def upload_s3(file_path, remote_file_name, bucket):\n",
    "    s3 = boto3.resource('s3')\n",
    "    data = open(file_path, 'rb')\n",
    "    s3.Bucket(bucket).put_object(Key=remote_file_name, Body=data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [],
   "source": [
    "upload_s3('result.csv', 'pythonbrasil.csv', 'pythonbrasil')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusão"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Durante o tutorial passamos por todas as etapas do processo ETL, claro que a solução que implementamos aqui é simples e trabalha com um pequeno volume de dados, a partir do momento em que o volume de dados aumenta é preciso buscar ferramentas otimizadas, mas o processo no geral continua o mesmo. E para levar essa *pipeline* para produção o que falta? \n",
    "\n",
    "Em produção é importante utilizar ferramentas que paralelizem a execução das tarefas e que também sejam capazes de lidar com falhas durante o processo, no caso de *pipelines* que realizam processamento em lote(como a que nós implementamos), temos por exemplo ferramentas como:\n",
    "- [Apache Airflow](https://airflow.apache.org/)\n",
    "- [Luigi](https://github.com/spotify/luigi)\n",
    "- [Apache Beam](https://beam.apache.org/)\n",
    "- [Prefect](https://www.prefect.io/)\n",
    "\n",
    "Essa mesma *pipeline* que nós implementamos foi a primeira versão do que eu fiz para alimentar o [fundos.sharke.com.br](fundos.sharke.com.br), hoje está bem mais complexa e roda no [Apache Airflow](https://airflow.apache.org/):\n",
    "\n",
    "![](img/airflow_dag.png)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}