<jupyter_start><jupyter_text>Load metadata avro<jupyter_code>reader = sqlContext.read.format('com.databricks.spark.avro')
meta = reader.load('data/spark_metadata.avro')
meta.columns<jupyter_output><empty_output><jupyter_text>Loading topic distributions<jupyter_code>topdisFile = 'data/enron_small_topic_distributions.tuples'

csvLoader = sqlContext.read.format('com.databricks.spark.csv')
topdis = csvLoader.options(delimiter=',',header='false', inferschema='true').load(topdisFile)
topdis.columns
strip_first_col = udf(lambda row: int(row[1:]), IntegerType())
topdis = topdis.withColumn('C0',strip_first_col(topdis['C0']))<jupyter_output><empty_output><jupyter_text>Load dictionary CSV<jupyter_code>dicFile = 'enron_small_dic.csv'

csvLoader = sqlContext.read.format('com.databricks.spark.csv')
dic = csvLoader.options(delimiter='\t', header='false', inferschema='true').load(dicFile)
dic = dic.select(dic['C0'].alias('id'), dic['C1'].alias('word'), dic['C2'].alias('count'))
dic.columns<jupyter_output><empty_output><jupyter_text>Load clustertopics CSV<jupyter_code>clutoFile = 'enron_small_clustertopics.csv'

csvLoader = sqlContext.read.format('com.databricks.spark.csv')
cluto = csvLoader.options(delimiter=',', header='false', inferschema='true').load(clutoFile)
#dic = dic.select(dic['C0'].alias('id'), dic['C1'].alias('word'), dic['C2'].alias('count'))
cluto.columns<jupyter_output><empty_output><jupyter_text>Load topicswords CSV<jupyter_code>towoFile = 'enron_small_lda_transposed.csv'

csvLoader = sqlContext.read.format('com.databricks.spark.csv')
towo = csvLoader.options(delimiter=',', header='false', inferschema='true').load(towoFile)
#dic = dic.select(dic['C0'].alias('id'), dic['C1'].alias('word'), dic['C2'].alias('count'))'

towo.columns<jupyter_output><empty_output><jupyter_text>Merge topdis which has document id and with metadata, based on document id<jupyter_code>meta.columns
metasmall = meta.select('id',unix_timestamp(meta['date'],"yyyy-MM-dd'T'HH:mm:ssX").alias("timestamp"))
doctopdat = topdis.join(metasmall, metasmall.id == topdis.C0,'inner')
metasmall.show()
maxdate = doctopdat.select(max('timestamp').alias('maxtimestamp')).collect()[0]['maxtimestamp']
mindate = doctopdat.select(min('timestamp').alias('mintimestamp')).collect()[0]['mintimestamp']  
print maxdate, mindate

import numpy as np
binborders =map(int,list(np.linspace(start=mindate,stop=maxdate,num=30,endpoint=False)))

print binborders[-1]
#f = udf(lambda x:np.argmax(x < binborders) -1,IntegerType())
def f(x):
    for i in range(len(binborders)):
        if x < binborders[i]:            
            return binborders[i-1]
    return binborders[-1]
    
doctopdat = doctopdat.withColumn('bin',udf(f,LongType())(doctopdat['timestamp']))
newdataframe = doctopdat.groupBy('bin').avg()
newdataframe.take(1)
996179834 / (3600*24*7*52)
newdataframe.columns
print ((maxdate / (365*24.0*3600)) - (mindate / (365*24.0*3600))) * 12, "months"<jupyter_output>7.14959284627 months
{'fraction_non_alphanumeric': 0.1787878787878788, 'fraction_numerical': 0.024915824915824916, 'mean_word_length': 8.461783439490446, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '110814', 'license': 'Apache-2.0', 'n_tokens_mistral': 1154, 'n_tokens_neox': 1098, 'n_words': 133, 'path': 'Vincent spark exploration.ipynb', 'repo_name': 'nlesc-sherlock/corpora-vis', 'size': '2970'}