<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
Step1:  Need a udf that returns a mapType of string
Step2:  Now let's write this out and go back and join to the main DF for some summaries
Step3:  How much more information might we be able to find in records that records that are not JSON parsable?
<ASSISTANT_TASK:>
Python Code:
def json_to_map(s):
    Convert a string containing JSON into a dictionary,
    Skip flattening for now.
    try:
        return json.loads(s)
    except:
        return {}
    
json_to_map_udf = udf(json_to_map, MapType(StringType(), StringType()))
print(json_to_map('{ "solr_long_lat": "-5.87403,30.49728", "related_record_types": "PreservedSpecimen|PreservedSpecimen", "related_record_links": "YPM-IP-530950|YPM-IP-530951" }'))
idb_map = (idb_dyn
 .withColumn("props_map", json_to_map_udf(col("props_str")))
)
idb_map.select(col("props_map")).show(10, truncate=False)
idb_triples = (idb_map
 .select(col("uuid"), 
         col("recordset"),
         col("institutioncode"),
         explode(col("props_map")).alias("key", "value"))
)
idb_triples.cache()
idb_triples.count()
idb_triples.show(20, truncate=False)
(idb_triples
 .groupBy(col("key"))
 .count()
 .sort(col("count"), ascending=False)
 .limit(1000)
 ).toPandas()
(idb_triples
 .groupBy(col("institutioncode"))
 .count()
 .sort(col("count"), ascending=False)
 .limit(1000)
 ).toPandas()
(idb_triples
 .filter(col("key") == "NSF_TCN")
 .count()
)
(idb_triples
 .filter(col("key") == "NSF_TCN")
 .groupBy(col("institutioncode"), col("value"))
 .count()
 .sort(col("count"), ascending=False)
 .limit(1000)
 ).toPandas()
Explanation: Need a udf that returns a mapType of string:string that explode will expand
End of explanation
#(idb_triples
# .write
# .parquet("/tmp/idigbio-20171014T023306-json-triples.parquet")
#)
Explanation: Now let's write this out and go back and join to the main DF for some summaries
End of explanation
(idb_triples
 .select(length(col("key")).alias("len_key"))
 .avg(col("len_key"))
 .show()
)
#joined = idb_dyn.join(idb_triples, idb_dyn["uuid"] == idb_triples["uuid"], "inner")
#joined.show(3, truncate=False)
#joined.count()
Explanation: How much more information might we be able to find in records that records that are not JSON parsable?
End of explanation
<END_TASK>