spark-notebook/SilverToGoldDataProcess.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 2,
  "cells": [
    {
      "cell_type": "code",
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 8,
              "statement_ids": [
                8
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:17.4516709Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:17.6016128Z",
              "execution_finish_time": "2024-08-30T15:50:18.1353298Z",
              "parent_msg_id": "01dcd36c-9e8e-4c65-9391-5298725d2abf"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 8, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "from pyspark.sql.functions import *\n",
        "\n",
        "# Define paths\n",
        "silver_base_path = \"abfss://fintech@airbnbdatagds.dfs.core.windows.net/silver/fintech/\"\n",
        "output_base_path = \"abfss://fintech@airbnbdatagds.dfs.core.windows.net/gold/fintech/\"\n",
        "\n",
        "# Load data from the silver layer\n",
        "accounts_df = spark.read.format(\"delta\").load(f\"{silver_base_path}Accounts/\")\n",
        "customers_df = spark.read.format(\"delta\").load(f\"{silver_base_path}Customers/\")\n",
        "loans_df = spark.read.format(\"delta\").load(f\"{silver_base_path}Loans/\")\n",
        "payments_df = spark.read.format(\"delta\").load(f\"{silver_base_path}Payments/\")\n",
        "transactions_df = spark.read.format(\"delta\").load(f\"{silver_base_path}Transactions/\")\n",
        "\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 9,
              "statement_ids": [
                9
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:17.4981663Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:18.2777553Z",
              "execution_finish_time": "2024-08-30T15:50:23.5932296Z",
              "parent_msg_id": "aae8a35f-9e1b-474d-a27a-cfbf5e4965ef"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 9, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "dim_customers_df = customers_df.select(\n",
        "    col(\"CustomerID\").alias(\"customer_id\"),\n",
        "    col(\"FirstName\").alias(\"first_name\"),\n",
        "    col(\"LastName\").alias(\"last_name\"),\n",
        "    col(\"Email\").alias(\"email\"),\n",
        "    col(\"PhoneNumber\").alias(\"phone_number\"),\n",
        "    col(\"Address\").alias(\"address\"),\n",
        "    col(\"City\").alias(\"city\"),\n",
        "    col(\"State\").alias(\"state\"),\n",
        "    col(\"Country\").alias(\"country\"),\n",
        "    col(\"ZipCode\").alias(\"zip_code\"),\n",
        "    col(\"SignupDate\").alias(\"signup_date\")\n",
        ")\n",
        "\n",
        "dim_customers_df.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}dim_customers/\")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 10,
              "statement_ids": [
                10
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:17.547399Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:23.7276848Z",
              "execution_finish_time": "2024-08-30T15:50:29.1370214Z",
              "parent_msg_id": "99d6b762-a356-49ab-b89f-328a7fbe7318"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 10, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "dim_accounts_df = accounts_df.select(\n",
        "    col(\"AccountID\").alias(\"account_id\"),\n",
        "    col(\"AccountType\").alias(\"account_type\"),\n",
        "    col(\"Balance\").alias(\"balance\"),\n",
        "    col(\"OpenDate\").alias(\"open_date\"),\n",
        "    col(\"AccountAgeYears\").alias(\"account_age_years\")\n",
        ")\n",
        "\n",
        "dim_accounts_df.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}dim_accounts/\")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 11,
              "statement_ids": [
                11
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:17.5996913Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:29.299071Z",
              "execution_finish_time": "2024-08-30T15:50:33.2622088Z",
              "parent_msg_id": "7e805b00-f63e-4cd2-803b-849ef2f28f0c"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 11, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "dim_loans_df = loans_df.select(\n",
        "    col(\"LoanID\").alias(\"loan_id\"),\n",
        "    col(\"LoanType\").alias(\"loan_type\"),\n",
        "    col(\"LoanAmount\").alias(\"loan_amount\"),\n",
        "    col(\"InterestRate\").alias(\"interest_rate\"),\n",
        "    col(\"LoanStartDate\").alias(\"loan_start_date\"),\n",
        "    col(\"LoanEndDate\").alias(\"loan_end_date\"),\n",
        "    col(\"TotalInterest\").alias(\"total_interest\"),\n",
        "    col(\"LoanDurationYears\").alias(\"loan_duration_years\")\n",
        ")\n",
        "\n",
        "dim_loans_df.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}dim_loans/\")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 12,
              "statement_ids": [
                12
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:17.7759174Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:33.4124288Z",
              "execution_finish_time": "2024-08-30T15:50:42.097262Z",
              "parent_msg_id": "0ed8f51f-b136-4c4d-991d-4ef939b8ccfc"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 12, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "fact_payments_df = payments_df \\\n",
        "    .join(loans_df.select(\"LoanID\", \"CustomerID\"), \"LoanID\") \\\n",
        "    .select(\n",
        "        col(\"PaymentID\").alias(\"payment_id\"),\n",
        "        col(\"LoanID\").alias(\"loan_id\"),\n",
        "        col(\"CustomerID\").alias(\"customer_id\"),\n",
        "        col(\"PaymentDate\").alias(\"payment_date\"),\n",
        "        col(\"PaymentAmount\").alias(\"payment_amount\"),\n",
        "        col(\"PaymentMethod\").alias(\"payment_method\")\n",
        "    )\n",
        "\n",
        "fact_payments_df.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}fact_payments/\")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "gdssparkpool",
              "statement_id": 13,
              "statement_ids": [
                13
              ],
              "state": "finished",
              "livy_statement_state": "available",
              "spark_jobs": null,
              "session_id": "6",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T15:50:18.2160766Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T15:50:42.2253683Z",
              "execution_finish_time": "2024-08-30T15:50:47.5380183Z",
              "parent_msg_id": "798b26a4-980e-4667-8526-25279d046616"
            },
            "text/plain": "StatementMeta(gdssparkpool, 6, 13, Finished, Available, Finished)"
          },
          "metadata": {}
        }
      ],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "fact_transactions_df = transactions_df \\\n",
        "    .join(accounts_df.select(\"AccountID\", \"CustomerID\"), \"AccountID\") \\\n",
        "    .select(\n",
        "        col(\"TransactionID\").alias(\"transaction_id\"),\n",
        "        col(\"AccountID\").alias(\"account_id\"),\n",
        "        col(\"CustomerID\").alias(\"customer_id\"),\n",
        "        col(\"TransactionDate\").alias(\"transaction_date\"),\n",
        "        col(\"Amount\").alias(\"amount\"),\n",
        "        col(\"TransactionType\").alias(\"transaction_type\"),\n",
        "        col(\"Description\").alias(\"description\")\n",
        "    )\n",
        "\n",
        "fact_transactions_df.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}fact_transactions/\")\n",
        ""
      ]
    }
  ],
  "metadata": {
    "description": null,
    "save_output": true,
    "kernelspec": {
      "name": "synapse_pyspark",
      "display_name": "python"
    },
    "language_info": {
      "name": "python"
    },
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  }
}