<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Initialize PySpark
First, we use the findspark package to initialize PySpark.
Step1:  Hello, World!
Loading data, mapping it and collecting the records into RAM...
Step2:  Creating Objects from CSV using pyspark.RDD.map
Using a function with a map operation to create objects (dicts) as records...
Step3:  pyspark.RDD.groupBy
Using the groupBy operator to count the number of jobs per person...
Step4:  Exercises
Use pyspark.RDD.groupBy to group executives by job title, then prepare records with the job title and the count of the number of executives with that job.
Map vs FlatMap
We need to understand the difference between the map and flatmap operators. Map groups items per-record, while flatMap creates a single large group of items.
Step5:  Creating Rows
We can create pyspark.sql.Rows out of python objects so we can create pyspark.sql.DataFrames. This is desirable because once we have DataFrames we can run Spark SQL on our data.
Step6:  Exercises
First count the number of companies for each executive, then create a pyspark.sql.Row for this result.
Step8:  Creating DataFrames from RDDs
Using the RDD.toDF() method to create a dataframe, registering the DataFrame as a temporary table with Spark SQL, and counting the jobs per person using Spark SQL.
Step9:  SparkContext.parallelize()
The oppotiste of pyspark.RDD.collect() is SparkContext.parallelize(). Whereas collect pulls data from Spark's memory into local RAM, parallelize sends data from local memory to Spark's memory. 
You can access it like this
Step11:  Exercises
Create your own RDD of dict elements with named fields using sc.parallelize. Make it at least 5 records long.
Convert this RDD of dicts into an RDD of pyspark.sql.Row elements.
Convert this RDD of pyspark.sql.Rows into a pyspark.sql.DataFrame.
Run a SQL GROUP BY/COUNT on your new DataFrame.
Step12:  Creating RDDs from DataFrames
We can easily convert back from a DataFrame to an RDD using the pyspark.sql.DataFrame.rdd() method, along with pyspark.sql.Row.asDict() if we desire a Python dict of our records.
Step13:  Exercises
Using the data from item 4 in the exercise above, convert the data back to its original form, a local collection of dict elements.
Step14:  Loading and Inspecting Parquet Files
Using the SparkSession to load files as DataFrames and inspecting their contents...
Step15:  DataFrame Workflow
Step16:  From Minutes to Hours
Now lets convert our AirTime from minutes to hours by dividing by 60.
Step17:  Raw Calculation
Now lets calculate miles per hour!
Step18:  Investigating nulls
Looks like we have some errors in some records in our calculation because of missing fields? Lets bring back in the Distance and AirTime fields to see where the problem is coming from.
Step19:  Filtering nulls
Now that we know some records are missing AirTimes, we can filter those records using pyspark.sql.DataFrame.filter(). Starting from the beginning, lets recalculate our values.
Step20:  Averaging Speed
How fast does the fleet travel overall? Lets compute the average speed for the entire fleet.
Step22:  It looks like the average speed of the fleet is 408 mph. Note how along the way we chekced the data for sanity, which led to confidence in our answer. SQL by contast can hide the internals of a query, which might have skewed our average significantly! 
SQL-Based Speed Calculation
Now lets work the same thing out in SQL. Starting from the top
Step23:  Evaluating SQL
The SQL based solution seems to be better in this case, because we can simply express our calculation all in one place. When complexity grows however, it is best to break a single query into multiple stages where you use SQL or Dataflow programming to massage the data into shape.
Calculating Histograms
Having computed the speed in miles per hour and the overall average speed of passenger jets in the US, lets dig deeper by using the RDD API's histogram method to calculate histograms buckets and values, which will then use to visualize data.
Step25:  Visualizing Histograms
The problem with the output above is that it is hard to interpret. For better understanding, we need a visualization. We can use matplotlib inline in a Jupyter Notebook to visualize this distribution and see what the tendency of speed of airplanes is around the mean of 408 mph.
Step26:  Iterating on a Histogram
That looks interesting, but the bars seem to fat the really see what is going on. Lets double the number of buckets from 10 to 20. We can reuse the create_hist() method to do so.
Step27:  Speed Summary
You've now seen how to calculate different values in both SQL and Dataflow style, how to switch between the two methods, how to switch between the pyspark.RDD and pyspark.sql.DataFrame APIs and you're starting to build a proficiency in PySpark!
Counting Airplanes in the US Fleet
Lets convert our on_time_dataframe (a DataFrame) into an RDD to calculate the total number of airplanes in the US fleet.
Step28:  Exercise 1
Step29:  Calculating with DataFrame.groupBy
We can use Spark SQL to calculate things using DataFrames, but we can also group data and calculate as we did with RDDs. For a full list of methods you can apply to grouped DataFrames, see the documentation for pyspark.sql.GroupedData. Below we will demonstrate some of these methods.
Step30:  Pivoting DataFrames
One useful function of DataFrames is pivot. Pivot lets you compute pivot tables from data. Lets use pivot to calculate the average flight times between Atlanta ATL and other airports.
Step32:  Plotting Scatterplots
Another type of visualization that is of interest to data scientists is the scatterplot. A scatterplot enables us to compare the trend of one value plotted against the other. For example, we could calculate the relationship between Origin and Dest Distance and the Mph speed figure we calculated earlier. Are longer flights generally faster, or not?
To prepare a scatterplot, we need to use matplotlib again, so we'll need to look at what its scatterplot API expects. The matplotlib.pyplot.scatter API takes two independant lists of values for the variables x and y, so we must compute them for Distance and Mph.
Step33:  Collecting Data
Note that we will have to convert our data from existing within our Spark cluster's memory to within our local computer's memory where matplotlib runs.
Step34:  Sampling Data
When I tried to plot this data, it took a very long time to draw. This is because... well, how many unique values are there for each variable? Lets see.
Step35:  It is hard to plot 5.7 million dots on a scatterplot that will fit on a computer screen. So lets sample our data. We can use PySpark DataFrame's sample method. Lets take a 0.1% random sample without replacement, which will leave us with 5,687 or so data points - something we can more easily manage.
Step36:  Note that we need to sample once and then split the datasets out - otherwise the data for a single observation will be scrambled across variables. We don't want that! All our scatterplots would show no relationships at all.
Step37:  Fun with matplotlib.pyplot.scatter
Now we feed the scatter API distance as x and speed as y, giving it a title and x and y axes. Note that we also specify a size in inches via the figure.figsize rcParam.
Step38:  Interpreting Our Scatterplot
We can see pretty clearly that as distance increases, average speed across that distance increases rapidly and then levels off as the distance increases.
Exercises
Query the on_time_dataframe to focus on two numeric fields. 
Plot a histogram of one of these fields
Plot a scatterplot of both of these fields
Predicting Speed Given Distance
It is often the case that once we characterize a distribution, we want to create a function to predict one variable given the other. Lets take this example further by fitting a polynomial regression to describe our data. We use sklearn.pipeline.Pipeline to chain a sklearn.preprocessing.PolynomialFeatures to a sklearn.linear_model.LinearRegression. Other than that, we simply define x and y, and fit a model to those values. Then we finally compute a cross value score, to see the model's performance. We'll see this pattern again when we use large data tools in Spark MLlib.
Step39:  Visualizing Polynomial Fit
Because we are running a polynomial regression, where we get to decide the degree of the polynomial. To help decide, lets plot a polynomial fit line to the data using matplotlib.
Step40:  Joining Data in PySpark
Next we're going to learn how to join between datasets using PySpark. We're going to pick up from an example we're going to work in chapter 6, and explore it more deeply. To begin with, we will prepare a list of TailNum (tail numbers) from the FAA flight records. These uniquely identify each airplane from each flight.
Unique Tail Numbers
Step41:  FAA Airplane Records
We will trim the FAA records down to just the TailNum, Model and Engine_Model. Note that we go ahead and rename the TailNum field to FAATailNum using the pyspark.sql.functions.alias() method. This avoids having two fields referenced by the same name once we perform our joins.
Step42:  Inner Joins
You may be familiar with an inner join from SQL. An inner join joins two datasets based on the presence of a key from one dataset in the other. Records which don't have a key that appears in the other table do not appear in the final output.
Step43:  Inner Join Results
Note that there are as many records in the output as there were in the FAA Airplane dataset - indicating that there was a representative of every tail number from that dataset in the on-time performance flight records. Lets take a look at the records themselves.
Step44:  Note how convenient it is that we renamed one of the keys FAATailNum. If we hadn't, we'd have two columns with the same name now and would have trouble referring to one or the other.
Left Outer Join
Another type of join is the left outer join. It ensures that one record will remain in the output from the left side of the join no matter what. If a match on the join keys is found, the fields for the record on the right will be filled. If a match is not found, they will be empty. 
Lets look at how this works with our two datasets.
Step45:  Left Outer Join Result
Note that there were 4,898 records on the left side of our join and there are the same number on the output of our join. Lets take a look at what both matched and unmatched records look like
Step46:  Note that some records have fields filled out, and some don't.
Right Outer Join
Another type of join is the right outer join. This works the opposite of a left outer join. In this case, the output will preserve a record for each and every record on the right side of the join. Use the right_outer join type to perform this kind of join.
Exercises
Go back and perform a right outer join on the preceding two datasets. Is the distinct() call on the FAA on-time performance records still needed? Why or why not?
Using RDDs and Map/Reduce to Prepare a Complex Record
Step47:  Counting Late Flights
Step48:  Counting Flights with Hero Captains
"Hero Captains" are those that depart late but make up time in the air and arrive on time or early.
Step49:  Printing Our Results
Step51:  Computing the Average Lateness Per Flights
Step53:  Inspecting Late Flights
Step55:  Determining Why Flights Are Late
Step56:  Computing a Histogram of Weather Delayed Flights
Step57:  Preparing a Histogram for Visualization by d3.js
Step58:  Building a Classifier Model to Predict Flight Delays
Loading Our Data
Step59:  Check Data for Nulls
Step60:  Add a Route Column
Demonstrating the addition of a feature to our model...
Step61:  Bucketizing ArrDelay into ArrDelayBucket
Step62:  Indexing Our String Fields into Numeric Fields
Step63:  Combining Numeric Fields into a Single Vector
Step64:  Training Our Model in an Experimental Setup
<ASSISTANT_TASK:>
Python Code:
from pyspark.sql import SparkSession
# Initialize PySpark with MongoDB support
APP_NAME = "Introducing PySpark"
spark = (
    SparkSession.builder.appName(APP_NAME)
    # Load support for MongoDB and Elasticsearch
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1,org.elasticsearch:elasticsearch-spark-30_2.12:7.14.2")
    # Add Configuration for MongopDB
    .config("spark.mongodb.input.uri", "mongodb://mongo:27017/test.coll")
    .config("spark.mongodb.output.uri", "mongodb://mongo:27017/test.coll")
    .getOrCreate()
)
sc = spark.sparkContext
sc.setLogLevel("ERROR")
print("\nPySpark initialized...")
Explanation: Initialize PySpark
First, we use the findspark package to initialize PySpark.
End of explanation
# Load the text file using the SparkContext
csv_lines = sc.textFile("../data/example.csv")
# Map the data to split the lines into a list
data = csv_lines.map(lambda line: line.split(","))
# Collect the dataset into local RAM
data.collect()
Explanation: Hello, World!
Loading data, mapping it and collecting the records into RAM...
End of explanation
# Turn the CSV lines into objects
def csv_to_record(line):
  parts = line.split(",")
  record = {
    "name": parts[0],
    "company": parts[1],
    "title": parts[2]
  }
  return record
