Publishing v2.2.6

Author: SireInsectus

ASP 5 - Streaming/ASP 5.1aL - Coupon Sales Lab.py

@@ -23,7 +23,7 @@
 
 # COMMAND ----------
 
-# MAGIC %run ../Includes/Classroom-Setup
+# MAGIC %run ../Includes/Classroom-Setup-5.1a
 
 # COMMAND ----------
 
@@ -45,9 +45,7 @@
 
 # COMMAND ----------
 
-assert df.isStreaming
-assert df.columns == ["order_id", "email", "transaction_timestamp", "total_item_quantity", "purchase_revenue_in_usd", "unique_items", "items"]
-print("All test pass")
+DA.tests.validate_1_1(df)
 
 # COMMAND ----------
 
@@ -69,9 +67,7 @@
 
 # COMMAND ----------
 
-schema_str = str(coupon_sales_df.schema)
-assert "StructField(items,StructType(List(StructField(coupon" in schema_str, "items column was not exploded"
-print("All test pass")
+DA.tests.validate_2_1(coupon_sales_df.schema)
 
 # COMMAND ----------
 
@@ -90,21 +86,17 @@
 coupons_checkpoint_path = f"{DA.paths.checkpoints}/coupon-sales"
 coupons_output_path = f"{DA.paths.working_dir}/coupon-sales/output"
 
-coupon_sales_query = (coupon_sales_df.FILL_IN
- )
+coupon_sales_query = (coupon_sales_df.FILL_IN)
+
+DA.block_until_stream_is_ready(coupon_sales_query)
 
 # COMMAND ----------
 
 # MAGIC %md **3.1: CHECK YOUR WORK**
 
 # COMMAND ----------
 
-DA.block_until_stream_is_ready("coupon_sales")
-assert coupon_sales_query.isActive
-assert len(dbutils.fs.ls(coupons_output_path)) > 0
-assert len(dbutils.fs.ls(coupons_checkpoint_path)) > 0
-assert "coupon_sales" in coupon_sales_query.lastProgress["name"]
-print("All test pass")
+DA.tests.validate_3_1(coupon_sales_query)
 
 # COMMAND ----------
 
@@ -128,9 +120,7 @@
 
 # COMMAND ----------
 
-assert type(query_id) == str
-assert list(query_status.keys()) == ["message", "isDataAvailable", "isTriggerActive"]
-print("All test pass")
+DA.tests.validate_4_1(query_id, query_status)
 
 # COMMAND ----------
 
@@ -148,8 +138,7 @@
 
 # COMMAND ----------
 
-assert not coupon_sales_query.isActive
-print("All test pass")
+DA.tests.validate_5_1(coupon_sales_query)
 
 # COMMAND ----------
 

ASP 5 - Streaming/ASP 5.1bL - Hourly Activity by Traffic Lab.py

@@ -22,7 +22,7 @@
 
 # COMMAND ----------
 
-# MAGIC %run ../Includes/Classroom-Setup
+# MAGIC %run ../Includes/Classroom-Setup-5.1b
 
 # COMMAND ----------
 
@@ -31,12 +31,10 @@
 # Directory of hourly events logged from the BedBricks website on July 3, 2020
 hourly_events_path = f"{DA.paths.datasets}/ecommerce/events/events-2020-07-03.json"
 
-df = (spark
- .readStream
- .schema(schema)
- .option("maxFilesPerTrigger", 1)
- .json(hourly_events_path)
- )
+df = (spark.readStream
+ .schema(schema)
+ .option("maxFilesPerTrigger", 1)
+ .json(hourly_events_path))
 
 # COMMAND ----------
 
@@ -49,17 +47,15 @@
 # COMMAND ----------
 
 # TODO
-events_df = (df.FILL_IN
- )
+events_df = (df.FILL_IN)
 
 # COMMAND ----------
 
 # MAGIC %md **1.1: CHECK YOUR WORK**
 
 # COMMAND ----------
 
-assert "StructField(createdAt,TimestampType,true" in str(events_df.schema)
-print("All test pass")
+DA.tests.validate_1_1(events_df.schema)
 
 # COMMAND ----------
 
@@ -86,8 +82,7 @@
 
 # COMMAND ----------
 
-assert str(traffic_df.schema) == "StructType(List(StructField(traffic_source,StringType,true),StructField(active_users,LongType,false),StructField(hour,IntegerType,true)))"
-print("All test pass")
+DA.tests.validate_2_1(traffic_df.schema)
 
 # COMMAND ----------
 
@@ -132,8 +127,7 @@
 
 # COMMAND ----------
 
-for s in spark.streams.active:
- print(s.name)
+DA.tests.validate_4_1()
 
 # COMMAND ----------
 

