Leverage your SQL skills in PySpark to handle big data.

PySpark SQL

It's time to start performing some analysis–this is where PySpark SQL really shines. PySpark SQL DataFrames have a variety of built-in methods that can help with analyzing data. Let's get into a few examples!

Imagine we'd like to filter our data to pages from a specific Wikipedia `language_code` (e.g., `"kw.m"`). This site is not very active, so it's easy to use all of this hour's data for demonstration purposes. We can display this result with the code below:
    
```python
hrly_views_df\
    .filter(hrly_views_df.language_code == "kw.m")\
    .show(truncate=False)
```

```
+-------------+-----------------------+------------+-------------------+
|language_code|article_title          |hourly_count|monthly_count|
+-------------+-----------------------+------------+-------------------+
|kw.m         |Bresel_Diabarth_Spayn  |1           |0                  |
|kw.m         |Can_an_Pescador_Kernûak|1           |0                  |
|kw.m         |Ferdinand_Magellan     |1           |0                  |
|kw.m         |Justė_Arlauskaitė      |16          |0                  |
|kw.m         |Lithouani              |2           |0                  |
|kw.m         |Nolwenn_Leroy          |1           |0                  |
|kw.m         |Ohio                   |1           |0                  |
|kw.m         |Taywan                 |1           |0                  |
+-------------+-----------------------+------------+-------------------+
```

This code uses the `DataFrame.filter()` method to select relevant rows. This is analogous to a SQL "WHERE" clause. In this case, our condition checks the column `language_code` for the value `"kw.m"`. What if we want to remove the `monthly_count` column and display the data ordered by the `hourly_count`? To do so, we could use the following: 

```python
hrly_views_df\
    .filter(hrly_views_df.language_code == "kw.m")\
    .select(['language_code', 'article_title', 'hourly_count'])\
    .orderBy('hourly_count', ascending=False)\    
    .show(5, truncate=False)
```

```markdown
+-------------+-----------------------+------------+-------------------+
|language_code|article_title          |hourly_count|total_monthly_count|
+-------------+-----------------------+------------+-------------------+
|kw.m         |Justė_Arlauskaitė      |16          |0                  |
|kw.m         |Lithouani              |2           |0                  |
|kw.m         |Bresel_Diabarth_Spayn  |1           |0                  |
|kw.m         |Can_an_Pescador_Kernûak|1           |0                  |
|kw.m         |Nolwenn_Leroy          |1           |0                  |
+-------------+-----------------------+------------+-------------------+
```

- `DataFrame.select()` is used to choose which columns to return in our result. You can think of `DataFrame.select(["A", "B", "C"])` as analogous to `SELECT A, B, C FROM DataFrame` in SQL.

- `DataFrame.orderBy()` is analogous to SQL's `ORDER BY`. We use `.orderBy('hourly_count', ascending=False)` to specify the sort column and order logic. This would be analogous to `ORDER BY hourly_count DESC` in SQL.

What if we'd like to select the sum of `hourly_count` by `language_code`? This could help us answer questions like "Which sites were most active this hour?" We can do that with the following:

```python
hrly_views_df\
    .select(['language_code', 'hourly_count'])\
    .groupBy('language_code')\
    .sum() \
    .orderBy('sum(hourly_count)', ascending=False)\
    .show(5, truncate=False)
```

```markdown
+-------------+-----------------+
|language_code|sum(hourly_count)|
+-------------+-----------------+
|en.m         |8095763          |
|en           |2693185          |
|de.m         |1313505          |
|es.m         |963835           |
|ru.m         |927583           |
+-------------+-----------------+
```

This code uses `DataFrame.groupBy('language_code').sum()` to calculate the sum of all columns grouped by `language_code`, `.groupBy(field)` and `.sum()` are analogous to SQL's `GROUP BY` and `SUM` functions respectively. This code also orders our results with `.orderBy()`, using the name of the constructed column, `'sum(hourly_count)'`.

There are many ways to use the DataFrame methods to query our data. However, if you're familiar with SQL, you may prefer to use standard SQL statements. In the next section, we'll explore how you can use standard SQL to explore data with PySpark.


Querying PySpark DataFrames

