<jupyter_start><jupyter_text>Solutions - Problem 2Get number of flights departed from each of the US state in the month of 2008 January.<jupyter_code>%%HTML
<iframe width="560" height="315" src="https://www.youtube.com/embed/HJcR4TDkutg?rel=0&amp;controls=1&amp;showinfo=0" frameborder="0" allowfullscreen></iframe><jupyter_output><empty_output><jupyter_text>* We have to use airport codes to determine state of each of the US airport.* We need to use airtraffic data to get departure details.* To solve this problem we have to perform inner join. Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our [10 node state of the art cluster/labs](https://labs.itversity.com/plans) to learn Spark SQL using our unique integrated LMS.<jupyter_code>from pyspark.sql import SparkSession

import getpass
username = getpass.getuser()

spark = SparkSession. \
    builder. \
    config('spark.ui.port', '0'). \
    config("spark.sql.warehouse.dir", f"/user/{username}/warehouse"). \
    enableHiveSupport(). \
    appName(f'{username} | Python - Joining Data Sets'). \
    master('yarn'). \
    getOrCreate()<jupyter_output><empty_output><jupyter_text>If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.**Using Spark SQL**```spark2-sql \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Scala**```spark2-shell \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Pyspark**```pyspark2 \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```<jupyter_code>spark.conf.set("spark.sql.shuffle.partitions", "2")
airtrafficPath = "/public/airtraffic_all/airtraffic-part/flightmonth=200801"
airtraffic = spark. \
    read. \
    parquet(airtrafficPath)
airtraffic. \
    select(
        "Year", "Month", "DayOfMonth", 
        "Origin", "Dest", "CRSDepTime"
    ). \
    show()
airtraffic.count()
airportCodesPath = "/public/airtraffic_all/airport-codes"
def getValidAirportCodes(airportCodesPath):
    airportCodes = spark. \
        read. \
        option("sep", "\t"). \
        option("header", True). \
        option("inferSchema", True). \
        csv(airportCodesPath). \
        filter("!(State = 'Hawaii' AND IATA = 'Big') AND Country = 'USA'")
    return airportCodes
airportCodes = getValidAirportCodes(airportCodesPath)
airportCodes.count()
from pyspark.sql.functions import col, lit, count
airtraffic. \
    join(airportCodes, col("IATA") == col("Origin"), "inner"). \
    groupBy("State"). \
    agg(count(lit(1)).alias("FlightCount")). \
    orderBy(col("FlightCount").desc()). \
    show()<jupyter_output>+-----+-----------+
|State|FlightCount|
+-----+-----------+
|   CA|      72853|
|   TX|      63930|
|   FL|      41042|
|   IL|      39812|
|   GA|      35527|
|   NY|      28414|
|   CO|      23288|
|   AZ|      20768|
|   OH|      19209|
|   NC|      17942|
|   MI|      17824|
|   NV|      17763|
| null|      14090|
|   TN|      13549|
|   PA|      13491|
|   UT|      12709|
|   NJ|      12498|
|   MN|      12357|
|   MO|      11808|
|   WA|      10210|
+-----+-----------+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.16356423558132438, 'fraction_numerical': 0.04150137320720171, 'mean_word_length': 2.911694510739857, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 14, 'https://': 2, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '566704', 'license': 'MIT', 'n_tokens_mistral': 1215, 'n_tokens_neox': 1113, 'n_words': 264, 'path': '06_joining_data_sets/09_solutions_problem_2.ipynb', 'repo_name': 'itversity/pyspark', 'size': '3277'}