1-transform-bronze-to-silver.ipynb

{
    "metadata": {
        "kernelspec": {
            "name": "SQL",
            "display_name": "SQL",
            "language": "sql"
        },
        "language_info": {
            "name": "sql",
            "version": ""
        },
        "application/vnd.databricks.v1+notebook": {
            "dashboards": [],
            "environmentMetadata": null,
            "language": "python",
            "notebookMetadata": {
                "pythonIndentUnit": 4
            },
            "notebookName": "1-transform-bronze-to-silver",
            "widgets": {}
        }
    },
    "nbformat_minor": 2,
    "nbformat": 4,
    "cells": [
        {
            "cell_type": "markdown",
            "source": [
                "\n",
                "### Transform and Export Processed Data to ADLS \n",
                "\n",
                "Extract data from imported database assuming default Access NI schema.\n",
                "Data is converted to flat denormalized CSV files and exported to ADLS **silver** container.\n",
                "\n",
                "One CSV file will be created for each Sequence File name that was called.\n"
            ],
            "metadata": {
                "application/vnd.databricks.v1+cell": {
                    "cellMetadata": {
                        "byteLimit": 2048000,
                        "rowLimit": 10000
                    },
                    "inputWidgets": {},
                    "nuid": "950763e6-b778-4d57-a3e2-249b4875d6a5",
                    "showTitle": false,
                    "title": ""
                },
                "azdata_cell_guid": "18f0da48-8106-4ee1-a3ad-990243df99a5"
            }
        },
        {
            "cell_type": "code",
            "source": [
                "# Declare script variables\n",
                "debug_mode = True\n",
                "database_name = \"dbo\"\n",
                "\n",
                "bronze_source = \"REDACTED\"\n",
                "bronze_target = bronze_source + database_name\n",
                "bronze_mnt = \"/mnt/bronze/\" + database_name\n",
                "\n",
                "silver_source = \"REDACTED\"\n",
                "silver_target = silver_source + database_name\n",
                "silver_mnt = \"/mnt/silver/\" + database_name\n",
                "\n",
                "gold_source = \"REDACTED\"\n",
                "gold_target = gold_source + database_name\n",
                "gold_mnt = \"/mnt/gold/\" + database_name"
            ],
            "metadata": {
                "application/vnd.databricks.v1+cell": {
                    "cellMetadata": {
                        "byteLimit": 2048000,
                        "rowLimit": 10000
                    },
                    "inputWidgets": {},
                    "nuid": "c8357662-8a99-4e9c-b4ba-8e71e0e20591",
                    "showTitle": false,
                    "title": ""
                },
                "azdata_cell_guid": "4389c7f5-a90d-4d17-a99c-5e6102a8eaed",
                "language": "sql"
            },
            "outputs": [],
            "execution_count": null
        },
        {
            "cell_type": "code",
            "source": [
                "# Get all desired test runs\n",
                "# Cast Start_Date_Time as Timestamp, Test_Socket_Index AND Execution_Time as Integer, and Data as 3-Digit-Double\n",
                "\n",
                "sql_string_crosstab = f\"\"\"\n",
                "    SELECT  STEP_SEQCALL.SEQUENCE_FILE_PATH,\n",
                "            UUT_RESULT.ID,\n",
                "            CAST(UUT_RESULT.START_DATE_TIME AS TIMESTAMP), \n",
                "            UUT_RESULT.STATION_ID, \n",
                "            UUT_RESULT.UUT_SERIAL_NUMBER,\n",
                "            CAST(UUT_RESULT.TEST_SOCKET_INDEX AS INT), \n",
                "            UUT_RESULT.UUT_STATUS, \n",
                "            CAST(UUT_RESULT.EXECUTION_TIME AS INT),\n",
                "            STEP_RESULT.STEP_NAME, \n",
                "            ROUND(CAST(PROP_RESULT.DATA AS DOUBLE), 3) AS DATA\n",
                "    FROM UUT_RESULT LEFT JOIN STEP_RESULT \n",
                "        ON UUT_RESULT.ID = STEP_RESULT.UUT_RESULT\n",
                "            LEFT JOIN PROP_RESULT \n",
                "                ON STEP_RESULT.ID = PROP_RESULT.STEP_RESULT\n",
                "                    LEFT JOIN STEP_SEQCALL \n",
                "                        ON STEP_SEQCALL.STEP_RESULT = STEP_RESULT.STEP_PARENT\n",
                "    WHERE UUT_STATUS <> 'Terminated' \n",
                "        and UUT_STATUS <> 'Error' \n",
                "        and STATION_ID is not NULL\n",
                "        and STEP_RESULT.STEP_TYPE = 'NumericLimitTest'   \n",
                "        and PROP_RESULT.TYPE_NAME = 'NumericLimitTest'\n",
                "    ORDER BY START_DATE_TIME, TEST_SOCKET_INDEX, STEP_RESULT.ORDER_NUMBER ASC\n",
                "    \"\"\"\n",
                "\n",
                "test_results = spark.sql(sql_string_crosstab)\n",
                "\n",
                "# Transform raw Column of filepaths to clean Column of Sequence_Name's\n",
                "rdd=test_results.rdd.map(lambda x: \n",
                "    (x['SEQUENCE_FILE_PATH'].split('\\\\')[-1].split('.')[0]\n",
                "    ,x['ID']\n",
                "    ,x['START_DATE_TIME']\n",
                "    ,x['STATION_ID']\n",
                "    ,x['UUT_SERIAL_NUMBER']\n",
                "    ,x['TEST_SOCKET_INDEX']\n",
                "    ,x['UUT_STATUS']\n",
                "    ,x['EXECUTION_TIME']\n",
                "    ,x['STEP_NAME']\n",
                "    ,x['DATA']\n",
                "    )\n",
                ")  \n",
                "run_steps=rdd.toDF(\n",
                "    [\"Sequence_Name\"\n",
                "    ,\"Run_ID\"\n",
                "    ,\"Test_Start\"\n",
                "    ,\"Station_ID\"\n",
                "    ,\"Serial_Number\"\n",
                "    ,\"Test_Socket\"\n",
                "    ,\"Test_Status\"\n",
                "    ,\"Test_Time_Sec\"\n",
                "    ,\"Step_Name\"\n",
                "    ,\"Data\"\n",
                "    ]\n",
                ")\n",
                "\n",
                "display(run_steps)"
            ],
            "metadata": {
                "application/vnd.databricks.v1+cell": {
                    "cellMetadata": {
                        "byteLimit": 2048000,
                        "rowLimit": 10000
                    },
                    "inputWidgets": {},
                    "nuid": "342b436b-fcc3-41a2-930e-40b14af52158",
                    "showTitle": false,
                    "title": ""
                },
                "jupyter": {
                    "outputs_hidden": true
                },
                "azdata_cell_guid": "0560c6b6-9cf1-4d23-974f-fd4f599127ab",
                "language": "sql"
            },
            "outputs": [],
            "execution_count": null
        },
        {
            "cell_type": "code",
            "source": [
                "# Efficiently partition the Run_Steps into Sequence_Name folders\n",
                "run_steps.write.mode(\"overwrite\").option(\"header\", \"true\").partitionBy(\"SEQUENCE_NAME\").csv(silver_target)\n"
            ],
            "metadata": {
                "application/vnd.databricks.v1+cell": {
                    "cellMetadata": {
                        "byteLimit": 2048000,
                        "rowLimit": 10000
                    },
                    "inputWidgets": {},
                    "nuid": "dbca3a79-cf94-4035-9d18-6f981c9c3ee7",
                    "showTitle": false,
                    "title": ""
                },
                "azdata_cell_guid": "1cfc1d65-2ebc-4005-8244-d8b3639092ab",
                "language": "sql"
            },
            "outputs": [],
            "execution_count": null
        }
    ]
}