# Apply the function to every record
records = csv_lines.map(csv_to_record)
# Inspect the first item in the dataset
records.first()
Explanation: Creating Objects from CSV using pyspark.RDD.map
Using a function with a map operation to create objects (dicts) as records...
End of explanation
# Group the records by the name of the person
grouped_records = records.groupBy(lambda x: x["name"])
# Show the first group
print(grouped_records.first())
# Count the groups
job_counts = grouped_records.map(
  lambda x: {
    "name": x[0],
    "job_count": len(x[1])
  }
)
job_counts.collect()
Explanation: pyspark.RDD.groupBy
Using the groupBy operator to count the number of jobs per person...
End of explanation
# Compute a relation of words by line
words_by_line = csv_lines\
  .map(lambda line: line.split(","))
words_by_line.collect()
# Compute a relation of words
flattened_words = csv_lines\
  .map(lambda line: line.split(","))\
  .flatMap(lambda x: x)
flattened_words.collect()
lengths = flattened_words.map(lambda x: len(x))
lengths.collect()
lengths.sum() / lengths.count()
Explanation: Exercises
Use pyspark.RDD.groupBy to group executives by job title, then prepare records with the job title and the count of the number of executives with that job.
Map vs FlatMap
We need to understand the difference between the map and flatmap operators. Map groups items per-record, while flatMap creates a single large group of items.
End of explanation
from pyspark.sql import Row
# Convert the CSV into a pyspark.sql.Row
def csv_to_row(line):
  parts = line.split(",")
  row = Row(
    name=parts[0],
    company=parts[1],
    title=parts[2]
  )
  return row