ASP 5 - Streaming/ASP 5.1cL - Activity by Traffic Lab.py

@@ -31,32 +31,28 @@
 
 # COMMAND ----------
 
-# MAGIC %run ../Includes/Classroom-Setup
+# MAGIC %run ../Includes/Classroom-Setup-5.1c
 
 # COMMAND ----------
 
 # MAGIC %md ### 1. Read data stream
 # MAGIC - Set to process 1 file per trigger
 # MAGIC - Read from Delta with filepath stored in **`DA.paths.events`**
 # MAGIC 
-# MAGIC Assign the resulting DataFrame to **`df`**.
+# MAGIC Assign the resulting Query to **`df`**.
 
 # COMMAND ----------
 
 # TODO
 df = FILL_IN
 
-df.isStreaming
-
 # COMMAND ----------
 
 # MAGIC %md **1.1: CHECK YOUR WORK**
 
 # COMMAND ----------
 
-assert df.isStreaming
-assert df.columns == ["device", "ecommerce", "event_name", "event_previous_timestamp", "event_timestamp", "geo", "items", "traffic_source", "user_first_touch_timestamp", "user_id"]
-print("All test pass")
+DA.tests.validate_1_1(df)
 
 # COMMAND ----------
 
@@ -79,8 +75,7 @@
 
 # COMMAND ----------
 
-assert str(traffic_df.schema) == "StructType(List(StructField(traffic_source,StringType,true),StructField(active_users,LongType,false)))"
-print("All test pass")
+DA.tests.validate_2_1(traffic_df.schema)
 
 # COMMAND ----------
 
@@ -117,11 +112,7 @@
 
 # COMMAND ----------
 
-DA.block_until_stream_is_ready("active_users_by_traffic")
-assert traffic_query.isActive
-assert "active_users_by_traffic" in traffic_query.name
-assert traffic_query.lastProgress["sink"]["description"] == "MemorySink"
-print("All test pass")
+DA.tests.validate_4_1(traffic_query)
 
 # COMMAND ----------
 
@@ -163,8 +154,7 @@
 
 # COMMAND ----------
 
-assert not traffic_query.isActive
-print("All test pass")
+DA.tests.validate_6_1(traffic_query)
 
 # COMMAND ----------
 

Includes/Classroom-Setup-5.1a.py

@@ -0,0 +1,109 @@
+# Databricks notebook source
+# MAGIC %run ./_common
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_1_1(self, df):
+ suite = DA.tests.new("5.1a-1.1")
+
+ suite.test_true(df.isStreaming, description="The query is streaming")
+ 
+ columns = ['order_id', 'email', 'transaction_timestamp', 'total_item_quantity', 'purchase_revenue_in_usd', 'unique_items', 'items']
+ suite.test_sequence(actual_value=df.columns, 
+ expected_value=columns,
+ test_column_order=False,
+ description=f"DataFrame contains all {len(columns)} columns",
+ hint="Found [[ACTUAL_VALUE]]")
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_2_1(self, schema:StructType):
+ 
+ suite = DA.tests.new("5.1a-2.1")
+ 
+ suite.test_equals(actual_value=type(schema), expected_value=StructType, description="Schema is of type StructType", hint="Found [[ACTUAL_VALUE]]")
+ 
+ suite.test_length(schema.fieldNames(), 7, description="Schema contians seven fields", hint="Found [[LEN_ACTUAL_VALUE]]: [[ACTUAL_VALUE]]")
+
+ suite.test_struct_field(schema, "order_id", "LongType", None)
+ suite.test_struct_field(schema, "email", "StringType", None) 
+ suite.test_struct_field(schema, "transaction_timestamp", "LongType", None)
+ suite.test_struct_field(schema, "total_item_quantity", "LongType", None)
+ suite.test_struct_field(schema, "purchase_revenue_in_usd", "DoubleType", None)
+ suite.test_struct_field(schema, "unique_items", "LongType", None)
+ suite.test_struct_field(schema, "items", "StructType", None)
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_3_1(self, query):
+ suite = DA.tests.new("5.1a-3.1")
+
+ suite.test_true(query.isActive, description="The query is active")
+
+ suite.test_equals(coupon_sales_query.lastProgress["name"], "coupon_sales",
+ description="The query name is \"coupon_sales\".")
+ 
+ coupons_output_path = f"{DA.paths.working_dir}/coupon-sales/output"
+ suite.test(actual_value=None, test_function=lambda: len(dbutils.fs.ls(coupons_output_path)) > 0,
+ description=f"Found at least one file in .../coupon-sales/output")
+
+ coupons_checkpoint_path = f"{DA.paths.checkpoints}/coupon-sales"
+ suite.test(actual_value=None, test_function=lambda: len(dbutils.fs.ls(coupons_checkpoint_path)) > 0,
+ description=f"Found at least one file in .../coupon-sales")
+
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_4_1(self, query_id, query_status):
+ suite = DA.tests.new("5.1a-4.1")
+
+ suite.test_sequence(actual_value=query_status.keys(),
+ expected_value=["message", "isDataAvailable", "isTriggerActive"],
+ test_column_order=False,
+ description="Valid status value.")
+
+ suite.test_equals(type(query_id), str, description="Valid query_id value.")
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_5_1(self, query):
+ suite = DA.tests.new("5.1a-5.1")
+
+ suite.test_false(query.isActive, description="The query is not active")
+
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+
+# COMMAND ----------
+
+DA = DBAcademyHelper(**helper_arguments) # Create the DA object
+DA.reset_environment() # Reset by removing databases and files from other lessons
+DA.init(install_datasets=True, # Initialize, install and validate the datasets
+ create_db=True) # Continue initialization, create the user-db
+
+DA.paths.sales = f"{DA.paths.datasets}/ecommerce/sales/sales.delta"
+DA.paths.users = f"{DA.paths.datasets}/ecommerce/users/users.delta"
+DA.paths.events = f"{DA.paths.datasets}/ecommerce/events/events.delta"
+DA.paths.products = f"{DA.paths.datasets}/products/products.delta"
+
+DA.conclude_setup() # Conclude setup by advertising environmental changes
+

