spark-notebook/BronzeToSilverDataProcess.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 2,
  "cells": [
    {
      "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": "2",
              "normalized_state": "finished",
              "queued_time": "2024-08-30T13:10:00.2515056Z",
              "session_start_time": null,
              "execution_start_time": "2024-08-30T13:10:00.4245173Z",
              "execution_finish_time": "2024-08-30T13:10:26.7761104Z",
              "parent_msg_id": "345cec6d-a90a-463c-84f1-8cf17770180c"
            },
            "text/plain": "StatementMeta(gdssparkpool, 2, 9, Finished, Available, Finished)"
          },
          "metadata": {}
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": "Bronze To Silver Completed !!\n"
        }
      ],
      "metadata": {},
      "source": [
        "from pyspark.sql.functions import *\n",
        "\n",
        "# Define paths\n",
        "base_path = \"abfss://fintech@airbnbdatagds.dfs.core.windows.net/bronze/fintech/\"\n",
        "output_base_path = \"abfss://fintech@airbnbdatagds.dfs.core.windows.net/silver/fintech/\"\n",
        "\n",
        "spark.conf.set(\"spark.databricks.delta.schema.autoMerge.enabled\", \"true\")\n",
        "\n",
        "# Transformation for Accounts dataset\n",
        "def transform_accounts():\n",
        "    df = spark.read.parquet(f\"{base_path}Accounts/Accounts.parquet\")\n",
        "    # Example transformation: Calculate account age in years\n",
        "    df_transformed = df.withColumn(\"AccountAgeYears\", \n",
        "                                   round(datediff(current_date(), col(\"OpenDate\")) / 365.25, 2))\n",
        "    df_transformed.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}Accounts/\")\n",
        "\n",
        "# Transformation for Customers dataset\n",
        "def transform_customers():\n",
        "    df = spark.read.parquet(f\"{base_path}Customers/Customers.parquet\")\n",
        "    # Example transformation: Create a full name column and mask the email address\n",
        "    df_transformed = df.withColumn(\"FullName\", concat_ws(\" \", col(\"FirstName\"), col(\"LastName\"))) \\\n",
        "                       .withColumn(\"MaskedEmail\", \n",
        "                                   concat(lit(\"***@\"), substring_index(col(\"Email\"), \"@\", -1)))\n",
        "    df_transformed.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}Customers/\")\n",
        "\n",
        "# Transformation for Loans dataset with explicit casting\n",
        "def transform_loans():\n",
        "    df = spark.read.parquet(f\"{base_path}Loans/Loans.parquet\")\n",
        "    # Example transformation: Calculate total interest with explicit casting to match the Delta table\n",
        "    df_transformed = df.withColumn(\"TotalInterest\", \n",
        "                                   (col(\"LoanAmount\") * col(\"InterestRate\") / 100).cast(\"decimal(28,8)\")) \\\n",
        "                       .withColumn(\"LoanDurationYears\", \n",
        "                                   round(datediff(col(\"LoanEndDate\"), col(\"LoanStartDate\")) / 365.25, 2))\n",
        "    df_transformed.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}Loans/\")\n",
        "\n",
        "# Transformation for Payments dataset\n",
        "def transform_payments():\n",
        "    df = spark.read.parquet(f\"{base_path}Payments/Payments.parquet\")\n",
        "    # Example transformation: Calculate days since last payment\n",
        "    df_transformed = df.withColumn(\"DaysSinceLastPayment\", \n",
        "                                   datediff(current_date(), col(\"PaymentDate\")))\n",
        "    df_transformed.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}Payments/\")\n",
        "\n",
        "# Transformation for Transactions dataset\n",
        "def transform_transactions():\n",
        "    df = spark.read.parquet(f\"{base_path}Transactions/Transactions.parquet\")\n",
        "    # Example transformation: Categorize transaction types\n",
        "    df_transformed = df.withColumn(\"TransactionCategory\", \n",
        "                                   when(col(\"TransactionType\") == \"Deposit\", \"Income\")\n",
        "                                   .when(col(\"TransactionType\") == \"Withdrawal\", \"Expense\")\n",
        "                                   .otherwise(\"Other\"))\n",
        "    df_transformed.write.format(\"delta\").mode(\"overwrite\").save(f\"{output_base_path}Transactions/\")\n",
        "\n",
        "# Process each table\n",
        "transform_accounts()\n",
        "transform_customers()\n",
        "transform_loans()\n",
        "transform_payments()\n",
        "transform_transactions()\n",
        "\n",
        "print(\"Bronze To Silver Completed !!\")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        ""
      ]
    }
  ],
  "metadata": {
    "description": null,
    "save_output": true,
    "kernelspec": {
      "name": "synapse_pyspark",
      "display_name": "python"
    },
    "language_info": {
      "name": "python"
    },
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  }
}