import re
idb_df_version = "20161119"  # Hardcoded version of the idb parquet to use

df = sqlContext.read.load("/guoda/data/idigbio-{0}.parquet".format(idb_df_version))

"""
Explanation: Unique Value Code
With the ES generated iDigBio dataframe, build a text list of all the unique values in each field.
End of explanation
"""


small_df = (df
            .where(df["stateprovince"] == "vermont")
            .where(df["genus"] == "acer")
            )
print(small_df.count())
small_df.printSchema()

"""
Explanation: Small subset
Start by making a small selection to work with
End of explanation
"""


fields = ["stateprovince", "specificepithet"]
#, "data.dwc:specificepithet"]

p = re.compile('[\W_]+')
for field in fields:
    slug = p.sub("_", field)
    output_fn = "idigbio-{0}-unique-{1}".format(idb_df_version, slug)
    (small_df
     .groupBy(df[field])
     .count()
     .write
     .format("com.databricks.spark.csv")
     .mode("overwrite")
     .option("header", "false")
     .save("/outputs/{0}.csv".format(output_fn))
    )

"""
Explanation: Test set of fields to do this for
Set up the fields to iterate over to generate a file per field. Hard code for now. Just a few to make sure the summarization process works.
End of explanation
"""


field_set = set()
for s in small_df.schema:
    #print(s.dataType)
    #if str(s.dataType) in ["StringType", "FloatType", "TimestampType", "InegertType", "BooleanType", "DoubleType"]:
    if not str(s.dataType).startswith("StructType"):
        field_set.add(s.name)
    else:
        for sub in s.dataType:
            field_set.add(".".join([s.name, sub.name]))
print(field_set)
print(len(field_set))

"""
Explanation: Looks like the nested fields are pretty killer. They take a lot longer than the top level ones and it GC memory killed Spark in this notebook even for data.dwc:genus which should be smallish (50k).
Changed small_df to acer in vermont instead of vermont and the test summary to specificepithet.
Doesn't seem to matter, eats all memory even when the small dataset is 477 records.
Building field list to iterate over
Now build up the full list of fields, in corperate this into the real job
End of explanation
"""