Includes/Classroom-Setup-5.1b.py

@@ -0,0 +1,73 @@
+# Databricks notebook source
+# MAGIC %run ./_common
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_1_1(self, schema):
+ suite = DA.tests.new("5.1b-1.1")
+
+ suite.test_equals(actual_value=type(schema), expected_value=StructType, description="Schema is of type StructType", hint="Found [[ACTUAL_VALUE]]")
+ 
+ suite.test_length(schema.fieldNames(), 12, description="Schema contians 12 field", hint="Found [[LEN_ACTUAL_VALUE]]: [[ACTUAL_VALUE]]")
+ 
+ suite.test_struct_field(schema, "device", "StringType", None)
+ suite.test_struct_field(schema, "ecommerce", "StructType", None)
+ suite.test_struct_field(schema, "event_name", "StringType", None)
+ suite.test_struct_field(schema, "event_previous_timestamp", "LongType", None)
+ suite.test_struct_field(schema, "event_timestamp", "LongType", None)
+ suite.test_struct_field(schema, "geo", "StructType", None)
+ suite.test_struct_field(schema, "items", "ArrayType", None)
+ suite.test_struct_field(schema, "traffic_source", "StringType", None)
+ suite.test_struct_field(schema, "user_first_touch_timestamp", "LongType", None)
+ suite.test_struct_field(schema, "user_id", "StringType", None)
+ suite.test_struct_field(schema, "hour", "IntegerType", None)
+ suite.test_struct_field(schema, "createdAt", "TimestampType", None)
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_2_1(self, schema):
+ suite = DA.tests.new("5.1b-2.1")
+
+ suite.test_equals(actual_value=type(schema), expected_value=StructType, description="Schema is of type StructType", hint="Found [[ACTUAL_VALUE]]")
+ 
+ suite.test_length(schema.fieldNames(), 3, description="Schema contians three field", hint="Found [[LEN_ACTUAL_VALUE]]: [[ACTUAL_VALUE]]")
+ 
+ suite.test_struct_field(schema, "traffic_source", "StringType", None)
+ suite.test_struct_field(schema, "active_users", "LongType", None)
+ suite.test_struct_field(schema, "hour", "IntegerType", None)
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+
+# COMMAND ----------
+
+@TestHelper.monkey_patch
+def validate_4_1(self):
+ suite = DA.tests.new("5.1b-4.1")
+
+ suite.test_length(spark.streams.active, 0, description="All queries have stopped streaming")
+ 
+ suite.display_results()
+ assert suite.passed, "One or more tests failed."
+
+# COMMAND ----------
+
+DA = DBAcademyHelper(**helper_arguments) # Create the DA object
+DA.reset_environment() # Reset by removing databases and files from other lessons
+DA.init(install_datasets=True, # Initialize, install and validate the datasets
+ create_db=True) # Continue initialization, create the user-db
+
+DA.paths.sales = f"{DA.paths.datasets}/ecommerce/sales/sales.delta"
+DA.paths.users = f"{DA.paths.datasets}/ecommerce/users/users.delta"
+DA.paths.events = f"{DA.paths.datasets}/ecommerce/events/events.delta"
+DA.paths.products = f"{DA.paths.datasets}/products/products.delta"
+
+DA.conclude_setup() # Conclude setup by advertising environmental changes
+