<jupyter_start><jupyter_text>CS246 - Colab 9 Studying COVID-19  Setup Let's setup Spark on your Colab environment.  Run the cell below!<jupyter_code>!pip install pyspark
!pip install -U -q PyDrive
!apt install openjdk-8-jdk-headless -qq
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"<jupyter_output>Collecting pyspark
  Downloading pyspark-3.1.2.tar.gz (212.4 MB)
[K     |████████████████████████████████| 212.4 MB 66 kB/s 
[?25hCollecting py4j==0.10.9
  Downloading py4j-0.10.9-py2.py3-none-any.whl (198 kB)
[K     |████████████████████████████████| 198 kB 53.2 MB/s 
[?25hBuilding wheels for collected packages: pyspark
  Building wheel for pyspark (setup.py) ... [?25l[?25hdone
  Created wheel for pyspark: filename=pyspark-3.1.2-py2.py3-none-any.whl size=212880768 sha256=83025252e94a700476029a48389319a0610d4fe189acca9ffd66f818fff09677
  Stored in directory: /root/.cache/pip/wheels/a5/0a/c1/9561f6fecb759579a7d863dcd846daaa95f598744e71b02c77
Successfully built pyspark
Installing collected packages: py4j, pyspark
Successfully installed py4j-0.10.9 pyspark-3.1.2
The following additional packages will be installed:
  openjdk-8-jre-headless
Suggested packages:
  openjdk-8-demo openjdk-8-source libnss-mdns fonts-dejavu-extra
  fonts-ipafont-gothic fonts-ipafont-mincho fonts-wqy-microhe[...]<jupyter_text>Now we authenticate a Google Drive client to download the files we will be processing in our Spark job.**Make sure to follow the interactive instructions.**<jupyter_code>from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
from google.colab import auth
from oauth2client.client import GoogleCredentials

# Authenticate and create the PyDrive client
auth.authenticate_user()
gauth = GoogleAuth()
gauth.credentials = GoogleCredentials.get_application_default()
drive = GoogleDrive(gauth)
id='1YT7ttUAafCjbVdm6obeHp1TWAK0rEtoR'
downloaded = drive.CreateFile({'id': id})
downloaded.GetContentFile('time_series_covid19_confirmed_global.csv')

id='<KEY>'
downloaded = drive.CreateFile({'id': id})
downloaded.GetContentFile('time_series_covid19_deaths_global.csv')

id='1CNxszuZTeIw-5cF5yrzKM<KEY>'
downloaded = drive.CreateFile({'id': id})
downloaded.GetContentFile('time_series_covid19_recovered_global.csv')<jupyter_output><empty_output><jupyter_text>If you executed the cells above, you should be able to see the dataset we will use for this Colab under the "Files" tab on the left panel.Next, we import some of the common libraries needed for our task.<jupyter_code>import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline

import pyspark
from pyspark.sql import *
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark import SparkContext, SparkConf<jupyter_output><empty_output><jupyter_text>Let's initialize the Spark context.<jupyter_code># create the session
conf = SparkConf().set("spark.ui.port", "4050")

# create the context
sc = pyspark.SparkContext(conf=conf)
spark = SparkSession.builder.getOrCreate()<jupyter_output><empty_output><jupyter_text>You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**).<jupyter_code>spark<jupyter_output><empty_output><jupyter_text>If you are running this Colab on the Google hosted runtime, the cell below will create a *ngrok* tunnel which will allow you to still check the Spark UI.<jupyter_code>!wget https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip
!unzip ngrok-stable-linux-amd64.zip
get_ipython().system_raw('./ngrok http 4050 &')
!curl -s http://localhost:4040/api/tunnels | python3 -c \
    "import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])"<jupyter_output>--2021-09-28 02:30:02--  https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip
Resolving bin.equinox.io (bin.equinox.io)... 172.16.31.10, 172.16.31.10, 172.16.58.3, ...
Connecting to bin.equinox.io (bin.equinox.io)|172.16.31.10|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 13832437 (13M) [application/octet-stream]
Saving to: ‘ngrok-stable-linux-amd64.zip’

ngrok-stable-linux- 100%[===================>]  13.19M  5.44MB/s    in 2.4s    

2021-09-28 02:30:04 (5.44 MB/s) - ‘ngrok-stable-linux-amd64.zip’ saved [13832437/13832437]

Archive:  ngrok-stable-linux-amd64.zip
  inflating: ngrok                   
Traceback (most recent call last):
  File "<string>", line 1, in <module>
IndexError: list index out of range<jupyter_text>Data Loading In this Colab, we will be analyzing the timeseries data of the Coronavirus COVID-19 Global Cases, collected by Johns Hopkins CSSE.Here you can check a realtime dashboard based on this dataset: [https://www.arcgis.com/apps/opsdashboard/index.html?fbclid=IwAR2hQKsEZ3D38wVtXGryUhP9CG0Z6MYbUM_boPEaV8FBe71wUvDPc65ZG78/bda7594740fd40299423467b48e9ecf6](https://www.arcgis.com/apps/opsdashboard/index.html?fbclid=IwAR2hQKsEZ3D38wVtXGryUhP9CG0Z6MYbUM_boPEaV8FBe71wUvDPc65ZG78/bda7594740fd40299423467b48e9ecf6)---*   ```confirmed```: dataframe containing the cumulative number of confirmed COVID-19 cases, divided by geographical area*   ```deaths```: dataframe containing the cumulative number of deaths due to COVID-19, divided by geographical area*   ```recovered```: dataframe containing the cumulative number of recoevered patients, divided by geographical areaThe data sets contain data entries for each day, representing the cumulative totals as of that day.<jupyter_code>confirmed = spark.read.csv('time_series_covid19_confirmed_global.csv', header=True)
deaths = spark.read.csv('time_series_covid19_deaths_global.csv', header=True)
recovered = spark.read.csv('time_series_covid19_recovered_global.csv', header=True)
confirmed.printSchema()
confirmed.show()
deaths.show()
recovered.show()<jupyter_output>+--------------------+-------------------+---------+----------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-----[...]<jupyter_text>Your Task We are aware of the stress we are all experiencing because of the still-ongoing pandemic and the fact that many of you have projects and exams due this week due to the lack of a finals week. As such, we decided to conclude our series of Colabs with a **lightweight task** -- given everything you have learned about Spark during the quarter, this Colab should take you just a few minutes to complete. Consider the entries for May 1, 2021, in the timeseries, and compute:*   number of confirmed COVID-19 cases across the globe*   number of deaths due to COVID-19 (across the globe)*   number of recovered patients across the globe<jupyter_code># YOUR CODE HERE
number_of_confirmed = confirmed.agg({"5/1/21":"sum"}).collect()[0]["sum(5/1/21)"]
number_of_confirmed
number_of_deaths = deaths.agg({"5/1/21":"sum"}).collect()[0]["sum(5/1/21)"]
number_of_deaths
number_of_recovered = recovered.agg({"5/1/21":"sum"}).collect()[0]["sum(5/1/21)"]
number_of_recovered<jupyter_output><empty_output><jupyter_text>Consider the data points for March 1, 2020, and March 1, 2021, and filter out the geographical locations where less than 50 cases have been confirmed.For the areas still taken into consideration after the filtering step, compute the ratio between number of deaths and number of confirmed cases.<jupyter_code># YOUR CODE HERE
query_confirmed = confirmed.select('Province/State', 'Country/Region', col('3/1/21').alias('confirmed(3/1/20)'))
query_deaths = deaths.select('Province/State', 'Country/Region', col('3/1/20').alias('deaths(3/1/20)'))
query_recovered = recovered.select('Province/State', 'Country/Region', col('3/1/20').alias('recovered(3/1/20)'))

df_march_1_2020 = query_confirmed.join(query_deaths, ['Province/State', 'Country/Region'])
df_march_1_2020 = df_march_1_2020.join(query_recovered, ['Province/State', 'Country/Region'])

df_march_1_2020 = df_march_1_2020.filter(df_march_1_2020['confirmed(3/1/20)'] >= 50)

df_march_1_2020 = df_march_1_2020.withColumn("recovered/confirmed", df_march_1_2020['recovered(3/1/20)'] / df_march_1_2020['confirmed(3/1/20)'])
df_march_1_2020 = df_march_1_2020.withColumn("deaths/confirmed", df_march_1_2020['deaths(3/1/20)'] / df_march_1_2020['confirmed(3/1/20)'])
df_march_1_2020.show()
query_confirmed = confirmed.select('Province/State', 'Country/Region', col('3/1/21').alias('confirmed(3/1/21)'))
query_deaths = deaths.select('Province/State', 'Country/Region', col('3/1/21').alias('deaths(3/1/21)'))
query_recovered = recovered.select('Province/State', 'Country/Region', col('3/1/21').alias('recovered(3/1/21)'))

df_march_1_2021 = query_confirmed.join(query_deaths, ['Province/State', 'Country/Region'])
df_march_1_2021 = df_march_1_2021.join(query_recovered, ['Province/State', 'Country/Region'])

df_march_1_2021 = df_march_1_2021.filter(df_march_1_2021['confirmed(3/1/21)'] >= 50)

df_march_1_2021 = df_march_1_2021.withColumn("recovered/confirmed", df_march_1_2021['recovered(3/1/21)'] / df_march_1_2021['confirmed(3/1/21)'])
df_march_1_2021 = df_march_1_2021.withColumn("deaths/confirmed", df_march_1_2021['deaths(3/1/21)'] / df_march_1_2021['confirmed(3/1/21)'])
df_march_1_2021.show()<jupyter_output>+--------------------+--------------+-----------------+--------------+-----------------+-------------------+--------------------+
|      Province/State|Country/Region|confirmed(3/1/21)|deaths(3/1/21)|recovered(3/1/21)|recovered/confirmed|    deaths/confirmed|
+--------------------+--------------+-----------------+--------------+-----------------+-------------------+--------------------+
|Australian Capita...|     Australia|              118|             3|              115| 0.9745762711864406|0.025423728813559324|
|     New South Wales|     Australia|             5183|            54|                0|                0.0|0.010418676442214933|
|  Northern Territory|     Australia|              105|             0|              103| 0.9809523809523809|                 0.0|
|          Queensland|     Australia|             1335|             6|             1313| 0.9835205992509364|0.004494382022471...|
|     South Australia|     Australia|              617|             4|              610| 0[...]<jupyter_text>Consider the data points for March 1, 2021, and May 1, 2021, in the timeseries, and filter out the geographical locations where less than 50 deaths have been confirmed (as of March 1, 2021).For the areas still taken into consideration after the filtering step, compute the percent increase in cumulative deaths between the two dates.<jupyter_code># YOUR CODE HERE
query_confirmed = confirmed.select('Province/State', 'Country/Region', col('5/1/21').alias('confirmed(5/1/21)'))
query_deaths = deaths.select('Province/State', 'Country/Region', col('5/1/21').alias('deaths(5/1/21)'))
query_recovered = recovered.select('Province/State', 'Country/Region', col('5/1/21').alias('recovered(5/1/21)'))

df_may_1_2021 = query_confirmed.join(query_deaths, ['Province/State', 'Country/Region'])
df_may_1_2021 = df_may_1_2021.join(query_recovered, ['Province/State', 'Country/Region'])

df_may_1_2021 = df_may_1_2021.filter(df_may_1_2021['confirmed(5/1/21)'] >= 50)

df_may_1_2021 = df_may_1_2021.withColumn("recovered/confirmed", df_may_1_2021['recovered(5/1/21)'] / df_may_1_2021['confirmed(5/1/21)'])
df_may_1_2021 = df_may_1_2021.withColumn("deaths/confirmed", df_may_1_2021['deaths(5/1/21)'] / df_may_1_2021['confirmed(5/1/21)'])


df_may_1_2021.show()
result = df_may_1_2021.join(df_march_1_2021, ['Province/State', 'Country/Region'])
result = result.withColumn("percent_increase", result["deaths(3/1/21)"] / result["deaths(5/1/21)"])
result.select(['Province/State', 'Country/Region', 'percent_increase']).show()<jupyter_output>+--------------------+--------------+------------------+
|      Province/State|Country/Region|  percent_increase|
+--------------------+--------------+------------------+
|          Martinique|        France|0.6428571428571429|
|       Faroe Islands|       Denmark|               1.0|
|               Gansu|         China|               1.0|
|        Heilongjiang|         China|               1.0|
|              Fujian|         China|               1.0|
|    French Polynesia|        France|0.9858156028368794|
|Australian Capita...|     Australia|               1.0|
|     New South Wales|     Australia|               1.0|
|            Shanghai|         China|               1.0|
|           Guangdong|         China|               1.0|
|              Hainan|         China|               1.0|
|              Shanxi|         China|              null|
|          Queensland|     Australia|0.8571428571428571|
|Turks and Caicos ...|United Kingdom|0.8235294117647058|
|Falkland Islands ...|United Ki[...]
{'fraction_non_alphanumeric': 0.21770299926847111, 'fraction_numerical': 0.07995610826627651, 'mean_word_length': 2.964907192575406, 'pattern_counts': {'":': 3, '<': 40, '<?xml version=': 0, '>': 44, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '427237', 'license': 'MIT', 'n_tokens_mistral': 5282, 'n_tokens_neox': 4585, 'n_words': 928, 'path': 'CS246_Colab_9.ipynb', 'repo_name': 'x3pi/mining-massive-datasets', 'size': '13711'}