<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Continuing Graphs of Specimens Over Time - Continents and GBIF
This notebook continues the work done in 01_iDigBio_Specimens_Collected_Over_Time.ipynb, you may want to review that notebook before contiuing. If you are interested in seeing what kinds of outputs are possible, you are welcome to skip to the graphs at the end of this notebook.
Our goal for this notebook is to see how many specimens were collected on each continent during each year from 1817 to 2017. A stacked bar graph with a different color for each continent is a good way to view this data.
We will introduce some additional concepts such as user defined functions, crosstabs, and how to write SQL statements instead of using the native pyspark methods. We'll move quickly though concepts already covered in earlier notebooks.
Setup
Import needed functions and libraries.
Step1:  Loading the data set
Step2:  Year collected by continent
To get a feel for what a summary of year and continent look like, we will again group by the year column but also add the continent column. The result will be a row for each unique combination of year and continent with the count of how many times that occurs.
Step4:  There are a lot of things that are not continents there! While iDigBio cleans stuff up a bit before indexing it, not everything is obvious to do. For the purposes of our graph, we will make an "other" continent and group things that are not one of the 7 continents (plus "oceania" which GBIF uses and we'll look at GBIF data lower down) into that category.
Fixing continent with a user defined function
To create a new column that has a fixed continent field, we will write a user defined function in Python and apply that function to every row in the data frame. This is exactly how the year() function from the pyspark library works. But, since we are writing the function in Python, we need to convert it into a data frame aware function with the pyspark function udf(). Once we do that, we can use our custom function just the same way we used year() above.
So let's write our Python function
Step5:  That output looks right. So now let's make a user defined function for Spark from our Python function. Notice that we pass the Python function as an argument to udf() and udf() returns to us a new function. In Python, functions are what's called "first class objects" and we can pass them around and manipulate them just like a variable.
Step6:  Now re-run our grouping. This time use the fix_continent_udf() function to calculate a new column and add it to the dataframe with the withColumn() method. Then we can refer to that new column in the group by method. (Yes, we could have put the fix_continent_udf() call into the group by just like we put the year() call but this is an opportunity to introduce withColumn().)
Step7:  That looks much better. Notice now that we can actually see all of the yearcollected=None values. Before there were many continents so they went well past the top 20 results we were looking at. Now we can see all 8 counts, one for each continent.
Cross tabs to make plottable data
What we have created at this point is a long list of year, continent, and count combinations. This representation of the data is normalized and can be thought of as the "long" form. Plotting wants to see each variable in one column though so the better represenation would the de-normalized or "wide" form. We can do this in Pandas with piviot() or in R you can use one of the reformating functions. We can also do this in Spark directly.
We will also take this opportunity to show how more typically a full data pipeline is shown in Spark. We will calculate the Pandas data frame all at once starting from the original df we loaded, filtering on years, creating the new fixed continents, and summarizing with group by. Each line in the pipeline below becomes the input for the next lower line. 
Notice that we first select only the columns of interest to let Spark know we don't need the other 200+ fields in iDigBio. Then we filter on year to cut down on the rows we need to work with. Only then do we run our udf which only runs on the filtered rows.
The new step in here is the crosstab() method. This is the method that takes all values in the column that is the second argument and makes them in to columns. You'll see the output after.
Step9:  That looks pretty intuitive and we made the whole thing with one Spark statement. You can tell it's a Pandas dataframe now because the display is nice and tabular.
Making the graph
Plotting isn't the focus of this notebook. Briefly, we'll make a function that stacks up multiple variables (continent counts) on top of each other and shades them different colors.
Step10:  and then we'll call our function to make the graph and add some axes.
Step11:  There it is, the number of specimen records in iDigBio by collection date and continent. We chose to add some vertical lines to the graph at particularly interesting dates. You can read the code comments to see what those are.
Doing the same with GBIF data
GBIF periocdically produces a similar graph to the above on their analytics page titled "Records by year of occurrence" but it's not very comparable because it include observation records as well. It would be great if we could run the same process on their data after filtering it for only specimens so we can look at how the contents of different occurrance record aggregators differs. (This is an expected and healthy thing. One of the goals of the GUODA collaboration is to begin to characterize the relationships between sets of aggregated data across domains without that work being done by one entity. Notice too that this copy of GBIF is pretty old. Again this is GUODA's fault as we have not been running our import regularly while we worked on improving other parts of our system.)
We can start by reading in the GBIF data that is stored in GUODA.
Step13:  We use a different process for formating GBIF data and as part of GUODA's process, the columns are named with a syntax that can be difficult to work with in Python. First let's use the SQL "as" syntax to rename the columns.
Spark data frames can also be manupulated with SQL statements instead of the pyspark functions. Note that Spark does not support the full SQL standard, it only supports syntx similar to the native functions.
The first step is to tell Spark what name we want to use to refer to our data frame with in the SQL. That is the createOrReplaceTempView() method. After than we can send our SQL as a string to the SQL Context associated with our Spark Context and we will get back a data frame. Because we'll use this data frame more than once, we'll persist it so it will not be recalculated.
As part of our SQL statement we will also filter for only those records that contain the string "SPECIMEN" in the basisOfRecord field. This may or may not be the best way to select only specimens from GBIF but for our graph it's good enough.
Step14:  We will again run the crosstab all at once on this data frame with renamed columns. This is exactly the same summarization we ran on the iDigBio data frame.
Step15:  And we'll use the exact same code for graphing as well. We just need to remove the "australia" and it's color from our continent list because GBIF doesn't use it.
<ASSISTANT_TASK:>
Python Code:
# col() selects columns from a data frame, year() works on dates, and udf() creates user
# defined functions
from pyspark.sql.functions import col, year, udf
# Plotting library and configuration to show graphs in the notebook
import matplotlib.pyplot as plt
%matplotlib inline
Explanation: Continuing Graphs of Specimens Over Time - Continents and GBIF
This notebook continues the work done in 01_iDigBio_Specimens_Collected_Over_Time.ipynb, you may want to review that notebook before contiuing. If you are interested in seeing what kinds of outputs are possible, you are welcome to skip to the graphs at the end of this notebook.
Our goal for this notebook is to see how many specimens were collected on each continent during each year from 1817 to 2017. A stacked bar graph with a different color for each continent is a good way to view this data.
We will introduce some additional concepts such as user defined functions, crosstabs, and how to write SQL statements instead of using the native pyspark methods. We'll move quickly though concepts already covered in earlier notebooks.
Setup
Import needed functions and libraries.
End of explanation
df = sqlContext.read.load("/guoda/data/idigbio-20190612T171757.parquet")
df.count()
Explanation: Loading the data set
End of explanation
yc_sum = (df
          .groupBy(year("datecollected").cast("integer").alias("yearcollected"),
                   "continent")
          .count()
          .orderBy("yearcollected")
          )
