main_script_report_balance.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "fc75f339-455f-4e37-ab16-b97de3a2e9ee",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# Projeto Final | Big Data\n",
    "----\n",
    "**Desenvolvimento e Avaliação de uma Arquitetura Distribuída para um Relatório de Saldo Mensal da Conta**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0cf703be-db69-4e92-8c92-f1999ece780c",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Load Functions and variables"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5951ca83-e64d-4b6d-b695-60749c3bd093",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Funções para testar a qualidade dos dados (Great Expectations)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "849c8615-6d02-45c8-9d5a-803efb4c03d9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run ./modules/data-quality"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "434e61d3-7d2a-474f-b0a2-55001192a5e9",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Loading function *create_path* to create folders"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5817b3f2-d284-4263-9832-5fae420ef964",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run ./modules/utils"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b037d742-b91c-464e-a804-91fd3bcbbf70",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Strings json para criação do schema dos campos atráves da StructType (Bronze e Silver)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e41443af-c5f2-434c-8c89-76a9f2893acc",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run ./modules/json_strings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "08c6cd08-f929-4d25-b8d2-9ee489070c5b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run ./modules/json_strings_silver"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4b5a14d0-dc8d-49a1-9dcf-f4d12902cd39",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Load bibs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "50791887-028b-4924-8dcf-4e7126b15e40",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import zipfile\n",
    "import os\n",
    "import json\n",
    "\n",
    "from pyspark.sql.types import StringType, StructType, StructField, IntegerType, DecimalType, LongType, DataType, TimestampType, DoubleType\n",
    "from pyspark.sql.functions import expr, last_day, col, min, max, to_date, current_date, sum, lit, concat, lpad"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "39882bf1-236e-4ed9-b03e-f6b0352820ef",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Load Paths and create dirs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b2ce8678-6186-437a-91ea-d28ab3263e80",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "source_path = '/FileStore/project_report_balance/'\n",
    "\n",
    "landing_path = source_path + 'landing/'\n",
    "bronze_path = source_path + 'bronze/'\n",
    "silver_path = source_path + 'silver/'\n",
    "gold_path = source_path + 'gold/'\n",
    "\n",
    "path_list = [source_path, landing_path, bronze_path, silver_path, gold_path]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "89ea341d-c629-45e3-a17d-b8a4be0de5bf",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "dbutils.fs.rm(bronze_path, True)\n",
    "dbutils.fs.rm(silver_path, True)\n",
    "dbutils.fs.rm(gold_path, True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2da2a80e-c1f9-4f55-a5b7-4710dff3a973",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# create dirs\n",
    "for path in path_list:\n",
    "    create_path(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "42be0a94-181d-4fb7-90ab-6f92b6529e5a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Landing Zone"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "45850f82-9798-418f-b421-736da48478e0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "dir_path_list = ['accounts', 'city', 'country', 'customers', 'd_month', 'd_time', 'd_week', 'd_weekday', 'd_year', 'pix_movements', 'state', 'transfer_ins', 'transfer_outs']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f78a3aea-6e02-444e-985a-30630e6f6e7e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for dir_path in dir_path_list:\n",
    "    dbutils.fs.mkdirs(landing_path + dir_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7131533d-fa56-4b89-88da-42c1e9b6569f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Bronze layer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b00f2b51-5165-4f5b-b14a-a6fa069a398b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "## Carregamento dos dados da camada Bronze em parquet com schema definido \n",
    "\n",
    "for dir_name, json_str in zip(dir_path_list, json_str_list):\n",
    "    print(f'Criando dir {dir_name} na camada Bronze')\n",
    "\n",
    "    dir_path = landing_path + dir_name\n",
    "    csv_file_path = [arquivos.path for arquivos in dbutils.fs.ls(dir_path) if arquivos.name.endswith(('.csv', '.CSV'))]\n",
    "\n",
    "    print(f'Salvando dados em parquet no dir {dir_name} com schema definido')\n",
    "\n",
    "    # Loading json schema to create tables\n",
    "    schema_json = StructType.fromJson(json.loads(json_str))\n",
    "\n",
    "    df_csv = (spark.read.csv(csv_file_path[0], sep=',',header=True, schema = schema_json))\n",
    "\n",
    "    path_dir_bronze = bronze_path + dir_name\n",
    "    (df_csv\n",
    "        .write\n",
    "        .option(\"compression\",\"snappy\")\n",
    "        .mode(\"overwrite\")\n",
    "        .parquet(path_dir_bronze))\n",
    "    print('Dados salvos! \\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "35fe1bbf-48ec-46a3-a4d0-f326eb10fc26",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Silver layer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1715dc07-3c9b-4149-a10b-f14c8f791740",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Quality (Great Expectations)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "8eae9e2c-a6e8-4bf0-a2d9-3dc28a352683",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "###### Verificação de tipo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b87635db-9ae0-454e-850e-7adbd6e84bb2",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "tables_type = 'accounts'\n",
    "\n",
    "# Carregando dataframe\n",
    "table_path = bronze_path + tables_type\n",
    "df_table = spark.read.parquet(table_path) \n",
    "\n",
    "# Converter o DataFrame Spark em um DataFrame Great Expectations\n",
    "ge_df_table = SparkDFDataset(df_table)\n",
    "\n",
    "# Verificando os tipos das colunas\n",
    "print(f'Tabela analisada: {tables_type} - COLUNAS DATETIME')\n",
    "colunas_datetime = ['created_at']\n",
    "verifica_colunas_datetime(ge_df_table, colunas_datetime)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9872e37b-0fa6-4e16-92e6-b3a18eb176c6",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "###### Verificação de colunas categóricas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "485ddb37-151a-41d2-bd02-20ba7a3dd4d1",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "tables_cat = {\n",
    "    'accounts': {\n",
    "        'status': [\"active\", 'inactive']\n",
    "    }, \n",
    "    'pix_movements': {\n",
    "        'status': [\"failed\", 'completed'],\n",
    "        'in_or_out': [\"pix_in\", 'pix_out']\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "669efa7e-5470-4a2f-b4ad-82b6f70aed24",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for table, cols in tables_cat.items():\n",
    "    print(f'>>> Tabela analisada: {table}')\n",
    "    # Carregando dataframe\n",
    "    table_path = bronze_path + table\n",
    "    df_table = spark.read.parquet(table_path) \n",
    "\n",
    "    # Converter o DataFrame Spark em um DataFrame Great Expectations\n",
    "    ge_df_table = SparkDFDataset(df_table)\n",
    "\n",
    "    for col_analisada, valores_esperados in cols.items():\n",
    "        verificar_colunas_categoricas(ge_df_table, col_analisada, valores_esperados)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ff441afa-376f-4512-85bd-17c7ed2ed8ca",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "###### Verificação de colunas ID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "001c540e-cc53-4220-9b8a-bcc2e98f1278",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "tables_id = {\n",
    "    'accounts': ['account_id', 'customer_id']\n",
    "    , 'city': [\"state_id\",\"city_id\"]\n",
    "    , 'country': [\"country_id\"]\n",
    "    , 'customers': [\"customer_id\"] \n",
    "    , 'pix_movements': [\"id\",'account_id']\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c56c80ea-618a-4927-a92c-cecaa6be5238",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for table, id_cols_list in tables_id.items():\n",
    "    print(f'>>> Tabela analisada: {table}')\n",
    "    # Carregando dataframe\n",
    "    table_path = bronze_path + table\n",
    "    df_table = spark.read.parquet(table_path) \n",
    "\n",
    "    # Converter o DataFrame Spark em um DataFrame Great Expectations\n",
    "    ge_df_table = SparkDFDataset(df_table)\n",
    "\n",
    "    verificar_colunas_id(ge_df_table, id_cols_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ec94c71d-f620-4428-9911-7d7b7a59116f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "###### Verificação de colunas não vazias"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7e4deba0-7264-4dae-8722-f215f8985609",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "tables_non_empty  = {\n",
    "    'accounts': ['account_id', 'customer_id', 'account_branch', 'account_check_digit', 'account_number']\n",
    "    , 'city': ['city']\n",
    "    , 'country': ['country']\n",
    "    , 'customers': ['first_name', 'last_name', 'country_name', 'customer_city','cpf']\n",
    "    , 'pix_movements': ['account_id', 'id', 'pix_amount', 'pix_requested_at','pix_completed_at']\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "011ffa0e-fe13-471a-8592-337a19dadd64",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for table, cols_no_empty in tables_non_empty.items():\n",
    "    print(f'>>> Tabela analisada: {table}')\n",
    "    # Carregando dataframe\n",
    "    table_path = bronze_path + table\n",
    "    df_table = spark.read.parquet(table_path) \n",
    "\n",
    "    # Converter o DataFrame Spark em um DataFrame Great Expectations\n",
    "    ge_df_table = SparkDFDataset(df_table)\n",
    "\n",
    "    verificar_colunas_com_none(ge_df_table, cols_no_empty)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ff0caf2c-75f9-4abf-a095-f5512135cb3e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "###### Verificação de valores MIN e MAX"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "141fbf37-85ca-4f44-b7d7-b6d4d2dc5c25",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "tables_min_max  = {\n",
    "    'customers': [['cpf'],  0, 99999999999]\n",
    "    , 'pix_movements': [['pix_amount'],  0, 10000]\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2e321e44-aca1-43c9-8140-728c249ed5b9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for table, values in tables_min_max.items():\n",
    "    print(f'>>> Tabela analisada: {table}')\n",
    "    # Carregando dataframe\n",
    "    table_path = bronze_path + table\n",
    "    df_table = spark.read.parquet(table_path) \n",
    "\n",
    "    # Converter o DataFrame Spark em um DataFrame Great Expectations\n",
    "    ge_df_table = SparkDFDataset(df_table)\n",
    "\n",
    "    list_of_columns, minimo, maximo = values[0], values[1], values[2]\n",
    "    verificar_valores_min_max(ge_df_table, list_of_columns, minimo, maximo)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "975e9341-1aba-4107-afa3-eccf86e5aa48",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Creating Silver layer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "00dfd93c-19f8-4f47-ae29-4d959899eb46",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql\n",
    "CREATE SCHEMA IF NOT EXISTS silver LOCATION '/FileStore/project_report_balance/silver'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e7e46800-1a7c-437f-b581-cf916fad7829",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Creating Silver tables with StructType object from JSON file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e1a017d7-ee9b-4d7a-a7ca-b806af42990d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "lista_df_silver = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ba127b19-285c-4bb0-abbc-8592365d956b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "for dir_name, json_str in zip(dir_path_list, json_str_list):\n",
    "    print('Criando dataframe: ' + dir_name)\n",
    "\n",
    "    path_dir_bronze = bronze_path + dir_name\n",
    "\n",
    "    # Loading json schema to create tables\n",
    "    schema_json = StructType.fromJson(json.loads(json_str))\n",
    "\n",
    "    df_parquet = (spark.read.parquet(path_dir_bronze, sep = ',', header = True, schema = schema_json))\n",
    "    lista_df_silver[dir_name] = df_parquet\n",
    "\n",
    "    df_parquet.createOrReplaceTempView(dir_name)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7dbb9216-87b9-4864-aa0a-0f12a2c6927f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Creating Table silver.d_accounts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d625b07d-fdc6-4f9d-86aa-a91e169bd33b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "df_accounts = spark.read.table('accounts')\n",
    "df_accounts = df_accounts.withColumn('account', concat(col('account_branch'),col('account_check_digit'), col('account_number')))\n",
    "df_accounts = df_accounts.select(['account_id', 'status','account','created_at'])\n",
    "\n",
    "\n",
    "schema_json = StructType.fromJson(json.loads(json_str_list_silver[0]))\n",
    "\n",
    "path_dir_silver_d_account = silver_path + 'd_account'\n",
    "\n",
    "(df_accounts\n",
    "    .write\n",
    "    .saveAsTable('silver.d_accounts', compression = \"snappy\", mode = \"overwrite\", path = path_dir_silver_d_account, schema = schema_json)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ea93df87-d93c-4bc4-8dae-f3e7231021ae",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Creating Table silver.f_movements\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6f83f9f0-c73a-4d8c-a832-8c0dbc6ef8c4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "df_d_time = spark.read.table('d_time').select(['time_id', 'action_timestamp'])\n",
    "df_d_time = df_d_time.withColumn('ultimo_dia_mes', last_day('action_timestamp'))\n",
    "df_d_time = df_d_time.select(['time_id', 'ultimo_dia_mes'])    \n",
    "\n",
    "# pix\n",
    "df_pix = spark.read.table('pix_movements').select(['account_id','pix_amount','in_or_out','status','pix_requested_at', 'pix_completed_at'])\n",
    "df_pix = df_pix.join(df_d_time.alias('d_time').withColumnRenamed('ultimo_dia_mes','requested_at')\n",
    "            , col('d_time.time_id') == col('pix_movements.pix_requested_at')\n",
    "            , 'inner'\n",
    "    ).join(df_d_time.alias('d_time_completed').withColumnRenamed('ultimo_dia_mes','completed_at')\n",
    "            , col('d_time_completed.time_id') == col('pix_movements.pix_completed_at')\n",
    "            , 'left'\n",
    "    )\n",
    "df_pix = df_pix.withColumnRenamed('pix_amount','amount')\n",
    "df_pix = df_pix.select(['account_id','amount', 'in_or_out', 'status','requested_at','completed_at'])\n",
    "\n",
    "#transfer in\n",
    "df_transfer_ins = spark.read.table('transfer_ins').select(['account_id','amount','status','transaction_requested_at', 'transaction_completed_at'])\n",
    "df_transfer_ins = df_transfer_ins.withColumn('in_or_out', lit('transfer_in'))\n",
    "df_transfer_ins = df_transfer_ins.join(df_d_time.alias('d_time').withColumnRenamed('ultimo_dia_mes','requested_at')\n",
    "            , col('d_time.time_id') == col('transfer_ins.transaction_requested_at')\n",
    "            , 'inner'\n",
    "    ).join(df_d_time.alias('d_time_completed').withColumnRenamed('ultimo_dia_mes','completed_at')\n",
    "            , col('d_time_completed.time_id') == col('transfer_ins.transaction_completed_at')\n",
    "            , 'left'\n",
    "    )\n",
    "df_transfer_ins = df_transfer_ins.select(['account_id','amount', 'in_or_out', 'status','requested_at','completed_at'])\n",
    "\n",
    "# transfer outs\n",
    "df_transfer_outs = spark.read.table('transfer_outs').select(['account_id','amount','status','transaction_requested_at', 'transaction_completed_at'])\n",
    "df_transfer_outs = df_transfer_outs.withColumn('in_or_out', lit('transfer_out'))\n",
    "df_transfer_outs = df_transfer_outs.join(df_d_time.alias('d_time').withColumnRenamed('ultimo_dia_mes','requested_at')\n",
    "            , col('d_time.time_id') == col('transfer_outs.transaction_requested_at')\n",
    "            , 'inner'\n",
    "    ).join(df_d_time.alias('d_time_completed').withColumnRenamed('ultimo_dia_mes','completed_at')\n",
    "            , col('d_time_completed.time_id') == col('transfer_outs.transaction_completed_at')\n",
    "            , 'left'\n",
    "    )\n",
    "\n",
    "df_transfer_outs = df_transfer_outs.select(['account_id','amount', 'in_or_out', 'status','requested_at','completed_at'])\n",
    "\n",
    "df_f_movements = df_pix.unionAll(df_transfer_ins).unionAll(df_transfer_outs)\n",
    "\n",
    "schema_json_f_movements = StructType.fromJson(json.loads(json_str_list_silver[1]))\n",
    "path_dir_silver_f_movements= silver_path + 'f_movements'\n",
    "\n",
    "(df_f_movements\n",
    "    .write\n",
    "    .saveAsTable('silver.f_movements', compression = \"snappy\", mode = \"overwrite\", path = path_dir_silver_f_movements, schema = schema_json_f_movements)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0b36082d-e6cb-4028-9f84-a0040941fa03",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Creating Table silver.d_calendar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "501b81ec-c81a-4ddb-8558-548705cf2685",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "d_year_df = spark.read.table(\"d_year\").drop(col(\"year_id\")).dropDuplicates(['action_year'])\n",
    "d_month_df = spark.read.table(\"d_month\").drop(col(\"month_id\")).dropDuplicates(['action_month'])\n",
    "d_calendar_df = d_year_df.join(d_month_df)\n",
    "d_calendar_df = d_calendar_df.withColumnRenamed(\"action_year\",\"year\").withColumnRenamed(\"action_month\",\"month\")\n",
    "\n",
    "d_calendar_df = d_calendar_df.withColumn(\"ultimo_dia_mes\", last_day(to_date(concat(d_calendar_df[\"year\"], lit(\"-\"), lpad(d_calendar_df[\"month\"],2,\"0\"), lit(\"-\"), lit(\"01\")), \"yyyy-MM-dd\"))) \n",
    "\n",
    "d_calendar_df = d_calendar_df.alias(\"d_calendar\")\n",
    "schema_df_d_calendar = StructType([\n",
    "    StructField(\"year\", IntegerType(), False)\n",
    "    , StructField(\"month\", IntegerType(), False)\n",
    "    , StructField(\"ultimo_dia_mes\", DataType(), False)\n",
    "])\n",
    "\n",
    "d_calendar_path = silver_path + 'd_calendar'\n",
    "\n",
    "(d_calendar_df\n",
    "    .write\n",
    "    .saveAsTable('silver.d_calendar', compression=\"snappy\", mode=\"overwrite\", path=d_calendar_path, schema=schema_df_d_calendar)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "097ecf16-37ce-4545-9c0f-b33da343e50f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql\n",
    "SELECT * FROM silver.d_calendar LIMIT 5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "39f5fdda-9acc-4940-bf96-120d10bf6a1a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Gold layer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "be9e6e9f-87d7-4d94-bb55-854ce66d78b5",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Tabela/View 1: Lancamentos por mes "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7fa4e326-5f76-4594-b39f-bc817d497230",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Tabela/View 1 - Lancamentos por mes\n",
    "# Consulta ira criar uma tabela/view que vai apresentar as soma das entradas e saidas por mês e conta.\n",
    "sql_lancamentos_por_mes = \" \\\n",
    "   SELECT \\\n",
    "      account_id \\\n",
    "      , ultimo_dia_mes \\\n",
    "      , SUM(valor_saida) AS VALOR_SAIDA \\\n",
    "      , SUM(valor_entrada) AS VALOR_ENTRADA \\\n",
    "   FROM ( \\\n",
    "      SELECT account_id \\\n",
    "         , requested_at ultimo_dia_mes \\\n",
    "         , SUM(amount) valor_saida \\\n",
    "         , 0 valor_entrada \\\n",
    "      FROM silver.f_movements \\\n",
    "      WHERE status = 'completed' \\\n",
    "         and in_or_out in ('pix_out', 'transfer_out') \\\n",
    "      GROUP BY account_id \\\n",
    "         , requested_at \\\n",
    "      UNION ALL \\\n",
    "      SELECT account_id \\\n",
    "         , completed_at ultimo_dia_mes \\\n",
    "         , 0 valor_saida \\\n",
    "         , SUM(amount) valor_entrada \\\n",
    "      FROM silver.f_movements \\\n",
    "      WHERE status = 'completed' \\\n",
    "         and in_or_out in ('pix_in', 'transfer_in') \\\n",
    "         and completed_at is not NULL \\\n",
    "      GROUP BY account_id \\\n",
    "         , completed_at \\\n",
    "   ) lancamentos_por_mes \\\n",
    "   GROUP BY account_id, ultimo_dia_mes \\\n",
    "\"\n",
    " \n",
    "df_lancamentos_por_mes = spark.sql(sql_lancamentos_por_mes)\n",
    "df_lancamentos_por_mes.createOrReplaceTempView(\"lancamentos_por_mes\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "868e83d7-0a12-4c3e-8b83-49f782e20391",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Tabela/View 2: Total Por Mes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c696f5da-2a79-440d-bec9-20e6c7062499",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "   # Tabela/View 2 - Total Por Mes\n",
    "   # Essa consulta sql vai gerar uma tabela/view que vai listar a soma de todas as entrada e saida até o mes de analise.\n",
    "   sql_total_por_mes = \"\\\n",
    "      SELECT \\\n",
    "         d_calendar.ultimo_dia_mes As ultimo_dia_mes \\\n",
    "         , lancamentos_por_mes.account_id \\\n",
    "         , SUM(lancamentos_por_mes.VALOR_ENTRADA) TOTAL_ENTRADA \\\n",
    "         , SUM(lancamentos_por_mes.VALOR_SAIDA) TOTAL_SAIDA  \\\n",
    "          \\\n",
    "      FROM silver.d_calendar \\\n",
    "         LEFT JOIN lancamentos_por_mes  \\\n",
    "            ON lancamentos_por_mes.ultimo_dia_mes <= d_calendar.ultimo_dia_mes \\\n",
    " \\\n",
    "      GROUP BY d_calendar.ultimo_dia_mes \\\n",
    "         , lancamentos_por_mes.account_id \\\n",
    "   \"\n",
    "\n",
    "df_total_por_mes =  spark.sql(sql_total_por_mes)    \n",
    "df_total_por_mes.createOrReplaceTempView('total_por_mes') "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c66cb492-5073-4810-bf6d-4da7dfcfd8de",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Tabela/View 3: Acumulado por mes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9b1dd8ab-a80b-43a8-af1c-268042fbca2a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Tabela/View 3 - Acumulado por mes\n",
    "# Essa tabela apresenta o join das duas tabelas anteriores, e um calculo de saldo final por mes e conta.\n",
    "sql_acumulado_por_mes = \"\\\n",
    "    SELECT \\\n",
    "        d_accounts.account_id \\\n",
    "        , d_calendar.ultimo_dia_mes As ultimo_dia_mes \\\n",
    "        , lancamentos_por_mes.VALOR_ENTRADA TOTAL_ENTRADA \\\n",
    "        , lancamentos_por_mes.VALOR_SAIDA TOTAL_SAIDA \\\n",
    "        , COALESCE(total_por_mes.TOTAL_ENTRADA,0) - COALESCE(total_por_mes.TOTAL_SAIDA,0) AS SALDO_FINAL \\\n",
    "    FROM silver.d_accounts \\\n",
    "    INNER JOIN silver.d_calendar \\\n",
    "        ON d_calendar.ultimo_dia_mes >= TO_DATE(d_accounts.created_at, 'yyyy-MM-dd') \\\n",
    "    LEFT JOIN lancamentos_por_mes \\\n",
    "        ON lancamentos_por_mes.account_id = d_accounts.account_id \\\n",
    "        AND lancamentos_por_mes.ultimo_dia_mes = d_calendar.ultimo_dia_mes \\\n",
    "    LEFT JOIN total_por_mes \\\n",
    "        ON total_por_mes.account_id = d_accounts.account_id \\\n",
    "        AND total_por_mes.ultimo_dia_mes = d_calendar.ultimo_dia_mes \\\n",
    "\"\n",
    "df_acumulado_por_mes = spark.sql(sql_acumulado_por_mes)\n",
    "df_acumulado_por_mes.createOrReplaceTempView(\"acumulado_por_mes\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d163093a-8344-42e1-8ba5-abf164b63dcc",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Tabela/View Final: Saldo por mensal"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f73704ad-3270-4d37-96b2-8bb28343fef8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Tabela Final - Saldo Mensal - agg_saldo_mensal\n",
    "# Esse select vai aprensentar o saldo mensal final de cada conta e as entradas e saida de cada mês.\n",
    "sql_saldo_mensal = \"\\\n",
    "   SELECT \\\n",
    "       d_accounts.account_id \\\n",
    "       , d_accounts.account \\\n",
    "       , acumulado_por_mes.ultimo_dia_mes \\\n",
    "       , FORMAT_NUMBER(coalesce(acumulado_por_mes.total_entrada,0), 2) total_entrada \\\n",
    "       , FORMAT_NUMBER(coalesce(acumulado_por_mes.total_saida,0), 2) total_saida \\\n",
    "       , FORMAT_NUMBER(coalesce(acumulado_por_mes.saldo_final,0), 2) saldo_final \\\n",
    "   FROM silver.d_accounts \\\n",
    "      LEFT JOIN acumulado_por_mes \\\n",
    "         ON acumulado_por_mes.account_id = d_accounts.account_id \\\n",
    "   \"\n",
    "df_agg_saldo_mensal = spark.sql(sql_saldo_mensal)\n",
    "df_agg_saldo_mensal = df_agg_saldo_mensal.alias('saldo_mensal')\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "bf8643c8-0768-479e-8c29-d34004774af7",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Salvando o dataframe em tabela"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "446a6617-f77d-47d5-97fd-a92113d11a56",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql\n",
    "CREATE SCHEMA IF NOT EXISTS gold LOCATION '/FileStore/project_report_balance/gold'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2b9d734a-b579-42fa-a99f-a7803ed36732",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "schema_gold_agg_saldo_mensal = StructType([\n",
    "    StructField(\"account_id\", LongType(), False),\n",
    "    StructField(\"account\", StringType(), False),    \n",
    "    StructField(\"ultimo_dia_mes\", DataType(), False),\n",
    "    StructField(\"valor_entrada\", DoubleType(), False),\n",
    "    StructField(\"valor_saida\", DoubleType(), False),\n",
    "    StructField(\"saldo_final\", DoubleType(), False)\n",
    "])\n",
    "\n",
    "\n",
    "agg_saldo_mensal_path = gold_path + 'agg_saldo_mensal'\n",
    "(df_agg_saldo_mensal\n",
    "    .write\n",
    "    .saveAsTable('gold.agg_saldo_mensal', compression=\"snappy\", mode=\"overwrite\", path=agg_saldo_mensal_path, schema=schema_gold_agg_saldo_mensal)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c1d494b1-6267-4ebb-a357-f8c886f268b0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql \n",
    "SELECT * FROM gold.agg_saldo_mensal LIMIT 100"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f7977432-80b1-418f-ab13-bfa2a62b2d0e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Exemplos de Accounts e Saldo Mensal"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a999f49d-a8a7-4805-a4ce-c9a0e8dc03b9",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Account com algum registro de entrada igual a zero"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "df644858-4f31-4c96-8d47-50d2e3e3fba8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql \n",
    "SELECT * FROM gold.agg_saldo_mensal where account_id = 1910868644230470 order by account_id, ultimo_dia_mes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3ee87594-d5cb-401e-985e-e24c95585882",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Account com algum mes com entrada e saida zerado, ou seja, sem movimentacao no mês"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ec32adbd-551b-498f-b0f0-b4c2bf65476a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql \n",
    "SELECT * FROM gold.agg_saldo_mensal where account_id = 100642855136823056 order by account_id, ultimo_dia_mes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4378188e-559c-4fc2-8981-18c930c80180",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Account com algum mês com saldo negativo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "implicitDf": true,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "96e568df-8321-411f-a546-d1638f5a14c0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%sql \n",
    "SELECT * FROM gold.agg_saldo_mensal where account_id = 1972174676324008704 order by account_id, ultimo_dia_mes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "64afb672-2074-419a-9ca1-83daa5c1e6ff",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "environmentMetadata": null,
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 2977495270882142,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 4
   },
   "notebookName": "main_script_report_balance",
   "widgets": {}
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.10.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}