<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 2.1. Google Cloud Storage (CSV) & Spark DataFrames - Python
Create Dataproc Cluster with Jupyter
This notebook is designed to be run on Google Cloud Dataproc.
Follow the links below for instructions on how to create a Dataproc Cluster with the Juypter component installed.
Tutorial - Install and run a Jupyter notebook on a Dataproc cluster
Blog post - Apache Spark and Jupyter Notebooks made easy with Dataproc component gateway
Python 3 Kernel
Use a Python 3 kernel (not PySpark) to allow you to configure the SparkSession in the notebook.
Create Spark Session
Step1:  Enable repl.eagerEval
This will output the results of DataFrames in each step without the need to use df.show() and also improves the formatting of the output
Step2:  List files in a GCS bucket
List files in a Google Cloud Storage bucket using the google-cloud-storage python library which comes installed on Dataproc clusters. We will be using a publicly available dataset.
Step3:  Alternatively use the hdfs cmd to list files in a directory which supports GCS buckets
Step4:  Read CSV files from GCS into Spark Dataframe
Read CSV files from GCS into a dataframe and infer the schema
Step5:  If there is no header with column names as we can see with the dataset here or the schema is not infered correctly then read CSV files from GCS and define schema
Step6:  View the top 20 rows of the spark dataframe
Step7:  Print the shape of the dataframe. No of rows and no of columns
Step8:  Add hour column and filter the data to create a new dataframe with only 1 day of data
Step9:  Group by hour and order by top_bids
Step10:  Write Spark Dataframe to Google Cloud Storage in CSV format
Write the Spark Dataframe to Google Cloud Storage using 
If the GCS bucket  does not exist it will need to be created before running df.write
Instructions here for creating a GCS bucket
Step11:  Read the CSV file into new DataFrame to check it was successfuly saved
<ASSISTANT_TASK:>
Python Code:
from pyspark.sql import SparkSession
spark = SparkSession.builder \
  .appName('2.1. Google Cloud Storage (CSV) & Spark DataFrames') \
  .getOrCreate()
Explanation: 2.1. Google Cloud Storage (CSV) & Spark DataFrames - Python
Create Dataproc Cluster with Jupyter
This notebook is designed to be run on Google Cloud Dataproc.
Follow the links below for instructions on how to create a Dataproc Cluster with the Juypter component installed.
Tutorial - Install and run a Jupyter notebook on a Dataproc cluster
Blog post - Apache Spark and Jupyter Notebooks made easy with Dataproc component gateway
Python 3 Kernel
Use a Python 3 kernel (not PySpark) to allow you to configure the SparkSession in the notebook.
Create Spark Session
End of explanation
spark.conf.set("spark.sql.repl.eagerEval.enabled",True)
Explanation: Enable repl.eagerEval
This will output the results of DataFrames in each step without the need to use df.show() and also improves the formatting of the output
End of explanation
from google.cloud import storage
gcs_client = storage.Client()
bucket = gcs_client.bucket('solutions-public-assets')
list(bucket.list_blobs(prefix='time-series-master/'))
Explanation: List files in a GCS bucket
List files in a Google Cloud Storage bucket using the google-cloud-storage python library which comes installed on Dataproc clusters. We will be using a publicly available dataset.
End of explanation
!hdfs dfs -ls 'gs://solutions-public-assets/time-series-master'
Explanation: Alternatively use the hdfs cmd to list files in a directory which supports GCS buckets
End of explanation
df1 = spark \
  .read \
  .option ( "inferSchema" , "true" ) \
  .option ( "header" , "true" ) \
  .csv ( "gs://solutions-public-assets/time-series-master/GBPUSD_*.csv" )
df1.printSchema()
df1
Explanation: Read CSV files from GCS into Spark Dataframe
Read CSV files from GCS into a dataframe and infer the schema
End of explanation
from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import DoubleType, IntegerType, StringType, TimestampType, DateType
schema = StructType([
    StructField("venue", StringType()),
    StructField("currencies", StringType()),
    StructField("time_stamp", TimestampType()),
    StructField("bid", DoubleType()),
    StructField("ask", DoubleType())
])
df2 = spark \
  .read \
  .schema(schema) \
  .csv ( "gs://solutions-public-assets/time-series-master/GBPUSD_*.csv" )
df2.printSchema()
Explanation: If there is no header with column names as we can see with the dataset here or the schema is not infered correctly then read CSV files from GCS and define schema
End of explanation
df2
Explanation: View the top 20 rows of the spark dataframe
End of explanation
print((df2.count(), len(df2.columns)))
Explanation: Print the shape of the dataframe. No of rows and no of columns
End of explanation
import pyspark.sql.functions as F
df3 = df2.withColumn("hour", F.hour(F.col("time_stamp"))) \
  .filter(df2['time_stamp'] >= F.lit('2014-01-01 00:00:00')) \
  .filter(df2['time_stamp'] < F.lit('2014-01-02 00:00:10')).cache()
df3
print((df3.count(), len(df3.columns)))
Explanation: Add hour column and filter the data to create a new dataframe with only 1 day of data
End of explanation
import pyspark.sql.functions as F
df4 = df3 \
.groupBy("hour") \
.agg(F.sum('bid').alias('total_bids'))
df4.orderBy('total_bids', ascending=False)
Explanation: Group by hour and order by top_bids
End of explanation
# Update to your GCS bucket
gcs_bucket = 'dataproc-bucket-name'
gcs_filepath = 'gs://{}/currency/hourly_bids.csv'.format(gcs_bucket)
df4.coalesce(1).write \
  .mode('overwrite') \
  .csv(gcs_filepath)
Explanation: Write Spark Dataframe to Google Cloud Storage in CSV format
Write the Spark Dataframe to Google Cloud Storage using 
If the GCS bucket  does not exist it will need to be created before running df.write
Instructions here for creating a GCS bucket
End of explanation
!hdfs dfs -ls gs://dataproc-bucket-name/currency
df5 = spark.read \
  .option ( "inferSchema" , "true" ) \
  .option ( "header" , "true" ) \
  .csv('gs://dataproc-bucket-name/currency/*')
df5
Explanation: Read the CSV file into new DataFrame to check it was successfuly saved
End of explanation
<END_TASK>