The Spark ecosystem can be quite expansive, but the skills you've gained from this lesson should help you as you begin to branch out and run your own analyses. In this lesson you've learned:

- How to construct Spark DataFrames from raw data in Python and Spark RDDs.

- How to read and write data from disk into Spark DataFrames, including an introduction to file formats optimized for big-data workloads.

- How to perform data exploration and cleaning on distributed data.

- How the PySpark SQL API can allow you to perform analysis on distributed data more easily than working directly with RDDs by using DataFrames.

- How to use the PySpark SQL API to query your datasets with standard SQL.


Concluding Remarks

In this exercise, we're going to start to analyze our pageview data and learn how Spark can help with data exploration. Like Pandas, Spark DataFrames offer a series of operations for cleaning, inspecting, and transforming data. Earlier in the lesson, we mentioned that all DataFrames have a schema that defines their structure, columns, and datatypes. We can use `DataFrame.printSchema()` to show a DataFrame's schema.

```python
# Display DataFrame schema
hrly_views_df.printSchema()
```
```
root
|-- language_code: string (nullable = true)
|-- article_title: string (nullable = true)
|-- hourly_count: integer (nullable = true)
|-- monthly_count: integer (nullable = true)
```

We can then use `DataFrame.describe()` to see a high-level summary of the data by column. The result of `DataFrame.describe()` is a DataFrame in itself, so we append `.show()` to get it to display in our notebook. 

```python
hrly_views_df_desc = hrly_views_df.describe()
hrly_views_df_desc.show(truncate=False)
```

```
+-------+-------------+-------------+------------+-------------+
|summary|language_code|article_title|hourly_count|monthly_count|
+-------+-------------+-------------+------------+-------------+
|  count|      4654091|      4654091|     4654091|      4654091|
|   mean|         null|         null|     4.52417|          0.0|
| stddev|         null|         null|   182.92502|          0.0|
|    min|           aa|            -|           1|            0|
|    max|       zu.m.d|            -|      288886|            0|
+-------+-------------+-------------+------------+-------------+
```

From this summary, we can see a few interesting facts.
- About 4.65  million unique pages were visited this hour
- The most visited page had almost 289,000 visitors, while the mean page had just over 4.5 visitors.

Because this data was taken from the first hour of the month, it looks like the column `monthly_count` only contains zeros. Because it contains no meaningful information, we can drop this field with `DataFrame.drop("columns", "to", "drop")`.

```python
# Drop `monthly_count` and display new DataFrame
hrly_views_df = hrly_views_df.drop('monthly_count')
hrly_views_df.show(5)    
```

```
+-------------+---------------------------+------------+
|language_code|article_title              |hourly_count|
+-------------+---------------------------+------------+
|en           |Cividade_de_Terroso        |           2|
|en           |Peel_Session_(Autechre_EP) |           2|
|en           |Young_Street_Bridge        |           1|
|en           |Troy,_Alabama              |           1|
|en           |Charlotte_Johnson_Wahl     |          10|
+-------------+---------------------------+------------+
```

The data is starting to look pretty good, but let's make one more adjustment. The column `article_title` is a bit misleading: it seems this data contains articles, files, image pages, and wikipedia metadata pages.  We can replace this misleading header with a better name using `DataFrame.withColumnRenamed()`.

```python
hrly_views_df = hrly_views_df\
.withColumnRenamed('article_title', 'page_title')
``` 

Now when we call `.printSchema()` we see that the schema reflects the updates we've made to the DataFrame.

```
root
|-- language_code: string (nullable = true)
|-- page_title: string (nullable = true)
|-- hourly_count: integer (nullable = true)
```

You may have noticed that Spark assigned all columns `nullable = true`. Intuitively, we know that `article_title` shouldn't be null, but when the DataFrameReader reads a CSV, it assigns `nullable = true` to all columns. This is fine for now, but in some scenarios, you may wish to explicitly define a file's schema. If interested, you can refer to [PySpark's documentation on defining a file's schema.](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.schema.html)


Inspecting and Cleaning Data With PySpark

In this exercise, we'll learn how to pull in larger datasets from external sources. To start, we'll be using a dataset from Wikipedia that counts views of all articles by hour. For demonstration's sake, we'll use the first hour of 2022. Let's take a look at the code we might use to read a CSV of this data from a location on disk.

