Improve in ETL methods, sketch of test

Improve ETL methods, the inclusion of difference transformation, and plot of difference. Sketch of testing methods added.

Now spark is summoned with a proper class in dependencies.

Author: Carlos Vivar

COVID19_project/__main__.py

@@ -13,20 +13,21 @@
 from load import load_data
 from visualize import read_data, bokeh_app
 
+from dependencies import start_spark
+
 def main():
     """Main ETL script definition.
 
     :return: None
     """
 
     # start Spark application and get Spark session, logger and config
-    #spark, log, config = start_spark(
-    #    app_name='my_etl_job',
-    #    files=['configs/config.json'])
-    spark = SparkSession.builder.appName('COVID19').enableHiveSupport().getOrCreate()
+    spark, log, config = start_spark(
+       app_name='covid19',
+       files=['configs/config.json'])
 
     # log that main ETL job is starting
-    #log.warn('etl_job is up-and-running')
+    log.warn('covid19 job is up-and-running')
 
     # execute ETL pipeline
     data = extract_data(spark)
@@ -38,7 +39,7 @@ def main():
     bokeh_app(data_loaded) # Here we can specify the path
 
     # log the success and terminate Spark application
-    #log.warn('test_etl_job is finished')
+    log.warn('covid19 job is finished')
     spark.stop()
 
     return None

COVID19_project/extract.py

@@ -3,7 +3,27 @@
 from datetime import datetime
 import pandas as pd
 
-from pyspark.sql import SparkSession, DataFrame
+from pyspark.sql import SparkSession, 
+
+
+def data_validation(df: DataFrame) -> bool:
+    """Validate data extracted
+
+    :param df: Input DataFrame.
+    :return: Validation output in boolean
+    """
+
+    if df.empty:
+        print('\n* No data were downloaded \n*')
+        return False
+    
+    if not pd.Series(df["date"]).is_unique:
+        print('\n* Primary key check violated. Terminating extraction *\n')
+
+    if df.isnull().values.any():
+        raise Exception('\n* Null values found. Terminating extraction *\n')
+
+    return True
 
 def extract_data(spark) -> DataFrame:
     """Load data from Parquet file format.
@@ -27,7 +47,9 @@ def extract_data(spark) -> DataFrame:
         total_cases.append(item["cases"]["total"]["value"])
 
     # Improve this directly in spark.
-    pdf = pd.DataFrame({"date": dates, "total_cases": total_cases})
+    pdf = pd.DataFrame({"date": dates, "total_cases": total_cases}).dropna()
+
+    assert data_validation(pdf), '\n* Data validation not achieved *\n'
     sdf = spark.createDataFrame(pdf)
 
     return sdf

COVID19_project/transform.py

@@ -3,27 +3,9 @@
 from pyspark.sql import DataFrame
 from pyspark.sql.functions import udf, to_date
 from pyspark.sql.types import TimestampType
+import pyspark.pandas as ps
 
 
-def data_validation(df: DataFrame) -> bool:
-    """Transform original dataset.
-
-    :param df: Input DataFrame.
-    :return: Validation output in boolean
-    """
-
-    if df.empty:
-        print('\n* No data were downloaded \n*')
-        return False
-    
-    if not pd.Series(df["date"]).is_unique:
-        print('\n* Primary key check violated. Terminating extraction *\n')
-
-    if df.isnull().values.any():
-        raise Exception('\n* Null values found. Terminating extraction *\n')
-
-    return True
-
 def calc_moving_average(df: DataFrame, temporal_window:int) -> DataFrame:
     """Calcultation of moving average
 
@@ -47,6 +29,21 @@ def transform_date_to_datetime(date: datetime.date) -> datetime.datetime:
 
     return min_datetime
 
+def calc_daily_difference(df: DataFrame) -> DataFrame:
+    """Calcultation of daily difference
+
+    :param df: Input Spark DataFrame.
+    :return: Transformed DataFrame.
+    """
+    
+    psdf = df.pandas_api()
+    diff_series = psdf["total_cases"].diff()
+    diff_series.name = "difference_total_cases"
+    
+    diff_psdf = ps.merge(psdf, diff_series, left_index=True, right_index=True, how="left")
+    diff_df = diff_psdf.to_spark()
+
+    return diff_df
 
 
 def transform_data(df: DataFrame) -> DataFrame:
@@ -61,4 +58,7 @@ def transform_data(df: DataFrame) -> DataFrame:
     reg_transform_date_to_datetime = udf(lambda d: transform_date_to_datetime(d), TimestampType())
     df = df.withColumn("datetime", reg_transform_date_to_datetime("date"))
 
+    df = df.sort("datetime")
+    df = calc_daily_difference(df)
+    
     return df

COVID19_project/visualize.py

@@ -15,7 +15,7 @@ def read_data(spark: SparkSession) -> DataFrame:
     """
     #with spark_timezone("Europe/Berlin"):
     data_loaded = spark.read.parquet("db.parquet")
-    data_loaded = data_loaded.sort("date", "total_cases")
+    data_loaded = data_loaded.sort("date")
 
     # Check  datetime format
 
@@ -30,7 +30,7 @@ def bokeh_app(df: DataFrame) -> None:
     """
 
     x = df.select("datetime").rdd.flatMap(lambda x: x).collect()
-    y = df.select("total_cases").rdd.flatMap(lambda x: x).collect()
+    y = df.select("difference_total_cases").rdd.flatMap(lambda x: x).collect()
 
     p = figure(title="COVID 19", x_axis_label='Date', y_axis_label='Total Cases', x_axis_type='datetime')
     p.line(x, y, legend_label="Covid cases", line_width=2)

README.md