# Apply the function to get rows in an RDD
rows = csv_lines.map(csv_to_row)
rows.first()
Explanation: Creating Rows
We can create pyspark.sql.Rows out of python objects so we can create pyspark.sql.DataFrames. This is desirable because once we have DataFrames we can run Spark SQL on our data.
End of explanation
records = csv_lines.map(lambda line: line.split(','))
records.collect()
groups = records.groupBy(lambda x: x[0])
counts = groups.map(lambda x: (x[0], len(x[1])))
new_rows = counts.map(lambda x: Row(name=x[0], total=x[1]))
new_rows.collect()
new_rows.toDF().select("name","total").show()
Explanation: Exercises
First count the number of companies for each executive, then create a pyspark.sql.Row for this result.
End of explanation
# Convert to a pyspark.sql.DataFrame
rows_df = rows.toDF()
rows_df.show()
# Register the DataFrame for Spark SQL
rows_df.registerTempTable("executives")
# Generate a new DataFrame with SQL using the SparkSession
job_counts = spark.sql(
SELECT
  name,
  COUNT(*) AS total
  FROM executives
  GROUP BY name
)
job_counts.show()
# Go back to an RDD
job_counts.rdd.map(lambda x: x.asDict()).collect()
Explanation: Creating DataFrames from RDDs
Using the RDD.toDF() method to create a dataframe, registering the DataFrame as a temporary table with Spark SQL, and counting the jobs per person using Spark SQL.
End of explanation
my_rdd = sc.parallelize([1,2,3,4,5])
my_rdd.first()
Explanation: SparkContext.parallelize()
The oppotiste of pyspark.RDD.collect() is SparkContext.parallelize(). Whereas collect pulls data from Spark's memory into local RAM, parallelize sends data from local memory to Spark's memory. 
You can access it like this:
End of explanation
my_data = [
    {"name": "Russell Jurney", "interest": "Ancient Greece"},
    {"name": "Chris Jurney", "interest": "Virtual Reality"},
    {"name": "Bill Jurney", "interest": "Sports"},
    {"name": "Ruth Jurney", "interest": "Wildlife"},
    {"name": "Bob Smith", "interest": "Sports"}
]
my_rdd = sc.parallelize(my_data)
my_rows = my_rdd.map(lambda x: Row(name=x["name"], interest=x["interest"]))
my_df = my_rows.toDF()
my_df.show()
my_df.registerTempTable("people")
spark.sql(SELECT interest, COUNT(*) as total FROM people GROUP BY interest).show()
Explanation: Exercises
Create your own RDD of dict elements with named fields using sc.parallelize. Make it at least 5 records long.
Convert this RDD of dicts into an RDD of pyspark.sql.Row elements.
Convert this RDD of pyspark.sql.Rows into a pyspark.sql.DataFrame.
Run a SQL GROUP BY/COUNT on your new DataFrame.
End of explanation
job_counts.rdd.map(lambda x: x.asDict()).collect()
Explanation: Creating RDDs from DataFrames
We can easily convert back from a DataFrame to an RDD using the pyspark.sql.DataFrame.rdd() method, along with pyspark.sql.Row.asDict() if we desire a Python dict of our records.
End of explanation
my_df.rdd.map(lambda x: x.asDict()).collect()
Explanation: Exercises
Using the data from item 4 in the exercise above, convert the data back to its original form, a local collection of dict elements.
End of explanation
# Load the parquet file containing flight delay records
on_time_dataframe = spark.read.parquet('../data/on_time_performance.parquet')
# Register the data for Spark SQL
on_time_dataframe.registerTempTable("on_time_performance")
# Check out the columns
on_time_dataframe.columns
# Trim the fields and keep the result
trimmed_on_time = on_time_dataframe\
  .select(
    "FlightDate",
    "TailNum",
    "Origin",
    "Dest",
    "Carrier",
    "DepDelay",
    "ArrDelay"
  )
