Major update. Dockerfile and style correction.

I added an output folder where the database and the html will be saved. The dockerfile is working and ready for testing in mybinder. Requirements.txt has been updated with my versions.

Author: Carlos Vivar

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "python.linting.flake8Enabled": true,
+    "python.linting.enabled": true,
+    "python.analysis.typeCheckingMode": "basic"
+}

COVID19_project/__main__.py

@@ -11,6 +11,7 @@
 
 from dependencies.spark import start_spark
 
+
 def main():
     """Main ETL script definition.
 
@@ -19,29 +20,28 @@ def main():
 
     # start Spark application and get Spark session, logger and config
     spark, log, config = start_spark(
-       app_name='covid19',
-       files=['configs/config.json'])
+       app_name="covid19",
+       files=["configs/config.json"])
 
     # log that main ETL job is starting
-    log.warn('covid19 job is up-and-running')
+    log.warn("covid19 job is up-and-running")
 
     # execute ETL pipeline
     data = extract_data(spark)
     data_transformed = transform_data(data)
     load_data(spark, data_transformed)
 
     # execute visualization pipeline
-    data_loaded = read_data(spark) # Here we can specify the query
-    bokeh_app(data_loaded) # Here we can specify the path
+    data_loaded = read_data(spark)
+    bokeh_app(data_loaded)
 
     # log the success and terminate Spark application
-    log.warn('covid19 job is finished')
+    log.warn("covid19 job is finished")
     spark.stop()
 
     return None
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     # Check if the api is online and that the version is compatible. 
     main()
-    

COVID19_project/extract.py

@@ -11,25 +11,26 @@
 from pyspark.sql import SparkSession, DataFrame
 
 
-def data_validation(df: DataFrame) -> bool:
+def data_validation(df: pd.DataFrame) -> bool:
     """Validate data extracted
 
     :param df: Input DataFrame.
     :return: Validation output in boolean
     """
 
     if df.empty:
-        print('\n* No data were downloaded \n*')
+        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')
+        print("\n* Primary key check violated. Terminating extraction *\n")
 
     if df.isnull().values.any():
-        raise Exception('\n* Null values found. Terminating extraction *\n')
+        raise Exception("\n* Null values found. Terminating extraction *\n")
 
     return True
 
+
 def extract_data(spark: SparkSession) -> DataFrame:
     """Load data from Parquet file format.
 
@@ -40,24 +41,22 @@ def extract_data(spark: SparkSession) -> DataFrame:
     response = requests.get("https://api.covidtracking.com/v2/us/daily.json")
 
     if response.status_code != 200:
-        raise Exception(f'\n* Request failed with status code {response.status_code} *\n')
+        raise Exception(f"\n* Request failed with status code {response.status_code} *\n")
 
     js = response.json()
 
     dates = []
     total_cases = []
-
-    data = js['data']
+    data = js["data"]
 
     for item in data:
-        dates.append(item['date']) 
+        dates.append(item["date"]) 
         total_cases.append(item["cases"]["total"]["value"])
 
-
     # Improve this directly in spark.
     pdf = pd.DataFrame({"date": dates, "total_cases": total_cases}).dropna()
 
-    assert data_validation(pdf), '\n* Data validation not achieved *\n'
+    assert data_validation(pdf), "\n* Data validation not achieved *\n"
 
     sdf = spark.createDataFrame(pdf)
 

COVID19_project/load.py

@@ -8,21 +8,32 @@
 import os
 from pyspark.sql import SparkSession, DataFrame
 
+
 def load_data(spark: SparkSession, df: DataFrame, db_file: str = "db.parquet"):
     """Collect data locally and write to a parquet file.
 
+    :param spark: Spark session used. 
     :param df: DataFrame to store.
+    :param db_file: Database filename.
     :return: None
     """
+
+    # First let's check if output folder is created
+    if not os.path.exists("output"):
+        os.mkdir("output")
 
-    if os.path.exists(db_file):
-        df_old = spark.read.parquet(db_file).cache()
-        df_updated = df.union(df_old).dropDuplicates()#.cache()
+    # If there's a previous version of the db, we need to update it aqnd get
+    # rid of duplicates. This makes this loading process idempotem. The database 
+    # will remain similar independently of how many times this process is run. 
+    # This helps to nmake this ETL process async from others. 
+    if os.path.exists(f"output/{db_file}"):
+        df_old = spark.read.parquet(f"output/{db_file}").cache()
+        df_updated = df.union(df_old).dropDuplicates()
 
         print(f"DB updated with {df_updated.count()} entries")
 
-        df_updated.write.mode('overwrite').parquet(db_file)
+        df_updated.write.mode('overwrite').parquet(f"output/{db_file}")
     else:
-        df.write.parquet(db_file)
+        df.write.parquet(f"output/{db_file}")
 
     return None

COVID19_project/transform.py

@@ -26,6 +26,7 @@ def transform_item_date_to_datetime(date: datetime.date) -> datetime.datetime:
 
     return min_datetime
 
+
 def transform_col_date_to_datetime(df: DataFrame, input_name: str, output_name: str) -> DataFrame:
     """Transform column date into to column in datetime type.
 
@@ -40,6 +41,7 @@ def transform_col_date_to_datetime(df: DataFrame, input_name: str, output_name:
 
     return df
 
+
 def transform_col_string_to_date(df: DataFrame, input_name: str, output_name: str) -> DataFrame:
     """Transform column date in string to column date in date type.
     