```python
print(type(spark.read)) 
# <class 'pyspark.sql.readwriter.DataFrameReader'>

# Read CSV to DataFrame
hrly_views_df = spark.read\
.option('header', True) \
.option('delimiter', ' ') \
.option('inferSchema', True)\ 
.csv('views_2022_01_01_000000.csv')
```
There are a few things going on in this code, let's go through them one at a time:

This code uses the `SparkSession.read` function to create a new `DataFrameReader` 

The `DataFrameReader` has an `.option('option_name', 'option_value')` method that can be used to instruct Spark how exactly to read a file. In this case, we used the following options: 

- `.option('header', True)` — Indicate the file already contains a header row. By default, Spark assumes there is no header.

- `.option('delimiter', ' ')` — Indicates each column is separated by a space (' '). By default, Spark assumes CSV columns are separated by commas.

- `.option('inferSchema', True)` — Instructs Spark to sample a subset of rows before determining each column's type. By default, Spark will treat all CSV columns as strings.

The `DataFrameReader` also has a `.csv('path')` method which loads a CSV file and returns the result as a DataFrame. There are a few quick ways of checking that our data has been read in properly. The most direct way is checking `DataFrame.show()`.

```python
# Display first 5 rows of DataFrame
hrly_views_df.show(5, truncate=False)
```

```
+--------+---------------------------+------+-------+
|language|article_title              |hourly|monthly|
+--------+---------------------------+------+-------+
|en      |Cividade_de_Terroso        |2     |0      |
|en      |Peel_Session_(Autechre_EP) |2     |0      |
|en      |Young_Street_Bridge        |1     |0      |
|en      |Troy,_Alabama              |1     |0      |
|en      |Charlotte_Johnson_Wahl     |10    |0      |
+--------+---------------------------+------+-------+
```