@@ -271,6 +271,22 @@ parkSQL = spark.sql("select * from ParquetTable where salary >= 4000 ")
 
 After dealing with some problems realted to the date/datetime format I got the first MVP of the pipeline. Now data is extracted, dates transformed into a proper datetime type, and data loaded into a parquet db. Bokeh app is able to read this data from the database and plot a simple time-series plot in html. This is the first candidate to the first release. 
 
+Useful links:
+- https://crontab.guru/
+- https://github.com/AlexIoannides/pyspark-example-project
+- https://www.youtube.com/watch?v=nVI4xEH7yU8&ab_channel=Intellipaat
+- https://github.com/rvilla87/ETL-PySpark/blob/master/jupyter/ETL.ipynb
+- https://github.com/Amaguk2023/Pyspark_Spotify_ETL
+- https://github.com/hbaflast/pyspark-project-template
+- https://github.com/vivek-bombatkar/Spark-with-Python---My-learning-notes-
+- https://github.com/hyunjoonbok/PySpark/blob/master/PySpark%20Dataframe%20Complete%20Guide%20(with%20COVID-19%20Dataset).ipynb
+- https://www.databricks.com/glossary/extract-transform-load#:~:text=ETL%2C%20which%20stands%20for%20extract,downstream%20to%20solve%20business%20problems.
+- https://www.youtube.com/watch?v=AHMm1wfGuHE&t=1s&ab_channel=TuanVu
+- https://www.revisitclass.com/hadoop/how-to-write-a-spark-dataframe-to-hive-table-in-pyspark/
+- https://hshirodkar.medium.com/apache-hive-on-docker-4d7280ac6f8e
+- 
+
+
 ### Tutorial
 
 In order to build the docker image: 

dependencies/spark.py

@@ -15,7 +15,6 @@
 
 from dependencies import logging
 
-
 def start_spark(app_name='my_spark_app', master='local[*]', jar_packages=[],
                 files=[], spark_config={}):
     """Start Spark session, get Spark logger and load config files.
@@ -24,6 +23,7 @@ def start_spark(app_name='my_spark_app', master='local[*]', jar_packages=[],
     will apply when this is called from a script sent to spark-submit.
     All other arguments exist solely for testing the script from within
     an interactive Python console.
+
     This function also looks for a file ending in 'config.json' that
     can be sent with the Spark job. If it is found, it is opened,
     the contents parsed (assuming it contains valid JSON for the ETL job
@@ -32,6 +32,7 @@ def start_spark(app_name='my_spark_app', master='local[*]', jar_packages=[],
     this function. If the file cannot be found then the return tuple
     only contains the Spark session and Spark logger objects and None
     for config.
+
     The function checks the enclosing environment to see if it is being
     run from inside an interactive console session or from an
     environment which has a `DEBUG` environment variable set (e.g.
@@ -42,6 +43,7 @@ def start_spark(app_name='my_spark_app', master='local[*]', jar_packages=[],
     to using the spark-submit and Spark cluster defaults. This will also
     use local module imports, as opposed to those in the zip archive
     sent to spark via the --py-files flag in spark-submit.
+
     :param app_name: Name of Spark app.
     :param master: Cluster connection details (defaults to local[*]).
     :param jar_packages: List of Spark JAR package names.
@@ -100,4 +102,4 @@ def start_spark(app_name='my_spark_app', master='local[*]', jar_packages=[],
         spark_logger.warn('no config file found')
         config_dict = None
 
-    return spark_sess, spark_logger, 
+    return spark_sess, spark_logger, config_dict

tests/test_transform.py

@@ -0,0 +1,82 @@
+
+"""
+test_transform.py
+~~~~~~~~~~~~~~~
+This module contains unit tests for the transformation steps of the ETL
+job defined in etl_job.py. It makes use of a local version of PySpark
+that is bundled with the PySpark package.
+"""
+
+import unittest
+import json 
+
+from pyspark.sql.functions import mean
+from dependencies.spark import start_spark
+from COVID19_project.transform import transform_data
+
+
+class SparkTransformTests(unittest.TestCase):
+    """Test suite for transformation in transform.py
+    """
+
+    def setUp(self):
+        """Start Spark, define config and path to test data
+        """
+        self.config = json.loads("""{"steps_per_floor": 21}""")
+        self.spark, *_ = start_spark()
+        self.test_data_path = 'tests/test_data/'
+
+    def tearDown(self):
+        """Stop Spark
+        """
+        self.spark.stop()
+
+    def test_transform_data(self):
+        """Test data transformer.
+        Using small chunks of input data and expected output data, we
+        test the transformation step to make sure it's working as
+        expected.
+        """
+        # assemble
+        input_data = (
+            self.spark
+            .read
+            .parquet(self.test_data_path + 'test_data'))
+
+        expected_data = (
+            self.spark
+            .read
+            .parquet(self.test_data_path + 'test_results'))
+
+        expected_cols = len(expected_data.columns)
+        expected_rows = expected_data.count()
+
+        expected_avg_steps = (
+            expected_data
+            .agg(mean('steps_to_desk').alias('avg_steps_to_desk'))
+            .collect()[0]
+            ['avg_steps_to_desk'])
+
+        # act
+        data_transformed = transform_data(input_data, 21)
+
+        cols = len(expected_data.columns)
+        rows = expected_data.count()
+        avg_steps = (
+            expected_data
+            .agg(mean('steps_to_desk').alias('avg_steps_to_desk'))
+            .collect()[0]
+            ['avg_steps_to_desk'])
+
+
+        # assert
+        self.assertEqual(expected_cols, cols)
+        self.assertEqual(expected_rows, rows)
+        self.assertEqual(expected_avg_steps, avg_steps)
+        self.assertTrue([col in expected_data.columns
+                         for col in data_transformed.columns])
+
+
+
+if __name__ == '__main__':
+    unittest.main()