# Sample 0.01% of the data and show
trimmed_on_time.sample(False, 0.0001).show(10)
sampled_ten_percent = trimmed_on_time.sample(False, 0.1)
sampled_ten_percent.show(10)
Explanation: Loading and Inspecting Parquet Files
Using the SparkSession to load files as DataFrames and inspecting their contents...
End of explanation
fd = on_time_dataframe.select("AirTime", "Distance")
fd.show(6)
Explanation: DataFrame Workflow: Calculating Speed in Dataflow and SQL
We can go back and forth between dataflow programming and SQL programming using pyspark.sql.DataFrames. This enables us to get the best of both worlds from these two APIs. For example, if we want to group records and get a total count for each group... a SQL SELECT/GROUP BY/COUNT is the most direct way to do it. On the other hand, if we want to filter data, a dataflow API call like DataFrame.filter() is the cleanest way. This comes down to personal preference for the user. In time you will develop your own style of working.
Dataflow Programming
If we were to look at the AirTime along with the Distance, we could get a good idea of how fast the airplanes were going. Pretty cool! Lets do this using Dataflows first.
Trimming Our Data
First lets select just the two columns of interest: AirTime and Distance. We can always go back and select more columns if we want to extend our analysis, but trimming uneeded fields optimizes performance right away.
End of explanation
hourly_fd = fd.select((fd.AirTime / 60).alias('Hours'), "Distance")
hourly_fd.show(5)
Explanation: From Minutes to Hours
Now lets convert our AirTime from minutes to hours by dividing by 60.
End of explanation
miles_per_hour = hourly_fd.select(
    (hourly_fd.Distance / hourly_fd.Hours).alias('Mph')
)
miles_per_hour.show(10)
Explanation: Raw Calculation
Now lets calculate miles per hour!
End of explanation
fd.select(
    "AirTime", 
    (fd.AirTime / 60).alias('Hours'), 
    "Distance"
).show()
Explanation: Investigating nulls
Looks like we have some errors in some records in our calculation because of missing fields? Lets bring back in the Distance and AirTime fields to see where the problem is coming from.
End of explanation
fd = on_time_dataframe.select("AirTime", "Distance")
filled_fd = fd.filter(fd.AirTime.isNotNull())
hourly_fd = filled_fd.select(
    "AirTime", 
    (filled_fd.AirTime / 60).alias('Hours'), 
    "Distance"
)
mph = hourly_fd.select((hourly_fd.Distance / hourly_fd.Hours).alias('Mph'))
mph.show(10)
Explanation: Filtering nulls
Now that we know some records are missing AirTimes, we can filter those records using pyspark.sql.DataFrame.filter(). Starting from the beginning, lets recalculate our values.
End of explanation
from pyspark.sql.functions import avg
mph.select(
    pyspark.sql.functions.avg(mph.Mph)
).show()
Explanation: Averaging Speed
How fast does the fleet travel overall? Lets compute the average speed for the entire fleet.
End of explanation
on_time_dataframe.registerTempTable("on_time_performance")
mph = spark.sql(
SELECT ( Distance / ( AirTime/60 ) ) AS Mph 
FROM on_time_performance 
WHERE AirTime IS NOT NULL
ORDER BY AirTime
)
mph.show(10)
mph.registerTempTable("mph")
spark.sql("SELECT AVG(Mph) from mph").show()
Explanation: It looks like the average speed of the fleet is 408 mph. Note how along the way we chekced the data for sanity, which led to confidence in our answer. SQL by contast can hide the internals of a query, which might have skewed our average significantly! 
SQL-Based Speed Calculation
Now lets work the same thing out in SQL. Starting from the top:
End of explanation
# Compute a histogram of departure delays
mph\
  .select("Mph")\
  .rdd\
  .flatMap(lambda x: x)\
  .histogram(10)
Explanation: Evaluating SQL
The SQL based solution seems to be better in this case, because we can simply express our calculation all in one place. When complexity grows however, it is best to break a single query into multiple stages where you use SQL or Dataflow programming to massage the data into shape.
Calculating Histograms
Having computed the speed in miles per hour and the overall average speed of passenger jets in the US, lets dig deeper by using the RDD API's histogram method to calculate histograms buckets and values, which will then use to visualize data.
End of explanation
%matplotlib inline
import numpy as np
import matplotlib.mlab as mlab
import matplotlib.pyplot as plt
# Function to plot a histogram using pyplot
def create_hist(rdd_histogram_data):
  Given an RDD.histogram, plot a pyplot histogram
  heights = np.array(rdd_histogram_data[1])
  full_bins = rdd_histogram_data[0]
  mid_point_bins = full_bins[:-1]
  widths = [abs(i - j) for i, j in zip(full_bins[:-1], full_bins[1:])]
  bar = plt.bar(mid_point_bins, heights, width=widths, color='b')
  return bar
# Compute a histogram of departure delays
departure_delay_histogram = mph\
  .select("Mph")\
  .rdd\
  .flatMap(lambda x: x)\
  .histogram(10)
create_hist(departure_delay_histogram)
Explanation: Visualizing Histograms
The problem with the output above is that it is hard to interpret. For better understanding, we need a visualization. We can use matplotlib inline in a Jupyter Notebook to visualize this distribution and see what the tendency of speed of airplanes is around the mean of 408 mph.
End of explanation
# Compute a histogram of departure delays
departure_delay_histogram = mph\
  .select("Mph")\
  .rdd\
  .flatMap(lambda x: x)\
  .histogram(20)