Looks Good! In this exercise, we used a `DataFrameReader` to pull a CSV from disk into our local Spark environment. However, Spark can read a wide variety of file formats. You can refer to the [PySpark documentation](https://spark.apache.org/docs/latest/api/python/search.html?q=DataFrameReader) to explore all available `DataFrameReader` options and file formats. In the following exercise, we'll start to analyze the contents of this file.


Spark DataFrames from External Sources

While we can directly analyze data using Spark's Resilient Distributed Datasets (RDDs), we may not always want to perform complicated analysis directly on RDDs. Luckily, Spark offers a module called Spark SQL that can make common data analysis tasks simpler and faster. In this lesson, we'll introduce Spark SQL and demonstrate how it can be a powerful tool for accelerating the analysis of distributed datasets.

The name Spark SQL is an umbrella term, as there are several ways to interact with data when using this module. We'll cover two of these methods using the PySpark API:

* First, we'll learn the basics of inspecting and querying data in a Spark DataFrame.

* Then, we'll perform these same operations using standard SQL directly in our PySpark code. 

Before using either method, we must start a `SparkSession`, the entry point to Spark SQL. The session is a wrapper around a `sparkContext` and contains all the metadata required to start working with distributed data. 

The code below uses `SparkSession.builder` to set configuration parameters and create a new session. In the following example, we set one configuration parameter (`spark.app.name`) and call the `.getOrCreate()` method to initialize the new `SparkSession`.

```python
spark = SparkSession.builder\
    .config('spark.app.name', 'learning_spark_sql')\
    .getOrCreate()
```

We can access the `SparkContext` for a session with `SparkSession.sparkContext`.
    
```python
print(spark.sparkContext) 
# <SparkContext master=local[*] appName=learning_spark_sql>
```

From here, we can use the `SparkSession` to create DataFrames, read external files, register tables, and run SQL queries over saved data. When we're done with our analysis, we can clear the Spark cache and terminate the session with `SparkSession.stop()`. Now that we're familiar with the basics of `SparkSession`, the next step is to begin using Spark SQL to interact with data!

---

_How to Use Your Jupyter Notebook:_
* _You can run a cell in the Notebook to the right by placing your cursor in the cell and clicking the `Run` button or the `Shift`+`Enter/Return` keys._
* _When you are ready to evaluate the code in your Notebook, press the `Save` button at the top of the Notebook or `command`+`s` keys before clicking the `Test Work` button at the bottom. **Be sure to save your solution code in the cell marked `## YOUR SOLUTION HERE ##` or it will not be evaluated.**_
* _When you are ready to move on, click `Next`._

![screenshot of the buttons at the top of the Jupyter Notebook interface with Save and Run highlighted](https://static-assets.codecademy.com/Courses/big-data-pyspark/Jupyter-buttons.png)

Introducing PySpark SQL

A PySpark SQL DataFrame is a distributed collection of data with a specific row and column structure. Under the hood, DataFrames are built on top of RDDs. Like pandas, PySpark SQL DataFrames allow a developer to analyze data more easily than by writing functions directly on underlying data.

DataFrames can be created manually from RDDs using `rdd.toDF(["names", "of", "columns"])`. In the example below, we create a DataFrame from a manually constructed RDD and name its columns `article_title` and `view_count`.
    
```python
# Create an RDD from a list
hrly_views_rdd  = spark.sparkContext.parallelize([
    ["Betty_White" , 288886],
    ["Main_Page", 139564],
    ["New_Year's_Day", 7892],
    ["ABBA", 8154]
])

# Convert RDD to DataFrame
hrly_views_df = hrly_views_rdd\
    .toDF(["article_title", "view_count"])
```

Let's take a look at our new DataFrame. We can use the `DataFrame.show(n_rows)` method to print the first `n_rows` of a Spark DataFrame. It can also be helpful to pass `truncate=False` to ensure all columns are visible.

```python
hrly_views_df.show(4, truncate=False)
```

```markdown
+--------------+-----------+
| article_title| view_count|
+--------------+-----------+
|   Betty_White|     288886|
|     Main_Page|     139564|
|New_Year's_Day|       7892|
|          ABBA|       8154|
+--------------+-----------+
```

Great! Now that this data is loaded in as a DataFrame, we can access the underlying RDD with `DataFrame.rdd`. You likely won't need the underlying data often, but it can be helpful to keep in mind that a DataFrame is a structure built on top of an RDD. When we check the type of `hrly_views_df_rdd`, we can see that it's an RDD!

```python
# Access DataFrame's underlying RDD
hrly_views_df_rdd = hrly_views_df.rdd

# Check object type
print(type(hrly_views_df_rdd)) 
# <class 'pyspark.rdd.RDD'>
```



Creating Spark DataFrames

Once you've done some analysis, the next step is often saving the transformed data back to disk for others to use. In this final topic, we're going to cover how to efficiently save PySpark DataFrames.

Similar to the `SparkSession.read()` method, Spark offers a `SparkSession.write()` method. Let's perform a slight modification to our original Wikipedia views dataset and save it to disk. This code just uses `.select()` to select all columns except the `monthly_count` column (recall that earlier we discovered this column only contains zeros).

Because Spark runs all operations in parallel, it's typical to write DataFrames to a directory of files rather than a single CSV file. In the example below, Spark will split the underlying dataset and write multiple CSV files to `cleaned/csv/views_2022_01_01_000000/`. We can also use the `mode` argument of the `.csv()` method to overwrite any existing data in the target directory.

```python
hrly_views_df\
    .select(['language_code', 'article_title', 'hourly_count'])\
    .write.csv('cleaned/csv/views_2022_01_01_000000/', mode="overwrite")
```

Using `SparkSession.read()`, we can read the data from disk and confirm that it looks the same as the DataFrame we saved.

```python
# Read DataFrame back from disk
hrly_views_df_restored = spark.read\
    .csv('cleaned/csv/views_2022_01_01_000000/')
hrly_views_df_restored.printSchema()
```
```
root
|-- _c0: string (nullable = true)
|-- _c1: string (nullable = true)
|-- _c2: string (nullable = true)
```

Close, but not quite! It looks like this file didn't retain information about column headers or datatypes. Unfortunately, there's no way for a CSV to retain information about its format. Each time we read it, we'll need to tell Spark exactly how it must be processed. 

Luckily, there is a file format called "Parquet" that's specially designed for big data and solves this problem among many others. Parquet offers efficient data compression, is faster to perform analysis on than CSV, and preserves information about a dataset's schema. Let's try saving and re-reading this file to and from Parquet instead.

```python
# Write DataFrame to Parquet
hrly_views_slim_df
    .write.parquet('cleaned/parquet/views_2022_01_01_000000/', mode="overwrite")

# Read Parquet as DataFrame
hrly_views_df_restored = spark.read\
    .parquet('cleaned/parquet/views_2022_01_01_000000/')

# Check DataFrame's schema
hrly_views_df_restored.printSchema()
```
```
root
|-- language_code: string (nullable = true)
|-- article_title: string (nullable = true)
|-- hourly_count: integer (nullable = true)
```

Great, now anyone who wants to query this data can do so with the much more efficient Parquet data format!

Saving PySpark DataFrames

PySpark DataFrame's query methods are an improvement on performing analysis directly on RDDs. However, working with DataFrame methods still requires some practice, and the code can become quite verbose. Luckily, we can analyze data in Spark with standard SQL through the `SparkSession.sql()` method. This exercise will closely mirror the previous one, and we'll answer the same questions from that exercise using standard SQL. 

Before querying a DataFrame with SQL in Spark, it must be saved to the SparkSession's catalog. The following code saves the DataFrame as a local temporary view in memory. As long as the current `SparkSession` is active, we can use `SparkSession.sql()` to query it.

```python
hrly_views_df.createOrReplaceTempView('hourly_counts')
```

Each of the three sections of SQL below performs the same function as the DataFrame query methods described in the previous exercise. With the query below, we can filter our data to pages from a specific Wikipedia `language_code` (e.g., `"kw.m"`) using a `WHERE` clause.

```python
query = """SELECT * FROM hourly_counts WHERE language_code = 'kw.m'"""
spark.sql(query).show(truncate=False)
```
```
+-------------+-----------------------+------------+-------------+
|language_code|article_title          |hourly_count|monthly_count|
+-------------+-----------------------+------------+-------------+
|kw.m         |Bresel_Diabarth_Spayn  |           1|            0|
|kw.m         |Can_an_Pescador_Kernûak|           1|            0|
|kw.m         |Ferdinand_Magellan     |           1|            0|
|kw.m         |Justė_Arlauskaitė      |          16|            0|
|kw.m         |Lithouani              |           2|            0|
|kw.m         |Nolwenn_Leroy          |           1|            0|
|kw.m         |Ohio                   |           1|            0|
|kw.m         |Taywan                 |           1|            0|
+-------------+-----------------------+------------+-------------+
```

In the query below, we display all pages with `"kw.m"` as their `language_code` ordered by the `hourly_count` using an `ORDER BY` clause.

```python
query = """SELECT language_code, article_title, hourly_count
    FROM hourly_counts
    WHERE language_code = 'kw.m'
    ORDER BY hourly_count DESC"""

spark.sql(query).show(truncate=False)
```

```
+-------------+-----------------------+------------+-------------------+
|language_code|article_title          |hourly_count|total_monthly_count|
+-------------+-----------------------+------------+-------------------+
|kw.m         |Justė_Arlauskaitė      |          16|                  0|
|kw.m         |Lithouani              |           2|                  0|
|kw.m         |Bresel_Diabarth_Spayn  |           1|                  0|
|kw.m         |Can_an_Pescador_Kernûak|           1|                  0|
|kw.m         |Nolwenn_Leroy          |           1|                  0|
+-------------+-----------------------+------------+-------------------+
```

Finally, we select the sum of `hourly_count` by `language_code` over the entire DataFrame using a SQL statement with `GROUP BY`, `SUM`, and  `ORDER BY`.

```python
query = """SELECT language_code, SUM(hourly_count) as sum_hourly_count
    FROM hourly_counts
    GROUP BY language_code
    ORDER BY sum_hourly_count DESC"""

spark.sql(query).show(5, truncate=False)
```

```
+-------------+-----------------+
|language_code|sum(hourly_count)|
+-------------+-----------------+
|en.m         |8095763          |
|en           |2693185          |
|de.m         |1313505          |
|es.m         |963835           |
|ru.m         |927583           |
+-------------+-----------------+
```

Although querying data with SQL and DataFrame methods may look quite different, behind the scenes, Spark SQL translates everything to the same internal code. This means that as a developer, you can focus more on writing code for analysis in your preferred style rather than low-level execution details.