yc_sum.head(20)
Explanation: Year collected by continent
To get a feel for what a summary of year and continent look like, we will again group by the year column but also add the continent column. The result will be a row for each unique combination of year and continent with the count of how many times that occurs.
End of explanation
# allowed continents
continents = set(["africa", "australia", "asia", "oceania", "europe", "north america", "south america"])
def fix_continent(c):
    
    If the given string contains one of the continents, return which one. If not
    then return "other". Lower case the string and also replace "_" with " " to
    be able to clean up GBIF's representation of continents that contain spaces.
    
    for continent in continents:
        if (c) and (continent in c.lower().replace("_", " ")):
            return continent
    return "other"
# Some example calls to this function
print(fix_continent(None)) 
print(fix_continent("europe"))    
print(fix_continent("oceania"))
print(fix_continent("NORTH_AMERICA"))
print(fix_continent("North America, Canada, Manitoba, Churchill"))
print(fix_continent("East Indies, Indonesia: Pulo Pandjang, off Sumatra"))
print(fix_continent("Asia; Thailand; Pathum Thani"))
Explanation: There are a lot of things that are not continents there! While iDigBio cleans stuff up a bit before indexing it, not everything is obvious to do. For the purposes of our graph, we will make an "other" continent and group things that are not one of the 7 continents (plus "oceania" which GBIF uses and we'll look at GBIF data lower down) into that category.
Fixing continent with a user defined function
To create a new column that has a fixed continent field, we will write a user defined function in Python and apply that function to every row in the data frame. This is exactly how the year() function from the pyspark library works. But, since we are writing the function in Python, we need to convert it into a data frame aware function with the pyspark function udf(). Once we do that, we can use our custom function just the same way we used year() above.
So let's write our Python function:
End of explanation
fix_continent_udf = udf(fix_continent)
Explanation: That output looks right. So now let's make a user defined function for Spark from our Python function. Notice that we pass the Python function as an argument to udf() and udf() returns to us a new function. In Python, functions are what's called "first class objects" and we can pass them around and manipulate them just like a variable.
End of explanation
yc_sum = (df
          .withColumn("fixed", fix_continent_udf(df.continent))
          .groupBy(year("datecollected").cast("integer").alias("yearcollected"),
                   "fixed")
          .count()
          .orderBy("yearcollected")
          )
yc_sum.head(20)
Explanation: Now re-run our grouping. This time use the fix_continent_udf() function to calculate a new column and add it to the dataframe with the withColumn() method. Then we can refer to that new column in the group by method. (Yes, we could have put the fix_continent_udf() call into the group by just like we put the year() call but this is an opportunity to introduce withColumn().)
End of explanation
pandas_yc_cross = (df
            .select(year("datecollected").cast("integer").alias("yearcollected"),
                    df.continent)
            .filter(col("yearcollected") >= 1817)
            .filter(col("yearcollected") <= 2017)
            .withColumn("fixed", fix_continent_udf(df.continent))
            .crosstab("yearcollected", "fixed")
            .orderBy(col("yearcollected_fixed"))
            .toPandas()
            )
pandas_yc_cross.head(5)
Explanation: That looks much better. Notice now that we can actually see all of the yearcollected=None values. Before there were many continents so they went well past the top 20 results we were looking at. Now we can see all 8 counts, one for each continent.
Cross tabs to make plottable data
What we have created at this point is a long list of year, continent, and count combinations. This representation of the data is normalized and can be thought of as the "long" form. Plotting wants to see each variable in one column though so the better represenation would the de-normalized or "wide" form. We can do this in Pandas with piviot() or in R you can use one of the reformating functions. We can also do this in Spark directly.
We will also take this opportunity to show how more typically a full data pipeline is shown in Spark. We will calculate the Pandas data frame all at once starting from the original df we loaded, filtering on years, creating the new fixed continents, and summarizing with group by. Each line in the pipeline below becomes the input for the next lower line. 
Notice that we first select only the columns of interest to let Spark know we don't need the other 200+ fields in iDigBio. Then we filter on year to cut down on the rows we need to work with. Only then do we run our udf which only runs on the filtered rows.
The new step in here is the crosstab() method. This is the method that takes all values in the column that is the second argument and makes them in to columns. You'll see the output after.
End of explanation
import pandas as pd
import numpy as np
def stack_bar_plot(pdf, columns, colors):
    
    Matplotlib needs to have the bars stacked on top of each other by hand.
    First, we plot one column of data, then we make a ...
    
    plots = []
    bottoms = pd.DataFrame(np.zeros((len(pdf['yearcollected_fixed']), 1)))
    for c in columns:
        plots.append(
          plt.bar(pdf['yearcollected_fixed'], pdf[c], 
                  color=colors[len(plots)], edgecolor='none',
                  width=1.0, bottom=bottoms[0])
        )
        bottoms[0] += pdf[c]
    return plots
Explanation: That looks pretty intuitive and we made the whole thing with one Spark statement. You can tell it's a Pandas dataframe now because the display is nice and tabular.
Making the graph
Plotting isn't the focus of this notebook. Briefly, we'll make a function that stacks up multiple variables (continent counts) on top of each other and shades them different colors.
End of explanation
# set up ordering of continents and their colors
continents_list = sorted(continents)
continents_list.insert(0, "other")
colors = ["#888888", "#a75902", "#00e5c8", "#1bcda7", "#37b686", "#539f65", "#51827c", "#bf6603"]
plots = stack_bar_plot(pandas_yc_cross, continents_list, colors)
        
# Start of WWI
plt.axvline(x=1914)
# Start of WWII
plt.axvline(x=1939)
# "1988 - October 31: President Reagan signs the NSF Authorization Act of 1988,
# thereby authorizing the doubling of the NSF budget over the next five years."
plt.axvline(x=1988)
plt.legend(plots, continents_list, loc=2)
plt.title("Specimens in iDigBio by Collection Year and Continent")
plt.ylabel("Number of Specimen Records")
plt.xlabel("Year")
axes = plt.gca()
axes.set_xlim([1815, 2020])
axes.set_ylim([0, 1200000])
fig = plt.gcf()
fig.set_size_inches(12, 4)
Explanation: and then we'll call our function to make the graph and add some axes.
End of explanation
gbif_df = sqlContext.read.parquet("/guoda/data/gbif-idigbio.parquet/source=gbif/date=20160825")
Explanation: There it is, the number of specimen records in iDigBio by collection date and continent. We chose to add some vertical lines to the graph at particularly interesting dates. You can read the code comments to see what those are.
Doing the same with GBIF data
GBIF periocdically produces a similar graph to the above on their analytics page titled "Records by year of occurrence" but it's not very comparable because it include observation records as well. It would be great if we could run the same process on their data after filtering it for only specimens so we can look at how the contents of different occurrance record aggregators differs. (This is an expected and healthy thing. One of the goals of the GUODA collaboration is to begin to characterize the relationships between sets of aggregated data across domains without that work being done by one entity. Notice too that this copy of GBIF is pretty old. Again this is GUODA's fault as we have not been running our import regularly while we worked on improving other parts of our system.)
We can start by reading in the GBIF data that is stored in GUODA.
End of explanation
gbif_df.createOrReplaceTempView("gbif_df")
renamed_df = sqlContext.sql(
SELECT `http://rs.tdwg.org/dwc/terms/eventDate` as eventDate,
       `http://rs.tdwg.org/dwc/terms/continent` as continent
FROM gbif_df
WHERE `http://rs.tdwg.org/dwc/terms/basisOfRecord` LIKE "%SPECIMEN%"
).persist()
renamed_df.count()
Explanation: We use a different process for formating GBIF data and as part of GUODA's process, the columns are named with a syntax that can be difficult to work with in Python. First let's use the SQL "as" syntax to rename the columns.
Spark data frames can also be manupulated with SQL statements instead of the pyspark functions. Note that Spark does not support the full SQL standard, it only supports syntx similar to the native functions.
The first step is to tell Spark what name we want to use to refer to our data frame with in the SQL. That is the createOrReplaceTempView() method. After than we can send our SQL as a string to the SQL Context associated with our Spark Context and we will get back a data frame. Because we'll use this data frame more than once, we'll persist it so it will not be recalculated.
As part of our SQL statement we will also filter for only those records that contain the string "SPECIMEN" in the basisOfRecord field. This may or may not be the best way to select only specimens from GBIF but for our graph it's good enough.
End of explanation
pandas_yc_cross_gbif = (renamed_df
            .select(year("eventDate").cast("integer").alias("yearcollected"),
                    renamed_df.continent)
            .filter(col("yearcollected") >= 1817)
            .filter(col("yearcollected") <= 2017)
            .withColumn("fixed", fix_continent_udf(renamed_df.continent))
            .crosstab("yearcollected", "fixed")
            .orderBy(col("yearcollected_fixed"))
            .toPandas()
            )
pandas_yc_cross_gbif.head(5)
Explanation: We will again run the crosstab all at once on this data frame with renamed columns. This is exactly the same summarization we ran on the iDigBio data frame.
End of explanation
# GBIF doesn't have "australia"
try:
    continents_list.remove("australia")
    colors.remove("#1bcda7")
except:
    pass
plots = stack_bar_plot(pandas_yc_cross_gbif, continents_list, colors)
# Start of WWI
plt.axvline(x=1914)
# Start of WWII
plt.axvline(x=1939)
# "1988 - October 31: President Reagan signs the NSF Authorization Act of 1988, thereby authorizing the doubling of the NSF budget over the next five years."
plt.axvline(x=1988)
plt.legend(plots, continents_list, loc=2)
plt.title("Specimens in GBIF by Collection Year and Continent")
plt.ylabel("Number of Specimen Records")
plt.xlabel("Year")
axes = plt.gca()
axes.set_xlim([1815, 2020])
axes.set_ylim([0, 1200000])
fig = plt.gcf()
fig.set_size_inches(12, 4)
Explanation: And we'll use the exact same code for graphing as well. We just need to remove the "australia" and it's color from our continent list because GBIF doesn't use it.
End of explanation
<END_TASK>