%matplotlib inline
import matplotlib.pyplot as plt

"""
Explanation: Importing libraries
The following imports are for making a quick inline plot at the end
End of explanation
"""


from pyspark.sql.functions import col

"""
Explanation: This imports the col 
function which is used later to get the column object for something that is calculated as part
of the pipeline and does not yet have a name that can be refered to.
End of explanation
"""


# Confirm we've started SPARK and created a SQL context that can be used
sqlContext

idb_df = sqlContext.read.parquet("/guoda/data/idigbio-20190612T171757.parquet")

idb_df.count()

checklists = (idb_df
              .select(idb_df.countrycode, idb_df.scientificname)
              .filter(idb_df.countrycode != "")
              .groupBy(idb_df.countrycode, idb_df.scientificname)
              .count()
              )

print(checklists.count())
#checklists.head(10)

checklists.head(10)

"""
Explanation: Reading in the iDigBio data frame
There is a parquet file stored on the HDFS file system used by the Spark cluster that already has a copy of iDigBio data in it. This is a testing export and does not have all of the columns from iDigBio but it has a lot of the commonly used ones.
End of explanation
"""


arthropods = (idb_df
              .filter(idb_df.phylum == "arthropoda")
              .filter("countrycode IN ('usa', 'can', 'mex')")
              )
arthropods.count()

"""
Explanation: Filtering and summarizing data
All the iDigBio data is first filtered to only those records related to arthopods. Notice that there are two syntaxes for the filter function and each line uses one of them. The SQL-style IN clause was an easier way of expressing the country codes than a set of equality and or operations.
End of explanation
"""


species_counts = (arthropods
                  .groupBy(arthropods.scientificname)
                  .count()
                  .orderBy(col("count"), ascending=False)
                  )

print(species_counts.count())
species_counts.head(10)

"""
Explanation: The syntax for summarizing data also resembles SQL. (In fact, all this can be done with an actual SQL statement using registerTempTable() and sqlContext.sql() if you wanted to work only with SQL syntax.) Notice the use of the col() function to get the newly calculated column of counts. This is needed because there is no column arthropods.counts, the column only exists in the context of the calulation before the assignment to species_counts happens.
End of explanation
"""


top_species_for_graph = (species_counts
                         .filter("count > 500")
                         .toPandas()
                         )
print(top_species_for_graph.shape)

ax = plt.bar(left=top_species_for_graph.index,
            height=top_species_for_graph["count"])

"""
Explanation: Graph the distribution
To make a quick graph, the Spark dataframe needs to be converted to a structure that graphing libraries like matplotlib understand. The topPandas() method of a dataframe will convert the data to a Pandas dataframe but this is a single object in the memory of the notebook server, it is not on the Spark cluster. Additionally, graphing more than 10,000 things is unreadable so be very sure to cut down the size of the dataframe before converting to Pandas. Here only species with more than 100 occurences are graphed.
End of explanation
"""


(species_counts
 .write
 .format("com.databricks.spark.csv")
 .mode("overwrite")
 .option("header", "true")
 .save("/outputs/species_counts.csv")
)

"""
Explanation: Save the list
Finally, save the full dataframe out as a CSV file for other uses. Right now this writes the file to the HDFS cluster for which there is no user access. Please email mcollins@acis.ufl.edu to get a copy of your files.
End of explanation
"""