create_hist(departure_delay_histogram)
Explanation: Iterating on a Histogram
That looks interesting, but the bars seem to fat the really see what is going on. Lets double the number of buckets from 10 to 20. We can reuse the create_hist() method to do so.
End of explanation
# Dump the unneeded fields
tail_numbers = on_time_dataframe.rdd.map(lambda x: x.TailNum)
tail_numbers = tail_numbers.filter(lambda x: x != '' and x is not None)
# distinct() gets us unique tail numbers
unique_tail_numbers = tail_numbers.distinct()
# now we need a count() of unique tail numbers
airplane_count = unique_tail_numbers.count()
print("Total airplanes: {}".format(airplane_count))
Explanation: Speed Summary
You've now seen how to calculate different values in both SQL and Dataflow style, how to switch between the two methods, how to switch between the pyspark.RDD and pyspark.sql.DataFrame APIs and you're starting to build a proficiency in PySpark!
Counting Airplanes in the US Fleet
Lets convert our on_time_dataframe (a DataFrame) into an RDD to calculate the total number of airplanes in the US fleet.
End of explanation
origin_hour_dist = on_time_dataframe.filter(
    on_time_dataframe.AirTime.isNotNull()
).select(
    "Origin", 
    (on_time_dataframe.AirTime/60).alias("Hours"), 
    "Distance"
)
mph_origins = origin_hour_dist.select(
    "Origin", 
    (origin_hour_dist.Distance / origin_hour_dist.Hours).alias("Mph")
)
mph_origins.registerTempTable("mph_origins")
avg_speeds = mph_origins.groupBy("Origin").agg({"Mph": "avg"}).alias("Mph")
avg_speeds.show()
on_time_dataframe.columns
Explanation: Exercise 1: Characterizing Airports
Using the techniques we demonstated above, calculate any 3 out of 4 of the following things using both the SQL and the Dataflow methods for each one. That is: implement each calculation twice - once in SQL and once using Dataflows. Try to use both the RDD and DataFrame APIs as you work.
How many airports are there in the united states?
What is the average flight time for flights arriving in San Francisco (SFO)? What does the distribution of this value look like? Plot a histogram using the create_hist method shown above.
Which American airport has the fastest out-bound speeds? What does the distribution of the flight speeds at this one airport look like? Plot a histogram using the create_hist method shown above.
What were the worst travel dates in terms of overall delayed flights in the US in 2015?
End of explanation
# Calculate average of every numeric field
on_time_dataframe.groupBy("Origin").avg().show(1)
# Calculate verage AirTime per origin city
on_time_dataframe.groupBy("Origin").agg({"AirTime": "mean"}).show(1)
# Get the count of flights from each origin
on_time_dataframe.groupBy("Origin").count().show(1)
# Get the maximum airtime for flights leaving each city
on_time_dataframe.groupby("Origin").agg({"AirTime": "max"}).show(1)
# Get the maximum of all numeric columns for flights leaving each city
on_time_dataframe.groupBy("Origin").max().show(1)
# Get the shortest flight for each origin airport
on_time_dataframe.groupBy("Origin").agg({"AirTime": "min"}).show(1)
# Total minutes flown from each airport
on_time_dataframe.groupBy("Origin").agg({"AirTime": "sum"}).show(1)
Explanation: Calculating with DataFrame.groupBy
We can use Spark SQL to calculate things using DataFrames, but we can also group data and calculate as we did with RDDs. For a full list of methods you can apply to grouped DataFrames, see the documentation for pyspark.sql.GroupedData. Below we will demonstrate some of these methods.
End of explanation
on_time_dataframe\
    .filter("Origin == 'ATL'")\
    .groupBy("Origin")\
    .pivot("Dest")\
    .avg("AirTime")\
    .rdd\
    .map(lambda x: x.asDict())\
    .collect()[0]
