From 69a990701a070933280f05d4e4580fabe6817bea Mon Sep 17 00:00:00 2001 From: Haille W Date: Sat, 21 Feb 2026 11:01:57 +1100 Subject: [PATCH] fix notebook issue for sample run --- .../src/run_1_staging_load.ipynb | 580 +++--------------- 1 file changed, 89 insertions(+), 491 deletions(-) diff --git a/samples/test_data_and_orchestrator/src/run_1_staging_load.ipynb b/samples/test_data_and_orchestrator/src/run_1_staging_load.ipynb index bb8311f..440608e 100644 --- a/samples/test_data_and_orchestrator/src/run_1_staging_load.ipynb +++ b/samples/test_data_and_orchestrator/src/run_1_staging_load.ipynb @@ -1,496 +1,92 @@ { - "cells": [ - { - "cell_type": "markdown", - "metadata": { - "application/vnd.databricks.v1+cell": { - "cellMetadata": {}, - "inputWidgets": {}, - "nuid": "ee353e42-ff58-4955-9608-12865bd0950e", - "showTitle": false, - "title": "" - } - }, - "source": [ - "# Day 1 Load" - ] - }, - { - "cell_type": "code", - "execution_count": 2, - "metadata": {}, - "outputs": [], - "source": [ - "%run \"./initialize\"" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Base Tables and Sources" - ] - }, - { - "cell_type": "code", - "execution_count": 0, - "metadata": { - "application/vnd.databricks.v1+cell": { - "cellMetadata": { - "byteLimit": 2048000, - "rowLimit": 10000 - }, - "inputWidgets": {}, - "nuid": "6bca260b-13d1-448f-8082-30b60a85c9ae", - "showTitle": false, - "title": "" - } - }, - "outputs": [], - "source": [ - "# Base Data Load\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " DELETE_FLAG,\n", - " LOAD_TIMESTAMP)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com', NULL, '2023-01-01 10:00:00')\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', NULL, '2023-01-01 10:00:00')\n", - " , (10, 'Richard', 'Johnson', 'richard.johnson@example.com', NULL, '2023-01-01 10:00:00')\"\"\")\n", - "\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_address (\n", - " CUSTOMER_ID,\n", - " CITY,\n", - " STATE,\n", - " LOAD_TIMESTAMP)\n", - "VALUES\n", - " (1, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", - " , (2, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", - " , (NULL, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", - " , (4, 'Hobart', 'TAS', '2023-01-01 10:00:00')\n", - " , (10, 'Sydney', 'NSW', '2023-01-01 10:00:00')\"\"\")\n", - "\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_purchase (\n", - " CUSTOMER_ID,\n", - " PRODUCT,\n", - " QUANTITY,\n", - " PRICE,\n", - " PURCHASE_TIMESTAMP)\n", - "VALUES\n", - " (1, 'Apples', 1, 10.00, '2023-01-01 10:00:00')\n", - " , (1, 'Bananas', 2, 20.00, '2023-01-01 10:00:00')\n", - " , (2, 'Oranges', 3, 30.00, '2023-01-01 10:00:00')\n", - " , (2, 'Pears', 4, 40.00, '2023-01-01 10:00:00')\n", - " , (10, 'Apples', 5, 50.00, '2023-01-01 10:00:00')\n", - " , (10, 'Bananas', 6, 60.00, '2023-01-01 10:00:00')\"\"\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Base File Load\n", - "file_content = \"\"\"CUSTOMER_ID,FIRST_NAME,LAST_NAME,EMAIL,DELETE_FLAG,LOAD_TIMESTAMP\\n\n", - "1,John,Doe,john.doe@example.com,,2023-01-01 10:00:00\\n\n", - "2,Jane,Smith,jane.smith@example.com,,2023-01-01 10:00:00\\n\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_file_path}/customer_1.csv\",\n", - " file_content,\n", - " True)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Feature Tables and Sources" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Snapshot Sources\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_snapshot_source (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " DELETE_FLAG,\n", - " LOAD_TIMESTAMP)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com', NULL, '2023-01-01 10:00:00')\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', NULL, '2023-01-01 10:00:00')\n", - " , (10, 'Richard', 'Johnson', 'richard.johnson@example.com', NULL, '2023-01-01 10:00:00')\"\"\")\n", - "\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_historical_snapshot_source (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " LOAD_TIMESTAMP)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com', '2024-01-01 10:00:00')\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', '2024-01-01 10:00:00')\n", - " , (1, 'John', 'Doe', 'jdoe@example.com', '2024-01-04 10:00:00')\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', '2024-01-04 10:00:00')\n", - " , (3, 'Alice', 'Green', 'alice.green@example.com', '2024-01-04 10:00:00')\n", - " , (4, 'Joe', 'Bloggs', 'joe.bloggs@example.com', '2023-01-04 10:00:00')\n", - " , (1, 'John', 'Doe', 'jdoe@example.com', '2024-02-10 10:00:00')\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', '2024-02-10 10:00:00')\n", - " , (3, 'Alice', 'Green', 'alice.green@example.com', '2024-02-10 10:00:00')\"\"\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Historical table snapshots\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_snapshots (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " UPDATE_TIMESTAMP,\n", - " SNAPSHOT_TIMESTAMP,\n", - " SNAPSHOT_VERSION)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com', '2023-01-01 00:00:00', '2023-01-01 00:00:00',0)\n", - " , (2, 'Jane', 'Smith', 'jane.smith@example.com', '2023-01-01 00:00:00', '2023-01-01 00:00:00',0)\n", - " , (10, 'Richard', 'Johnson', 'richard.johnson@example.com', '2023-01-01 00:00:00','2023-01-01 00:00:00', 0)\"\"\")\n", - "\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_snapshots (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " UPDATE_TIMESTAMP,\n", - " SNAPSHOT_TIMESTAMP,\n", - " SNAPSHOT_VERSION)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'jdoe@example.com', '2023-01-02 00:00:00','2023-01-02 10:00:00', 1)\n", - " , (3, 'Alice', 'Green', 'alice.green@example.com', '2023-01-02 00:00:00','2023-01-02 10:00:00', 1)\n", - " , (4, 'Joe', 'Bloggs', 'joe.bloggs@example.com', '2023-01-02 00:00:00','2023-01-02 10:00:00', 1)\n", - " , (10, 'Richard', 'Johnson', 'richard.johnson@example.com', '2023-01-01 00:00:00','2023-01-02 10:00:00', 1);\"\"\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Delete all files and directories in snapshot directories\n", - "dbutils.fs.rm(customer_snapshot_file_path, True)\n", - "dbutils.fs.rm(customer_snapshot_partitioned_file_path, True)\n", - "dbutils.fs.rm(customer_snapshot_partitioned_parquet_file_path, True)\n", - "dbutils.fs.rm(template_samples_customer_file_path, True)\n", - "dbutils.fs.rm(template_samples_customer_address_file_path, True)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Historical Snapshot File Sources\n", - "\n", - "file_content = \"\"\"CUSTOMER_ID,FIRST_NAME,LAST_NAME,EMAIL,DELETE_FLAG,LOAD_TIMESTAMP\\n\n", - "1,John,Doe,john.doe@example.com,,2024-01-01 10:00:00\\n\n", - "2,Jane,Smith,jane.smith@example.com,,2024-01-01 10:00:00\\n\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_file_path}/customer_2024_01_01.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_partitioned_file_path}/YEAR=2024/MONTH=01/DAY=01/customer.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_file_path}/customer_2024_01_01.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "file_content = \"\"\"CUSTOMER_ID,FIRST_NAME,LAST_NAME,EMAIL,DELETE_FLAG,LOAD_TIMESTAMP\\n\n", - "1,John,Doe,jdoe@example.com,,2024-01-04 10:00:00\\n\n", - "3,Alice,Green,alice.green@example.com,,2024-01-04 10:00:00\\n\n", - "4,Joe,Bloggs,joe.bloggs@example.com,,2024-01-04 10:00:00\\n\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_file_path}/customer_2024_01_04.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_file_path}/sub_dir_test/customer_2024_01_04.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_partitioned_file_path}/YEAR=2024/MONTH=01/DAY=04/customer.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_file_path}/customer_2024_01_04.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "\n", - "file_content = \"\"\"CUSTOMER_ID,FIRST_NAME,LAST_NAME,EMAIL,DELETE_FLAG,LOAD_TIMESTAMP\\n\n", - "1,John,Doe,jdoe@example.com,,2024-02-10 10:00:00\\n\n", - "3,Alice,Green,alice.green@example.com,,2024-02-10 10:00:00\\n\n", - "4,Joe,Bloggs,joe.bloggs@example.com,,2024-02-10 10:00:00\\n\n", - "5,Sarah,Jones,sarah.jones@example.com,,2024-02-10 10:00:00\\n\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_file_path}/customer_2024_02_10.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{customer_snapshot_partitioned_file_path}/YEAR=2024/MONTH=02/DAY=10/customer.csv\",\n", - " file_content,\n", - " True)\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_file_path}/customer_2024_02_10.csv\",\n", - " file_content,\n", - " True)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Historical Snapshot File Sources for customer_address\n", - "\n", - "customer_address_file_content = \"\"\"CUSTOMER_ID,CITY,STATE,LOAD_TIMESTAMP\n", - "1,Melbourne,VIC,2024-01-01 10:00:00\n", - "2,Melbourne,VIC,2024-01-01 10:00:00\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_address_file_path}/customer_address_2024_01_01.csv\",\n", - " customer_address_file_content,\n", - " True\n", - ")\n", - "\n", - "customer_address_file_content = \"\"\"CUSTOMER_ID,CITY,STATE,LOAD_TIMESTAMP\n", - "1,Melbourne,VIC,2024-01-04 10:00:00\n", - "4,Hobart,TAS,2024-01-04 10:00:00\n", - "10,Sydney,NSW,2024-01-04 10:00:00\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_address_file_path}/customer_address_2024_01_04.csv\",\n", - " customer_address_file_content,\n", - " True\n", - ")\n", - "\n", - "customer_address_file_content = \"\"\"CUSTOMER_ID,CITY,STATE,LOAD_TIMESTAMP\n", - "1,Sydney,NSW,2024-02-10 10:00:00\n", - "4,Hobart,TAS,2024-02-10 10:00:00\n", - "10,Brisbane,QLD,2024-02-10 10:00:00\n", - "12,Perth,WA,2024-02-10 10:00:00\n", - "\"\"\"\n", - "\n", - "dbutils.fs.put(\n", - " f\"{template_samples_customer_address_file_path}/customer_address_2024_02_10.csv\",\n", - " customer_address_file_content,\n", - " True\n", - ")\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from pyspark.sql.types import StructType, StructField, StringType\n", - "\n", - "# Historical Snapshot Parquet Sources \n", - "schema = StructType([\n", - " StructField(\"CUSTOMER_ID\", StringType(), True),\n", - " StructField(\"FIRST_NAME\", StringType(), True),\n", - " StructField(\"LAST_NAME\", StringType(), True),\n", - " StructField(\"EMAIL\", StringType(), True),\n", - " StructField(\"DELETE_FLAG\", StringType(), True),\n", - " StructField(\"LOAD_TIMESTAMP\", StringType(), True)\n", - "])\n", - "\n", - "data = [\n", - " [\"1\", \"John\", \"Doe\", \"john.doe@example.com\", \"\", \"2024-01-01 10:00:00\"],\n", - " [\"2\", \"Jane\", \"Smith\", \"jane.smith@example.com\", \"\", \"2024-01-01 10:00:00\"],\n", - " [\"1\", \"John\", \"Doe\", \"john.doe@example.com\", \"\", \"2024-01-01 10:00:00\"]\n", - "\n", - "]\n", - "\n", - "df = spark.createDataFrame(data, schema=schema)\n", - "\n", - "df.write.parquet(\n", - " f\"{customer_snapshot_partitioned_parquet_file_path}/YEAR=2024/MONTH=01/DAY=01/customer.parquet\",\n", - " mode=\"overwrite\"\n", - ")\n", - "\n", - "\n", - "data = [\n", - " [\"1\", \"John\", \"Doe\", \"jdoe@example.com\", \"\", \"2024-01-04 10:00:00\"],\n", - " [\"3\", \"Alice\", \"Green\", \"alice.green@example.com\", \"\", \"2024-01-04 10:00:00\"], \n", - " [\"4\", \"Joe\", \"Bloggs\", \"joe.bloggs@example.com\", \"\", \"2024-01-04 10:00:00\"]\n", - "]\n", - "\n", - "df = spark.createDataFrame(data, schema=schema) \n", - "\n", - "df.write.parquet(\n", - " f\"{customer_snapshot_partitioned_parquet_file_path}/YEAR=2024/MONTH=01/DAY=04/customer.parquet\",\n", - " mode=\"overwrite\"\n", - ")\n", - "\n", - "\n", - "data = [\n", - " [\"1\", \"John\", \"Doe\", \"jdoe@example.com\", \"\", \"2024-02-10 10:00:00\"],\n", - " [\"3\", \"Alice\", \"Green\", \"alice.green@example.com\", \"\", \"2024-02-10 10:00:00\"], \n", - " [\"4\", \"Joe\", \"Bloggs\", \"joe.bloggs@example.com\", \"\", \"2024-02-10 10:00:00\"],\n", - " [\"5\", \"Sarah\", \"Jones\", \"sarah.jones@example.com\", \"\", \"2024-02-10 10:00:00\"]\n", - "]\n", - "\n", - "\n", - "df = spark.createDataFrame(data, schema=schema)\n", - "\n", - "df.write.parquet(\n", - " f\"{customer_snapshot_partitioned_parquet_file_path}/YEAR=2024/MONTH=02/DAY=10/customer.parquet\",\n", - " mode=\"overwrite\"\n", - ")\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Kafka Sink Sample Source Data Load\n", - "message_payload = '{\"Mortgage_id\": \"123\", \"Mortgage_fac\": \"M1-str\", \"Mortgage_score\": 30}'\n", - "spark.sql(f\"INSERT INTO {staging_schema}.kafka_sink_sample_source (Message_payload) VALUES ('{message_payload}')\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Table Migration Source\n", - "# Simulate a table ready for migration into a pipleline.\n", - "# This simulates a table being populated outside of SDP by classic pyspark jobs, from the same sources.\n", - "# Read the documentation on Table Migration for more details.\n", - "\n", - "# Simulates Day 1 data load.\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {bronze_schema}.table_to_migrate_scd2 (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL,\n", - " EFFECTIVE_FROM,\n", - " EFFECTIVE_TO)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com', '2023-01-01 10:00:00', NULL),\n", - " (2, 'Jane', 'Smith', 'jane.smith@example.com', '2023-01-01 10:00:00', NULL),\n", - " (10, 'Richard', 'Johnson', 'richard.johnson@example.com', '2023-01-01 10:00:00', NULL),\n", - " (30, 'Closed Same Day', 'Customer', 'closed_same_day.customer@example.com', '2023-01-01 10:00:00', '2023-01-01 10:00:00'),\n", - " (40, 'Closed Normal', 'Customer', 'cnormal.customer@example.com', '2023-01-02 10:00:00', '2023-05-01 10:00:00'),\n", - " (40, 'Closed Normal', 'Customer', 'closed_normal.customer@example.com', '2023-01-01 10:00:00', '2023-01-02 10:00:00')\"\"\")\n", - "\n", - "# Simulates Day 1 data load.\n", - "spark.sql(f\"\"\"INSERT INTO TABLE {bronze_schema}.table_to_migrate_scd0 (\n", - " CUSTOMER_ID,\n", - " FIRST_NAME,\n", - " LAST_NAME,\n", - " EMAIL)\n", - "VALUES\n", - " (1, 'John', 'Doe', 'john.doe@example.com'),\n", - " (2, 'Jane', 'Smith', 'jane.smith@example.com'),\n", - " (10, 'Richard', 'Johnson', 'richard.johnson@example.com')\"\"\")\n", - "\n", - "# Simulates Day 2 data load.\n", - "# spark.sql(f\"\"\"\n", - "# MERGE INTO {staging_schema}.table_to_migrate AS t\n", - "# USING (\n", - "# SELECT\n", - "# merge_key,\n", - "# id,\n", - "# first_name,\n", - "# last_name,\n", - "# email,\n", - "# delete_flag,\n", - "# load_timestamp\n", - "# FROM VALUES\n", - "# (NULL, 1, 'John', 'Doe', 'jdoe@example.com', NULL, TIMESTAMP('2023-01-02 10:00:00')),\n", - "# (1, 1, 'John', 'Doe', 'jdoe@example.com', NULL, TIMESTAMP('2023-01-02 10:00:00')),\n", - "# (3, 3, 'Alice', 'Green', 'alice.green@example.com', NULL, TIMESTAMP('2023-01-02 10:00:00')),\n", - "# (4, 4, 'Joe', 'Bloggs', 'joe.bloggs@example.com', NULL, TIMESTAMP('2023-01-02 10:00:00')),\n", - "# (10, 10,'Richard', 'Johnson','richard.johnson@example.com', 1, TIMESTAMP('2023-01-02 10:00:00'))\n", - "# AS t(merge_key, id, first_name, last_name, email, delete_flag, load_timestamp)\n", - "# ) AS src\n", - "# ON t.customer_id = src.merge_key and t.effective_to IS NULL\n", - "# WHEN MATCHED AND (\n", - "# t.first_name <> src.first_name\n", - "# OR t.last_name <> src.last_name\n", - "# OR t.email <> src.email\n", - "# OR src.delete_flag = 1\n", - "# ) THEN\n", - "# UPDATE SET t.effective_to = src.load_timestamp\n", - "\n", - "# WHEN NOT MATCHED THEN\n", - "# INSERT (customer_id, first_name, last_name, email, effective_from, effective_to)\n", - "# VALUES (src.id, src.first_name, src.last_name, src.email, src.load_timestamp, NULL);\n", - "# \"\"\")" - ] + "cells": [ + { + "cell_type": "markdown", + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": {}, + "inputWidgets": {}, + "nuid": "ee353e42-ff58-4955-9608-12865bd0950e", + "showTitle": false, + "title": "" } - ], - "metadata": { - "application/vnd.databricks.v1+notebook": { - "dashboards": [], - "language": "python", - "notebookMetadata": { - "pythonIndentUnit": 2 - }, - "notebookName": "notebook", - "widgets": {} - }, - "kernelspec": { - "display_name": "Python 3", - "language": "python", - "name": "python3" - }, - "language_info": { - "name": "python", - "version": "3.11.4" + }, + "source": [ + "# Day 1 Load" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "metadata": {}, + "outputs": [], + "source": [ + "%run \"./initialize\"" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Base Tables and Sources" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "inputWidgets": {}, + "nuid": "6bca260b-13d1-448f-8082-30b60a85c9ae", + "showTitle": false, + "title": "" } + }, + "outputs": [], + "source": [ + "# Base Data Load\n", + "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer (\n", + " CUSTOMER_ID,\n", + " FIRST_NAME,\n", + " LAST_NAME,\n", + " EMAIL,\n", + " DELETE_FLAG,\n", + " LOAD_TIMESTAMP)\n", + "VALUES\n", + " (1, 'John', 'Doe', 'john.doe@example.com', NULL, '2023-01-01 10:00:00')\n", + " , (2, 'Jane', 'Smith', 'jane.smith@example.com', NULL, '2023-01-01 10:00:00')\n", + " , (10, 'Richard', 'Johnson', 'richard.johnson@example.com', NULL, '2023-01-01 10:00:00')\"\"\")\n", + "\n", + "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_address (\n", + " CUSTOMER_ID,\n", + " CITY,\n", + " STATE,\n", + " LOAD_TIMESTAMP)\n", + "VALUES\n", + " (1, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", + " , (2, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", + " , (NULL, 'Melbourne', 'VIC', '2023-01-01 10:00:00')\n", + " , (4, 'Hobart', 'TAS', '2023-01-01 10:00:00')\n", + " , (10, 'Sydney', 'NSW', '2023-01-01 10:00:00')\"\"\")\n", + "\n", + "spark.sql(f\"\"\"INSERT INTO TABLE {staging_schema}.customer_purchase (\n", + " CUSTOMER_ID,\n", + " PRODUCT,\n", + " QUANTITY,\n", + " PRICE,\n", + " PURCHASE_TIMESTAMP)\n", + "VALUES\n", + " (1, 'Apples', 1, 10.00, '2023-01-01 10:00:00')\n", + " , (1, 'Bananas', 2, 20.00, '2023-01-01 10:00:00')\n", + " , (2, 'Oranges', 3, 30.00, '2023-01-01 10:00:00')\n", + " , (2, 'Pears', 4, 40.00, '2023-01-01 10:00:00')\n", + " , (10, 'Apples', 5, 50.00, '2023-01-01 10:00:00')\n", + " , (10, 'Bananas', 6, 60.00, '2023-01-01 10:00:00')\"\"\")" + ] }, { "cell_type": "code", @@ -777,7 +373,9 @@ "\n", "data = [\n", " [\"1\", \"John\", \"Doe\", \"john.doe@example.com\", \"\", \"2024-01-01 10:00:00\"],\n", - " [\"2\", \"Jane\", \"Smith\", \"jane.smith@example.com\", \"\", \"2024-01-01 10:00:00\"]\n", + " [\"2\", \"Jane\", \"Smith\", \"jane.smith@example.com\", \"\", \"2024-01-01 10:00:00\"],\n", + " [\"1\", \"John\", \"Doe\", \"john.doe@example.com\", \"\", \"2024-01-01 10:00:00\"]\n", + "\n", "]\n", "\n", "df = spark.createDataFrame(data, schema=schema)\n",