@@ -49,7 +51,7 @@ def transform_col_string_to_date(df: DataFrame, input_name: str, output_name: st
     :return: Transformed Spark DataFrame.
     """
 
-    df = df.withColumn(output_name, to_date(input_name, 'yyyy-MM-dd'))
+    df = df.withColumn(output_name, to_date(input_name, "yyyy-MM-dd"))
 
     return df
 
@@ -73,7 +75,8 @@ def calc_daily_difference(df: DataFrame, input_name: str , output_name: str) ->
 
     return diff_df
 
-def calc_rolling_mean(df: DataFrame, temporal_window:int, input_name: str, output_name: str) -> DataFrame:
+
+def calc_rolling_mean(df: DataFrame, temporal_window: int, input_name: str, output_name: str) -> DataFrame:
     """Calcultation of rolling mean
 
     :param df: Input Spark DataFrame.
@@ -105,7 +108,7 @@ def transform_data(df: DataFrame) -> DataFrame:
 
     df = df.sort("datetime")
 
-    df = calc_daily_difference(df, input_name="total_cases" , output_name="difference_total_cases" )
+    df = calc_daily_difference(df, input_name="total_cases" , output_name="difference_total_cases")
 
     df = calc_rolling_mean(df, 7, input_name="difference_total_cases", output_name="rolling_mean_total_cases")
 

COVID19_project/visualize.py

@@ -8,39 +8,41 @@
 from bokeh.plotting import figure, output_file, save
 from bokeh.models import ColumnDataSource
 
-def read_data(spark: SparkSession) -> DataFrame:
+def read_data(spark: SparkSession, db_file: str = "db.parquet") -> DataFrame:
     """Read data from database
 
     :param spark: spark session
+    :param db_file: database file in output folder
     :return: Spark DataFrame
     """
-    #with spark_timezone("Europe/Berlin"):
-    data_loaded = spark.read.parquet("db.parquet")
+
+    data_loaded = spark.read.parquet(f"output/{db_file}")
     data_loaded = data_loaded.sort("date").dropna()
 
     # Check  datetime format
 
     return data_loaded
 
 
-def bokeh_app(df: DataFrame) -> None:
+def bokeh_app(df: DataFrame, html_file: str = "covid19.html") -> None:
     """Bokeh Time-Series visualization
 
     :param df: Input Spark DataFrame
+    :param html_file: name of the file output
     :return: None
     """
 
     x = df.select("datetime").rdd.flatMap(lambda x: x).collect()
     y_diff = df.select("difference_total_cases").rdd.flatMap(lambda x: x).collect()
     y_roll = df.select("rolling_mean_total_cases").rdd.flatMap(lambda x: x).collect()
 
-    source = ColumnDataSource(data={'date': x, 'difference_total_cases': y_diff, 'rolling_mean_total_cases': y_roll})
+    source = ColumnDataSource(data={"date": x, "difference_total_cases": y_diff, "rolling_mean_total_cases": y_roll})
 
-    p = figure(title="COVID 19", x_axis_label='Date', y_axis_label='Cases', x_axis_type='datetime', plot_width=800)
-    p.line('date', 'difference_total_cases' , legend_label="Daily Difference", line_width=2, line_color='blue', source=source)
-    p.line('date', 'rolling_mean_total_cases', legend_label="Rolling Mean", line_width=2, line_color='red', source=source)
+    p = figure(title="COVID 19", x_axis_label="Date", y_axis_label="Cases", x_axis_type="datetime", plot_width=800)
+    p.line("date", "difference_total_cases", legend_label="Daily Difference", line_width=2, line_color="blue", source=source)
+    p.line("date", "rolling_mean_total_cases", legend_label="Rolling Mean", line_width=2, line_color="red", source=source)
 
-    output_file("covid19.html")
+    output_file(f"output/{html_file}")
     save(p)
 
     return None

Dockerfile

@@ -1,8 +1,10 @@
-FROM continuumio/miniconda3:4.7.12
+##FROM continuumio/miniconda3:4.7.12
+FROM jupyter/pyspark-notebook:latest
 
-RUN mkdir /opt/SDSC/
-ADD covid19_project /opt/SDSC/covid_project/
+COPY covid19_project ./covid19_project
+COPY configs ./configs
+COPY dependencies ./dependencies
+COPY tests ./tests
+COPY requirements.txt ./requirements.txt
 
-WORKDIR /opt/SDSC/
-ENV PYTHONPATH /opt/SDSC
-RUN python setup.py install
+RUN pip install -r requirements.txt

README.md

@@ -309,7 +309,9 @@ In order to test: `python -m unittest test/test_*.py`
 
 Still some work is required when using spark-submit with `$SPARK_HOME/bin/spark-submit --master local[*] --files configs/config.json covid19_project/__main__.py` 
 
+#### 28/10
 
+Applying some style corrections with flake8, and configuring correctly the docker container for mybinder. 
 
 ### Tutorial
 

docs/load.html

@@ -91,7 +91,11 @@ <h2>load.py<a class="headerlink" href="#load-py" title="Permalink to this headin
 <dd><p>Collect data locally and write to a parquet file.</p>
 <dl class="field-list simple">
 <dt class="field-odd">Parameters</dt>
-<dd class="field-odd"><p><strong>df</strong> – DataFrame to store.</p>
+<dd class="field-odd"><ul class="simple">
+<li><p><strong>spark</strong> – Spark session used.</p></li>
+<li><p><strong>df</strong> – DataFrame to store.</p></li>
+<li><p><strong>db_file</strong> – Database filename.</p></li>
+</ul>
 </dd>
 <dt class="field-even">Returns</dt>
 <dd class="field-even"><p>None</p>