Explanation: Pivoting DataFrames
One useful function of DataFrames is pivot. Pivot lets you compute pivot tables from data. Lets use pivot to calculate the average flight times between Atlanta ATL and other airports.
End of explanation
mph = spark.sql(
SELECT 
    Distance, 
    ( Distance / ( AirTime/60 ) ) AS Mph 
FROM on_time_performance 
WHERE AirTime IS NOT NULL
)
mph.show(10)
Explanation: Plotting Scatterplots
Another type of visualization that is of interest to data scientists is the scatterplot. A scatterplot enables us to compare the trend of one value plotted against the other. For example, we could calculate the relationship between Origin and Dest Distance and the Mph speed figure we calculated earlier. Are longer flights generally faster, or not?
To prepare a scatterplot, we need to use matplotlib again, so we'll need to look at what its scatterplot API expects. The matplotlib.pyplot.scatter API takes two independant lists of values for the variables x and y, so we must compute them for Distance and Mph.
End of explanation
distance = mph.select("Distance").rdd.flatMap(lambda x: x)
distance = distance.collect()
distance[0:10]
speed = mph.select("Mph").rdd.flatMap(lambda x: x)
speed = speed.collect()
speed[0:10]
Explanation: Collecting Data
Note that we will have to convert our data from existing within our Spark cluster's memory to within our local computer's memory where matplotlib runs.
End of explanation
print("Total distances: {:,}".format(len(distance)))
print("Total speeds: {:,}".format(len(speed)))
Explanation: Sampling Data
When I tried to plot this data, it took a very long time to draw. This is because... well, how many unique values are there for each variable? Lets see.
End of explanation
sample = mph.sample(False, 0.001)
sample.count()
Explanation: It is hard to plot 5.7 million dots on a scatterplot that will fit on a computer screen. So lets sample our data. We can use PySpark DataFrame's sample method. Lets take a 0.1% random sample without replacement, which will leave us with 5,687 or so data points - something we can more easily manage.
End of explanation
speed = sample.select("Mph").rdd.flatMap(lambda x: x).collect()
distance = sample.select("Distance").rdd.flatMap(lambda x: x).collect()
print("{:,} x {:,} records!".format(
    len(speed),
    len(distance)
))
Explanation: Note that we need to sample once and then split the datasets out - otherwise the data for a single observation will be scrambled across variables. We don't want that! All our scatterplots would show no relationships at all.
End of explanation
%matplotlib inline
import numpy as np
import matplotlib.pyplot as plt
plt.rcParams["figure.figsize"] = (18,12)
plt.scatter(
    distance, 
    speed, 
    alpha=0.5
)
plt.title("Distance x Speed")
plt.xlabel("Distance")
plt.ylabel("Speed")
plt.show()
Explanation: Fun with matplotlib.pyplot.scatter
Now we feed the scatter API distance as x and speed as y, giving it a title and x and y axes. Note that we also specify a size in inches via the figure.figsize rcParam.
End of explanation
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import cross_val_score
import numpy as np
x = np.array(distance)
y = np.array(speed)
x_test = np.arange(0, 5000, 100)
model = Pipeline([
    ('poly', PolynomialFeatures(degree=3)),
    ('linear', LinearRegression(fit_intercept=False))
])
model = model.fit(x[:, np.newaxis], y)
model.named_steps['linear'].coef_
y_out = model.predict(x_test.reshape(-1,1))
cross_val_score(model, x.reshape(-1,1), y)
Explanation: Interpreting Our Scatterplot
We can see pretty clearly that as distance increases, average speed across that distance increases rapidly and then levels off as the distance increases.
Exercises
Query the on_time_dataframe to focus on two numeric fields. 
Plot a histogram of one of these fields
Plot a scatterplot of both of these fields
Predicting Speed Given Distance
It is often the case that once we characterize a distribution, we want to create a function to predict one variable given the other. Lets take this example further by fitting a polynomial regression to describe our data. We use sklearn.pipeline.Pipeline to chain a sklearn.preprocessing.PolynomialFeatures to a sklearn.linear_model.LinearRegression. Other than that, we simply define x and y, and fit a model to those values. Then we finally compute a cross value score, to see the model's performance. We'll see this pattern again when we use large data tools in Spark MLlib.
End of explanation
%matplotlib inline
import numpy as np
import matplotlib.pyplot as plt
plt.rcParams["figure.figsize"] = (18,12)
plt.scatter(
    distance, 
    speed, 
    alpha=0.5
)
plt.plot(
    x_test,
    y_out, 
    color='orange', 
    linewidth=3
)
plt.title("Distance x Speed")
plt.xlabel("Distance")
plt.ylabel("Speed")
plt.show()
Explanation: Visualizing Polynomial Fit
Because we are running a polynomial regression, where we get to decide the degree of the polynomial. To help decide, lets plot a polynomial fit line to the data using matplotlib.
End of explanation
tail_numbers = on_time_dataframe.select("TailNum").distinct()
tail_numbers.show(6)
Explanation: Joining Data in PySpark
Next we're going to learn how to join between datasets using PySpark. We're going to pick up from an example we're going to work in chapter 6, and explore it more deeply. To begin with, we will prepare a list of TailNum (tail numbers) from the FAA flight records. These uniquely identify each airplane from each flight.
Unique Tail Numbers
End of explanation
faa_tail_number_inquiry = spark.read.json('../data/faa_tail_number_inquiry.jsonl')
airplane_records = faa_tail_number_inquiry.select(
    faa_tail_number_inquiry.TailNum.alias("FAATailNum"), 
    "Model",
    "Engine_Model"
)
airplane_records.show(6)
Explanation: FAA Airplane Records
We will trim the FAA records down to just the TailNum, Model and Engine_Model. Note that we go ahead and rename the TailNum field to FAATailNum using the pyspark.sql.functions.alias() method. This avoids having two fields referenced by the same name once we perform our joins.
End of explanation
# INNER JOIN
print(
    "FAA tail numbers: {:,}".format(
        tail_numbers.count()
    )
)
print(
    "Airplane records: {:,}".format(
        airplane_records.count()
    )
)
inner_joined = tail_numbers.join(
    airplane_records, 
    tail_numbers.TailNum == airplane_records.FAATailNum, 
    'inner'
)
print(
    "Joined records:   {:,}".format(
        inner_joined.count()
    )
)
Explanation: Inner Joins
You may be familiar with an inner join from SQL. An inner join joins two datasets based on the presence of a key from one dataset in the other. Records which don't have a key that appears in the other table do not appear in the final output.
End of explanation
inner_joined.show(6)
Explanation: Inner Join Results
Note that there are as many records in the output as there were in the FAA Airplane dataset - indicating that there was a representative of every tail number from that dataset in the on-time performance flight records. Lets take a look at the records themselves.
End of explanation
# INNER JOIN
print(
    "FAA tail numbers: {:,}".format(
        tail_numbers.count()
    )
)
print(
    "Airplane records: {:,}".format(
        airplane_records.count()
    )
)
left_outer_joined = tail_numbers.join(
    airplane_records, 
    tail_numbers.TailNum == airplane_records.FAATailNum, 
    'left_outer'
)
print(
    "Joined records:   {:,}".format(
        left_outer_joined.count()
    )
)
Explanation: Note how convenient it is that we renamed one of the keys FAATailNum. If we hadn't, we'd have two columns with the same name now and would have trouble referring to one or the other.
Left Outer Join
Another type of join is the left outer join. It ensures that one record will remain in the output from the left side of the join no matter what. If a match on the join keys is found, the fields for the record on the right will be filled. If a match is not found, they will be empty. 
Lets look at how this works with our two datasets.
End of explanation
left_outer_joined.show(6)
Explanation: Left Outer Join Result
Note that there were 4,898 records on the left side of our join and there are the same number on the output of our join. Lets take a look at what both matched and unmatched records look like:
End of explanation
# Filter down to the fields we need to identify and link to a flight
flights = on_time_dataframe.rdd.map(
    lambda x: (x.Carrier, x.FlightDate, x.FlightNum, x.Origin, x.Dest, x.TailNum)
)
# Group flights by tail number, sorted by date, then flight number, then origin/dest
flights_per_airplane = flights\
  .map(lambda nameTuple: (nameTuple[5], [nameTuple[0:5]]))\
  .reduceByKey(lambda a, b: a + b)\
  .map(lambda tuple:
      {
        'TailNum': tuple[0], 
        'Flights': sorted(tuple[1], key=lambda x: (x[1], x[2], x[3], x[4]))
      }
    )
