<jupyter_start><jupyter_text>Exploring Data with DataFrames and Spark SQLIn this exercise, you will explore data using the Spark DataFrames API and Spark SQL. Load Data Using an Explicit SchemaTo explore data, you must load it into a programmatic data object such as a DataFrame. If the structure of the data is known ahead of time, you can explicitly specify the schema for the DataFrame.In this exercise, you will work with data that records details of flights.<jupyter_code>from pyspark.sql.types import *

flightSchema = StructType([
  StructField("DayofMonth", IntegerType(), False),
  StructField("DayOfWeek", IntegerType(), False),
  StructField("Carrier", StringType(), False),
  StructField("OriginAirportID", IntegerType(), False),
  StructField("DestAirportID", IntegerType(), False),
  StructField("DepDelay", IntegerType(), False),
  StructField("ArrDelay", IntegerType(), False),
])

wkdir ='file:///mnt/c/Users/Adura/Google Drive/Projects/Jupyter/SparkMs/data/'
flights = spark.read.csv(wkdir + "raw-flight-data.csv", schema=flightSchema, header=True)
flights.show(10)<jupyter_output>+----------+---------+-------+---------------+-------------+--------+--------+
|DayofMonth|DayOfWeek|Carrier|OriginAirportID|DestAirportID|DepDelay|ArrDelay|
+----------+---------+-------+---------------+-------------+--------+--------+
|        19|        5|     DL|          11433|        13303|      -3|       1|
|        19|        5|     DL|          14869|        12478|       0|      -8|
|        19|        5|     DL|          14057|        14869|      -4|     -15|
|        19|        5|     DL|          15016|        11433|      28|      24|
|        19|        5|     DL|          11193|        12892|      -6|     -11|
|        19|        5|     DL|          10397|        15016|      -1|     -19|
|        19|        5|     DL|          15016|        10397|       0|      -1|
|        19|        5|     DL|          10397|        14869|      15|      24|
|        19|        5|     DL|          10397|        10423|      33|      34|
|        19|        5|     DL|          11278|      [...]<jupyter_text>Infer a Data SchemaIf the structure of the data source is unknown, you can have Spark auomatically infer the schema.In this case, you will load data about airports without knowing the schema.<jupyter_code>airports = spark.read.csv(wkdir + "airports.csv", header=True, inferSchema=True)
airports.show()<jupyter_output>+----------+-----------+-----+--------------------+
|airport_id|       city|state|                name|
+----------+-----------+-----+--------------------+
|     10165|Adak Island|   AK|                Adak|
|     10299|  Anchorage|   AK|Ted Stevens Ancho...|
|     10304|      Aniak|   AK|       Aniak Airport|
|     10754|     Barrow|   AK|Wiley Post/Will R...|
|     10551|     Bethel|   AK|      Bethel Airport|
|     10926|    Cordova|   AK|Merle K Mudhole S...|
|     14709|  Deadhorse|   AK|   Deadhorse Airport|
|     11336| Dillingham|   AK|  Dillingham Airport|
|     11630|  Fairbanks|   AK|Fairbanks Interna...|
|     11997|   Gustavus|   AK|    Gustavus Airport|
|     12523|     Juneau|   AK|Juneau International|
|     12819|  Ketchikan|   AK|Ketchikan Interna...|
|     10245|King Salmon|   AK| King Salmon Airport|
|     10170|     Kodiak|   AK|      Kodiak Airport|
|     13970|   Kotzebue|   AK| <NAME>ien Memorial|
|     13873|       Nome|   AK|        Nome Airport|
|     14256|[...]<jupyter_text>Use DataFrame MethodsSpark DataFrames provide functions that you can use to extract and manipulate data. For example, you can use the **select** function to return a new DataFrame containing columns selected from an existing DataFrame.<jupyter_code>cities = airports.select("city", "name")
cities.show()<jupyter_output>+-----------+--------------------+
|       city|                name|
+-----------+--------------------+
|Adak Island|                Adak|
|  Anchorage|Ted Stevens Ancho...|
|      Aniak|       Aniak Airport|
|     Barrow|Wiley Post/Will R...|
|     Bethel|      Bethel Airport|
|    Cordova|Merle K Mudhole S...|
|  Deadhorse|   Deadhorse Airport|
| Dillingham|  Dillingham Airport|
|  Fairbanks|Fairbanks Interna...|
|   Gustavus|    Gustavus Airport|
|     Juneau|Juneau International|
|  Ketchikan|Ketchikan Interna...|
|King Salmon| King Salmon Airport|
|     Kodiak|      Kodiak Airport|
|   Kotzebue| Ralph Wien Memorial|
|       Nome|        Nome Airport|
| Petersburg|Petersburg James ...|
|      Sitka|Sitka Rocky Gutie...|
| St. Mary's|  St. Mary's Airport|
|   Unalaska|    Unalaska Airport|
+-----------+--------------------+
only showing top 20 rows<jupyter_text>Combine OperationsYou can combine functions in a single statement to perform multiple operations on a DataFrame. In this case, you will use the **join** function to combine the **flights** and **airports** DataFrames, and then use the **groupBy** and **count** functions to return the number of flights from each airport.<jupyter_code>flightsByOrigin = flights.join(airports, flights.OriginAirportID == airports.airport_id).groupBy("city").count()
flightsByOrigin.show()<jupyter_output>+-----------------+------+
|             city| count|
+-----------------+------+
|          Phoenix| 90281|
|            Omaha| 13537|
|   Raleigh/Durham| 28436|
|        Anchorage|  7777|
|           Dallas| 19503|
|          Oakland| 25503|
|      San Antonio| 23090|
|     Philadelphia| 47659|
|       Louisville| 10953|
|Dallas/Fort Worth|105024|
|      Los Angeles|118684|
|       Sacramento| 25193|
|     Indianapolis| 18099|
|        Cleveland| 25261|
|        San Diego| 45783|
|    San Francisco| 84675|
|        Nashville| 34927|
|    Oklahoma City| 13967|
|          Detroit| 62879|
|         Portland| 30640|
+-----------------+------+
only showing top 20 rows<jupyter_text>Count the Rows in a DataFrameNow that you're familiar with working with DataFrames, a key task when building predictive solutions is to explore the data, determing statistics that will help you understand the data before building predictive models. For example, how many rows of flight data do you actually have?<jupyter_code>flights.count()<jupyter_output><empty_output><jupyter_text>Determine Summary StatisticsPredictive modeling is based on statistics and probability, so you will often start by looking at summary statistics. The **describe** function returns a DataFrame containing the **count**, **mean**, **standard deviation**, **minimum**, and **maximum** values for each numeric column.<jupyter_code>flights.describe().show()<jupyter_output>+-------+-----------------+------------------+-------+-----------------+------------------+-----------------+-----------------+
|summary|       DayofMonth|         DayOfWeek|Carrier|  OriginAirportID|     DestAirportID|         DepDelay|         ArrDelay|
+-------+-----------------+------------------+-------+-----------------+------------------+-----------------+-----------------+
|  count|          2719418|           2719418|2719418|          2719418|           2719418|          2691974|          2690385|
|   mean|15.79747468024408|3.8983907586108497|   null|12742.26441172339|12742.455345592329|10.53686662649788| 6.63768791455498|
| stddev|8.799860168985413|1.9859881390373355|   null|1501.972939702572|1501.9692528927803|36.09952806643145|38.64881489390095|
|    min|                1|                 1|     9E|            10140|             10140|              -63|              -94|
|    max|               31|                 7|     YV|            15376|             15376|             [...]<jupyter_text>Determine the Presence of DuplicatesThe data you have to work with won't always be perfect - often you'll want to *clean* the data; for example to detect and remove duplicates that might affect your model. You can use the **dropDuplicates** function to create a new DataFrame with the duplicates removed, enabling you to determine how many rows are duplicates of other rows.<jupyter_code>flights.count() - flights.dropDuplicates().count()<jupyter_output><empty_output><jupyter_text>Identify Missing ValuesAs well as determing if duplicates exist in your data, you should detect missing values, and either remove rows containing missing data or replace the missing values with a suitable relacement. The **dropna** function creates a DataFrame with any rows containing missing data removed - you can specify a subset of columns, and whether the row should be removed in *any* or *all* values are missing. You can then use this new DataFrame to determine how many rows contain missing values.<jupyter_code>flights.count() - flights.dropDuplicates().dropna(how="any", subset=["ArrDelay", "DepDelay"]).count()<jupyter_output><empty_output><jupyter_text>Clean the DataNow that you've identified that there are duplicates and missing values, you can clean the data by removing the duplicates and replacing the missing values. The **fillna** function replaces missing values with a specified replacement value. In this case, you'll remove all duplicate rows and replace missing **ArrDelay** and **DepDelay** values with **0**.<jupyter_code>data=flights.dropDuplicates().fillna(value=0, subset=["ArrDelay", "DepDelay"])
data.count()<jupyter_output><empty_output><jupyter_text>Check Summary StatisticsAfter cleaning the data, you should re-check the statistics - removing rows and changing values may affect the distribution of the data, which in turn could affect any predictive models you might create.<jupyter_code>data.describe().show()<jupyter_output>+-------+------------------+------------------+-------+------------------+------------------+------------------+------------------+
|summary|        DayofMonth|         DayOfWeek|Carrier|   OriginAirportID|     DestAirportID|          DepDelay|          ArrDelay|
+-------+------------------+------------------+-------+------------------+------------------+------------------+------------------+
|  count|           2696983|           2696983|2696983|           2696983|           2696983|           2696983|           2696983|
|   mean|15.798996508320593| 3.900369412784582|   null|12742.459424846207| 12742.85937657004|10.531134234068217|6.6679285705545785|
| stddev| 8.801267199135454|1.9864582421701973|   null|1502.0359941370625|1501.9939589817989| 36.06172819056573|38.583861473580725|
|    min|                 1|                 1|     9E|             10140|             10140|               -63|               -94|
|    max|                31|                 7|     YV|             15376|  [...]<jupyter_text>Explore Relationships in the DataPredictive modeling is largely based on statistical relationships between fields in the data. To design a good model, you need to understand how the data points relate to one another and identify any apparent correlation. The **corr** function calculates a correlation value between -1 and 1, indicating the strength of correlation between two fields. A strong positive correlation (near 1) indicates that high values for one column are often found with high values for the other, which a string negative correlation (near -1) indicates that *low* values for one column are often found with *high* values for the other. A correlation near 0 indicates little apparent relationship between the fields.<jupyter_code>data.corr("DepDelay", "ArrDelay")<jupyter_output><empty_output><jupyter_text>Use Spark SQLIn addition to using the DataFrame API directly to query data, you can persist DataFrames as table and use Spark SQL to query them using the SQL language. SQL is often more intuitive to use when querying tabular data structures.<jupyter_code>data.createOrReplaceTempView("flightData")
spark.sql("SELECT DayOfWeek, AVG(ArrDelay) AS AvgDelay FROM flightData GROUP BY DayOfWeek ORDER BY DayOfWeek").show()<jupyter_output>+---------+------------------+
|DayOfWeek|          AvgDelay|
+---------+------------------+
|        1| 7.077989660973244|
|        2|  4.39237404158651|
|        3| 7.234625279280266|
|        4|10.775574715480056|
|        5|  8.71110560964396|
|        6|2.1437428120738304|
|        7|  5.25403935972552|
+---------+------------------+<jupyter_text>Use the Inline SQL *Magic*Jupyter Notebooks support *magics*, which enable you to include inline code and functionality. For example, the **%%sql** magic enables you to write SQL queries and visualize the results directly in the notebook.Run the following query, and view the table of results that is returned.<jupyter_code># %%sql
# SELECT DepDelay, ArrDelay FROM flightData
spark.sql("SELECT DepDelay, ArrDelay FROM flightData").show()<jupyter_output>+--------+--------+
|DepDelay|ArrDelay|
+--------+--------+
|       1|     -22|
|       0|       6|
|      -6|       3|
|     113|     117|
|      -3|     -11|
|      -3|       0|
|      -4|     -11|
|     -11|     -26|
|      -4|      -2|
|      -1|       3|
|     109|     106|
|       0|      -4|
|      -8|     -25|
|      -5|      -9|
|      22|      31|
|      -3|     -19|
|      13|       2|
|      -2|     -10|
|      -3|     -10|
|      -3|     -13|
+--------+--------+
only showing top 20 rows<jupyter_text>Change the **Table** visualization of results above to a **Scatter** visualization to see the relationship between the **DepDelay** and **ArrDelay** values more clearly (use the **-** function to plot the actual values) - visualizations like this make it easier to show relationships as apparent *structure* in the data. For example, the positive correlation between **DepDelay** and **ArrDelay** seems to be a linear relationsip, creaing a diagonal line of plotted points.  Query Multiple TablesYou can create and query multiple temporary tables. Run the cells below to create a temporary table from the **airports** DataFrame, and then use an inline query to query it together with the flights data.<jupyter_code>airports.createOrReplaceTempView("airportData")
spark.sql("SELECT a.name, AVG(f.ArrDelay) AS avgdelay FROM flightData AS f JOIN airportData AS a
ON f.DestAirportID = a.airport_id GROUP BY a.name").show() 
%%sql
SELECT a.name, AVG(f.ArrDelay) AS avgdelay
FROM flightData AS f JOIN airportData AS a
ON f.DestAirportID = a.airport_id
GROUP BY a.name<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16431231991004286, 'fraction_numerical': 0.07590132827324478, 'mean_word_length': 2.265259293253786, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '182311', 'license': 'MIT', 'n_tokens_mistral': 4791, 'n_tokens_neox': 4055, 'n_words': 1327, 'path': 'DataAnalyticsWithSpark/Data Exploration/Python Data Exploration.ipynb', 'repo_name': 'AduraX/Big-Data-Analytics', 'size': '14230'}