!scala -version

"""
Explanation: 1.1. BigQuery Storage & 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 and include the spark-bigquery-connector required to use the BigQuery Storage API.
Scala Version
Check what version of Scala you are running so you can include the correct spark-bigquery-connector jar
End of explanation
"""


from pyspark.sql import SparkSession
spark = SparkSession.builder \
  .appName('1.1. BigQuery Storage & Spark DataFrames - Python')\
  .config('spark.jars', 'gs://spark-lib/bigquery/spark-bigquery-latest.jar') \
  .getOrCreate()

"""
Explanation: Create Spark Session
Include the correct version of the spark-bigquery-connector jar
Scala version 2.11 - 'gs://spark-lib/bigquery/spark-bigquery-latest.jar'.
Scala version 2.12 - 'gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar'.
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 new need to show df.show() and also improves the formatting of the output
End of explanation
"""


table = "bigquery-public-data.wikipedia.pageviews_2020"
df_wiki_pageviews = spark.read \
  .format("bigquery") \
  .option("table", table) \
  .option("filter", "datehour >= '2020-03-01' AND datehour < '2020-03-02'") \
  .load()

df_wiki_pageviews.printSchema()

"""
Explanation: Read BigQuery table into Spark DataFrame
Use filter() to query data from a partitioned table.
End of explanation
"""


df_wiki_en = df_wiki_pageviews \
  .select("title", "wiki", "views") \
  .where("views > 1000 AND wiki in ('en', 'en.m')") \
  .cache()

df_wiki_en

"""
Explanation: Select required columns and apply a filter using where() which is an alias for filter() then cache the table
End of explanation
"""


import pyspark.sql.functions as F

df_wiki_en_totals = df_wiki_en \
.groupBy("title") \
.agg(F.sum('views').alias('total_views'))

df_wiki_en_totals.orderBy('total_views', ascending=False)

"""
Explanation: Group by title and order by page views to see the top pages
End of explanation
"""


# Update to your GCS bucket
gcs_bucket = 'dataproc-bucket-name'

# Update to your BigQuery dataset name you created
bq_dataset = 'dataset_name'

# Enter BigQuery table name you want to create or overwite. 
# If the table does not exist it will be created when you run the write function
bq_table = 'wiki_total_pageviews'

df_wiki_en_totals.write \
  .format("bigquery") \
  .option("table","{}.{}".format(bq_dataset, bq_table)) \
  .option("temporaryGcsBucket", gcs_bucket) \
  .mode('overwrite') \
  .save()

"""
Explanation: Write Spark Dataframe to BigQuery table
Write the Spark Dataframe to BigQuery table using BigQuery Storage connector. This will also create the table if it does not exist. The GCS bucket and BigQuery dataset must already exist.
If the GCS bucket and BigQuery dataset do not exist they will need to be created before running df.write

Instructions here for creating a GCS bucket
Instructions here for creating a BigQuery Dataset
End of explanation
"""


%%bigquery
SELECT title, total_views
FROM dataset_name.wiki_total_pageviews
ORDER BY total_views DESC
LIMIT 10

"""
Explanation: Use BigQuery magic to query table
Use the BigQuery magic to check if the data was created successfully in BigQuery. This will run the SQL query in BigQuery and the return the results
End of explanation
"""