flights_per_airplane.first()
Explanation: Note that some records have fields filled out, and some don't.
Right Outer Join
Another type of join is the right outer join. This works the opposite of a left outer join. In this case, the output will preserve a record for each and every record on the right side of the join. Use the right_outer join type to perform this kind of join.
Exercises
Go back and perform a right outer join on the preceding two datasets. Is the distinct() call on the FAA on-time performance records still needed? Why or why not?
Using RDDs and Map/Reduce to Prepare a Complex Record
End of explanation
total_flights = on_time_dataframe.count()
# Flights that were late leaving...
late_departures = on_time_dataframe.filter(
  on_time_dataframe.DepDelayMinutes > 0
)
total_late_departures = late_departures.count()
print(f'{total_late_departures:,}')
# Flights that were late arriving...
late_arrivals = on_time_dataframe.filter(
  on_time_dataframe.ArrDelayMinutes > 0
)
total_late_arrivals = late_arrivals.count()
print(f'{total_late_arrivals:,}')
# Get the percentage of flights that are late, rounded to 1 decimal place
pct_late = round((total_late_arrivals / (total_flights * 1.0)) * 100, 1)
pct_late
Explanation: Counting Late Flights
End of explanation
# Flights that left late but made up time to arrive on time...
on_time_heros = on_time_dataframe.filter(
  (on_time_dataframe.DepDelayMinutes > 0)
  &
  (on_time_dataframe.ArrDelayMinutes <= 0)
)
total_on_time_heros = on_time_heros.count()
print(f'{total_on_time_heros:,}')
Explanation: Counting Flights with Hero Captains
"Hero Captains" are those that depart late but make up time in the air and arrive on time or early.
End of explanation
print("Total flights:   {:,}".format(total_flights))
print("Late departures: {:,}".format(total_late_departures))
print("Late arrivals:   {:,}".format(total_late_arrivals))
print("Recoveries:      {:,}".format(total_on_time_heros))
print("Percentage Late: {}%".format(pct_late))
Explanation: Printing Our Results
End of explanation
# Get the average minutes late departing and arriving
spark.sql(
SELECT
  ROUND(AVG(DepDelay),1) AS AvgDepDelay,
  ROUND(AVG(ArrDelay),1) AS AvgArrDelay
FROM on_time_performance
).show()
Explanation: Computing the Average Lateness Per Flights
End of explanation
# Why are flights late? Lets look at some delayed flights and the delay causes
late_flights = spark.sql(
SELECT
  ArrDelayMinutes,
  WeatherDelay,
  CarrierDelay,
  NASDelay,
  SecurityDelay,
  LateAircraftDelay
FROM
  on_time_performance
WHERE
  WeatherDelay IS NOT NULL
  OR
  CarrierDelay IS NOT NULL
  OR
  NASDelay IS NOT NULL
  OR
  SecurityDelay IS NOT NULL
  OR
  LateAircraftDelay IS NOT NULL
ORDER BY
  FlightDate
)
late_flights.sample(False, 0.01).show()
Explanation: Inspecting Late Flights
End of explanation
# Calculate the percentage contribution to delay for each source
total_delays = spark.sql(
SELECT
  ROUND(SUM(WeatherDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_weather_delay,
  ROUND(SUM(CarrierDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_carrier_delay,
  ROUND(SUM(NASDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_nas_delay,
  ROUND(SUM(SecurityDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_security_delay,
  ROUND(SUM(LateAircraftDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_late_aircraft_delay
FROM on_time_performance
)
total_delays.show()
Explanation: Determining Why Flights Are Late
End of explanation
# Eyeball the first to define our buckets
weather_delay_histogram = on_time_dataframe\
  .select("WeatherDelay")\
  .rdd\
  .flatMap(lambda x: x)\
  .histogram([1, 5, 10, 15, 30, 60, 120, 240, 480, 720, 24*60.0])
print(weather_delay_histogram)
# See above for definition
create_hist(weather_delay_histogram)
Explanation: Computing a Histogram of Weather Delayed Flights
End of explanation
# Transform the data into something easily consumed by d3
def histogram_to_publishable(histogram):
  record = {'key': 1, 'data': []}
  for label, value in zip(histogram[0], histogram[1]):
    record['data'].append(
      {
        'label': label,
        'value': value
      }
    )
  return record
# Recompute the weather histogram with a filter for on-time flights
weather_delay_histogram = on_time_dataframe\
  .filter(
    (on_time_dataframe.WeatherDelay.isNotNull())
    &
    (on_time_dataframe.WeatherDelay > 0)
  )\
  .select("WeatherDelay")\
  .rdd\
  .flatMap(lambda x: x)\
  .histogram([0, 15, 30, 60, 120, 240, 480, 720, 24*60.0])
print(weather_delay_histogram)
record = histogram_to_publishable(weather_delay_histogram)
record
Explanation: Preparing a Histogram for Visualization by d3.js
End of explanation
from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType, DateType, TimestampType
from pyspark.sql.types import StructType, StructField
from pyspark.sql.functions import udf
schema = StructType([
  StructField("ArrDelay", DoubleType(), True),     # "ArrDelay":5.0
  StructField("CRSArrTime", TimestampType(), True),    # "CRSArrTime":"2015-12-31T03:20:00.000-08:00"
  StructField("CRSDepTime", TimestampType(), True),    # "CRSDepTime":"2015-12-31T03:05:00.000-08:00"
  StructField("Carrier", StringType(), True),     # "Carrier":"WN"
  StructField("DayOfMonth", IntegerType(), True), # "DayOfMonth":31
  StructField("DayOfWeek", IntegerType(), True),  # "DayOfWeek":4
  StructField("DayOfYear", IntegerType(), True),  # "DayOfYear":365
  StructField("DepDelay", DoubleType(), True),     # "DepDelay":14.0
  StructField("Dest", StringType(), True),        # "Dest":"SAN"
  StructField("Distance", DoubleType(), True),     # "Distance":368.0
  StructField("FlightDate", DateType(), True),    # "FlightDate":"2015-12-30T16:00:00.000-08:00"
  StructField("FlightNum", StringType(), True),   # "FlightNum":"6109"
  StructField("Origin", StringType(), True),      # "Origin":"TUS"
])
features = spark.read.json(
  "../data/simple_flight_delay_features.jsonl.bz2",
  schema=schema
)
features.first()
Explanation: Building a Classifier Model to Predict Flight Delays
Loading Our Data
End of explanation
#
# Check for nulls in features before using Spark ML
#
null_counts = [(column, features.where(features[column].isNull()).count()) for column in features.columns]
cols_with_nulls = filter(lambda x: x[1] > 0, null_counts)
print(list(cols_with_nulls))
Explanation: Check Data for Nulls
End of explanation
#
# Add a Route variable to replace FlightNum
#
from pyspark.sql.functions import lit, concat
features_with_route = features.withColumn(
  'Route',
  concat(
    features.Origin,
    lit('-'),
    features.Dest
  )
)
features_with_route.select("Origin", "Dest", "Route").show(5)
Explanation: Add a Route Column
Demonstrating the addition of a feature to our model...
End of explanation
#
# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay
#
from pyspark.ml.feature import Bucketizer
splits = [-float("inf"), -15.0, 0, 30.0, float("inf")]
bucketizer = Bucketizer(
  splits=splits,
  inputCol="ArrDelay",
  outputCol="ArrDelayBucket"
)
ml_bucketized_features = bucketizer.transform(features_with_route)
# Check the buckets out
ml_bucketized_features.select("ArrDelay", "ArrDelayBucket").show()
Explanation: Bucketizing ArrDelay into ArrDelayBucket
End of explanation
#
# Extract features tools in with pyspark.ml.feature
#
from pyspark.ml.feature import StringIndexer, VectorAssembler
# Turn category fields into categoric feature vectors, then drop intermediate fields
for column in ["Carrier", "DayOfMonth", "DayOfWeek", "DayOfYear",
               "Origin", "Dest", "Route"]:
  string_indexer = StringIndexer(
    inputCol=column,
    outputCol=column + "_index"
  )
  ml_bucketized_features = string_indexer.fit(ml_bucketized_features)\
                                          .transform(ml_bucketized_features)
# Check out the indexes
ml_bucketized_features.show(6)
Explanation: Indexing Our String Fields into Numeric Fields
End of explanation
# Handle continuous, numeric fields by combining them into one feature vector
numeric_columns = ["DepDelay", "Distance"]
index_columns = ["Carrier_index", "DayOfMonth_index",
                   "DayOfWeek_index", "DayOfYear_index", "Origin_index",
                   "Origin_index", "Dest_index", "Route_index"]
vector_assembler = VectorAssembler(
  inputCols=numeric_columns + index_columns,
  outputCol="Features_vec"
)
final_vectorized_features = vector_assembler.transform(ml_bucketized_features)
# Drop the index columns
for column in index_columns:
  final_vectorized_features = final_vectorized_features.drop(column)
# Check out the features
final_vectorized_features.show()
Explanation: Combining Numeric Fields into a Single Vector
End of explanation
#
# Cross validate, train and evaluate classifier
#
# Test/train split
training_data, test_data = final_vectorized_features.randomSplit([0.7, 0.3])
# Instantiate and fit random forest classifier
from pyspark.ml.classification import RandomForestClassifier
rfc = RandomForestClassifier(
  featuresCol="Features_vec",
  labelCol="ArrDelayBucket",
  maxBins=4657,
  maxMemoryInMB=1024
)
model = rfc.fit(training_data)
# Evaluate model using test data
predictions = model.transform(test_data)
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator(labelCol="ArrDelayBucket", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = {}".format(accuracy))
# Check a sample
predictions.sample(False, 0.001, 18).orderBy("CRSDepTime").show(6)
Explanation: Training Our Model in an Experimental Setup
End of explanation
<END_TASK>