<jupyter_start><jupyter_text>pysparking_musiXmatch_project_with_commentsAbstractI investigate the presence of different languages in the musiXmatch dataset and K-means [1] ability to discover these. I automate the language labeling of the 240k songs. A feature representation is chosen, the most infrequent languages are removed. I chose a measure for clustering and develop a sense for what a good clustering is in terms of this measure by employing supervised learning and training a classifier on the data. I then initialize K-means in three ways and observe how well it clusters according to language. The third way of initialization can be seen as approximating semi-supervised learning. Implementation is with python and PySpark and presented as jupyter notebook. Introduction and problem formulationThe musiXmatch dataset consists of circa 240k song with lyrics in BOW-format. When examining this data set it caught my eye that, contrary to my expectation, not all words were English! I could easily pick out some Spanish, French, Italian, Swedish and Danish words and I suspected there were more. It also struck me as a very natural assumption that the lyrics of the songs would be predominately made up of words from one language. I felt it would be an interesting task to see how well a standard clustering algorithm such as K-means would pick out this particular latent structure without me telling it to. I also wanted to see if K-means would pick out more of the structure I was interested in and ignore other structures that surely must be there, especially within songs in the dominant language of English, if I managed to initialize it appropriately. So this task I set for myself: investigate the language structure of musiXmatch and how well K-means can discover it. setting some paths for data resources and a space to save intermediate results<jupyter_code>## First init some useful path and get some files :

# download some data from http://millionsongdataset.com/musixmatch/
# namely train file "mxm_dataset_train.txt", test file "mxm_dataset_test.txt" and
# "mxm_reverse_mapping.txt" found here http://millionsongdataset.com/sites/default/files/mxm_reverse_mapping.txt

dataPath="c:/home/Examining-musiXmatch-languages-clustering-with-PySpark/data/"

# the repo also has this which indicates that it is auxiliary info or pre-calculated results data
xtraDataPath=dataPath

aFileTst="mxm_dataset_test.txt"
aFileTrain="mxm_dataset_train.txt"<jupyter_output><empty_output><jupyter_text>start up pyspark<jupyter_code>import pyspark
from pyspark.sql import SparkSession
import os

#start up

conf = pyspark.SparkConf().setAll([('spark.executor.memory', '16g'),('spark.driver.memory','16g'),('spark.driver.host','127.0.0.1')])
sc = pyspark.SparkContext(master='local', appName='myAppName',conf=conf)
spark = SparkSession(sc)


numPart=3*os.cpu_count()#4 #3 times num (logical) processors on computer,

#rdd's for test and train:

raw_data_train = sc.textFile(dataPath+aFileTrain,numPart)
raw_data_test = sc.textFile(dataPath+aFileTst,numPart)<jupyter_output><empty_output><jupyter_text>ImplementationEach song is represented with the help of words from a 5k word long dictionary – the 5k most frequent words – which incidentally may mean that non-english songs are represented by fewer words. As it would be too much work and also because I don’t speak every language in the world I let a, mostly, reliable tool “googletrans” tell me language label for each of the 5k words. I wrote code to interpret the answers from the tool. Worth mentioning is that I did NOT, as is often done when building NLP solutions, remove the most frequent and most infrequent words as they would carry useful information concerning language. Next step was to use the assumption mentioned before about words of a song belonging predominately to one language and let a majority vote decide the language label of the songs. After this I removed about 30 songs with infrequent language labels. I created a feature representation where each unique word present in a song gave rise to a value 1/”num unique words in song” at the corresponding index of a 5k long feature vector. Because of memory and computational restraints and because only a fraction of the 5k words ever actually show up in any given song I used a sparse feature representation. I built a simple linear model (multinomial logistic regression [2]) to get an upper limit on what would be reasonable to expect from an unsupervised method like Kmeans. I measured performance in terms of adjusted mutual information (AMI). I then initialized K-means in the old fashioned random way, trained 20 times and recorded AMI. There after I did the same thing using the k-means++ initialization [3]. Finally; I found the average feature vector for each language and initialized with that then ran k-means and found the AMI. haveing a first look at the data and reversing the word stemming as well as is possible<jupyter_code>#Featue construction:

#get the wordlist:
word_rows = raw_data_test.filter(lambda x: '%' in x).map(lambda x: x.split(","))

wordList=word_rows.collect()[1]
wordList[0]=wordList[0][1:] #remove the "%"

numWords=len(wordList)
print(len(wordList))

#+1 because musicXmatch data is indexed starting from 1
wordListWithInd=[(i+1,w) for i,w in enumerate(wordList)]
print(wordListWithInd)


import pandas as pd

#to remove the normalization of the 5k words/"clean" them:
fname=dataPath+"mxm_reverse_mapping.txt" #found this here: http://millionsongdataset.com/sites/default/files/mxm_reverse_mapping.txt
dfRevMap=pd.read_csv(fname,sep='<SEP>',names=['normalized','Not Normalized'])
revNormDict={n:un for n,un in zip(dfRevMap['normalized'],dfRevMap['Not Normalized'] )}
wordListClean = [revNormDict.get(w,w) for w in wordList]<jupyter_output>5000
[(1, 'i'), (2, 'the'), (3, 'you'), (4, 'to'), (5, 'and'), (6, 'a'), (7, 'me'), (8, 'it'), (9, 'not'), (10, 'in'), (11, 'my'), (12, 'is'), (13, 'of'), (14, 'your'), (15, 'that'), (16, 'do'), (17, 'on'), (18, 'are'), (19, 'we'), (20, 'am'), (21, 'will'), (22, 'all'), (23, 'for'), (24, 'no'), (25, 'be'), (26, 'have'), (27, 'love'), (28, 'so'), (29, 'know'), (30, 'this'), (31, 'but'), (32, 'with'), (33, 'what'), (34, 'just'), (35, 'when'), (36, 'like'), (37, 'now'), (38, 'que'), (39, 'time'), (40, 'can'), (41, 'come'), (42, 'de'), (43, 'there'), (44, 'go'), (45, 'up'), (46, 'oh'), (47, 'la'), (48, 'one'), (49, 'they'), (50, 'out'), (51, 'down'), (52, 'get'), (53, 'she'), (54, 'was'), (55, 'see'), (56, 'if'), (57, 'got'), (58, 'never'), (59, 'from'), (60, 'he'), (61, 'feel'), (62, 'want'), (63, 'let'), (64, 'make'), (65, 'way'), (66, 'say'), (67, 'take'), (68, 'would'), (69, 'as'), (70, 'ca'), (71, 'day'), (72, 'at'), (73, 'babi'), (74, 'away'), (75, 'life'), (76, 'yeah'), (77, 'y'), ([...]<jupyter_text>asking google translate about language label for each of the 5k words. Then saveing so I won't need to redo this as it takes a long time (google limits number of requests per timeperiod)<jupyter_code>from time import time, sleep
from pprint import pprint

#remove this comment if you want to wait for googles answer about language for 5k words
#otherwise just load it in the next cell!
'''
#now get google to tell me which language the words come from:
import googletrans
gt=googletrans.Translator()

t0=time()
wordListWithIndAndTransInfo=[]
failCount=0
for i,w in enumerate(wordListClean):
    print(i)
    if failCount>5:#if google thinks im too heavy
        break
    sleep(2)#to not let google think im too heavy
    try:
        res=gt.detect(w)
        wordListWithIndAndTransInfo.append([i+1,w,res.lang,res.confidence])
        failCount=0
    except:
        wordListWithIndAndTransInfo.append([i+1,w,'?',0])
        failCount+=1

t1=time()
pprint(wordListWithIndAndTransInfo)
pprint(t1-t0)

fname=xtraDataPath+"RESLang5000WordsClean.txt"
with open(fname, "w+", encoding="utf-8") as f:
    for k in range(len(wordListWithIndAndTransInfo)):
        str2Write=str(wordListWithIndAndTransInfo[k][0])+','\
        +wordListWithIndAndTransInfo[k][1]+','\
        +wordListWithIndAndTransInfo[k][2]+','\
        +str(wordListWithIndAndTransInfo[k][3])+'\n'
        f.write(str2Write)
'''<jupyter_output><empty_output><jupyter_text>load and take a look at google translate answers<jupyter_code>#get the list of languages used,
#see that some are merged - need to un-merge!,
fname=xtraDataPath+"RESLang5000WordsClean.txt"
wordDFWithIndAndTransInfo=pd.read_csv(fname,sep=',',names=['wordIndex','word','rawLang','conf'])

#want to change \zh-CH\ which google translate give me to \zh\ which is an ISO639-1 code,
rawUsedLangSet=set(wordDFWithIndAndTransInfo['rawLang'])
rawUsedLangSet<jupyter_output><empty_output><jupyter_text>will re-interprete answers according to following table:<jupyter_code>#from https://cloud.google.com/translate/docs/languages
#needed to edit manually because never get "jv" from google translate but I do get "jw"
fname2=xtraDataPath+"GT_Language_support_edited.txt"
GTlangCodes639_1PLUS=pd.read_csv(fname2,sep='\t')
GTlangCodes639_1PLUS



authorativeLangSet=set(GTlangCodes639_1PLUS['ISO-639-1_Code'])
numLangTot=len(authorativeLangSet)
numLangTot<jupyter_output><empty_output><jupyter_text>resolve situation when answer consists of 2 languages:<jupyter_code>#googletrans sometimes spits back 2 suggestions concatenated - need to break into 2 pieces!

for rlnum, rawLang in enumerate(rawUsedLangSet):
    if authorativeLangSet.intersection([rawLang])==set():
        print('didnt find '+rawLang)
    else:
        print('found '+rawLang)
#will assume a maximum of 2 langs per word!

#breaking into languages (when necessary) here:

for rlnum, rawLang in enumerate(rawUsedLangSet):
    if authorativeLangSet.intersection([rawLang])==set():
        print('didnt find '+rawLang)
        matchRes=[(w1,w2) for i1,w1 in enumerate(list(authorativeLangSet)[:-1]) \
                  for w2 in list(authorativeLangSet)[i1:] \
                  if (w1+w2==rawLang) or (w2+w1==rawLang)]
        print('but there is a match: '+matchRes[0][0]+','+matchRes[0][1])
        if len(matchRes)>1:
            print('MORE THAN ONE MATCH!!!!')
    else:
        print('found '+rawLang)

lang2SplitDict={}
for _, rawLang in enumerate(rawUsedLangSet):
    if authorativeLangSet.intersection([rawLang])==set():
        matchRes=[(w1,w2) for i1,w1 in enumerate(list(authorativeLangSet)[:-1]) \
                  for w2 in list(authorativeLangSet)[i1:] \
                  if (w1+w2==rawLang) or (w2+w1==rawLang)]
        lang2SplitDict.update({rawLang:(matchRes[0][0],matchRes[0][1])})
    else:
        lang2SplitDict.update({rawLang:(rawLang,None)})

#an example:
print('\n')
pprint('jazh-CN gets split like so:')
pprint(lang2SplitDict['jazh-CN'])<jupyter_output>didnt find dear
but there is a match: ar,de
didnt find ites
but there is a match: it,es
found tr
didnt find trzh-CN
but there is a match: tr,zh-CN
didnt find dano
but there is a match: da,no
didnt find esit
but there is a match: it,es
didnt find arzh-CN
but there is a match: ar,zh-CN
found ro
found sl
found so
didnt find ptfr
but there is a match: pt,fr
found sq
didnt find dasv
but there is a match: da,sv
didnt find esgl
but there is a match: gl,es
didnt find frde
but there is a match: fr,de
didnt find itbg
but there is a match: it,bg
found ar
didnt find esbg
but there is a match: bg,es
didnt find sktl
but there is a match: tl,sk
found da
found af
didnt find nlde
but there is a match: de,nl
didnt find itfr
but there is a match: it,fr
found en
found bs
didnt find svno
but there is a match: sv,no
found uz
didnt find dezh-CN
but there is a match: zh-CN,de
found sv
didnt find esar
but there is a match: ar,es
didnt find ropt
but there is a match: ro,pt
didnt find esfr
but there is a match: [...]<jupyter_text>creating a few dicts that will come in handy later<jupyter_code>#some convenience-stuff:

wordDFWithIndAndTransInfo['lang1']=\
[lang2SplitDict[w][0] for w in wordDFWithIndAndTransInfo['rawLang']]

wordDFWithIndAndTransInfo['lang2']=\
[lang2SplitDict[w][1] for w in wordDFWithIndAndTransInfo['rawLang']]

langCode2numDict={c:i for i,c in enumerate(GTlangCodes639_1PLUS['ISO-639-1_Code']) }
len([i for i,(w1,w2) in enumerate(zip(wordDFWithIndAndTransInfo['lang1'],wordDFWithIndAndTransInfo['lang2'])) if w1=='en' or w2=='en'])

langNum2CodeDict={i:c for i,c in enumerate(GTlangCodes639_1PLUS['ISO-639-1_Code']) }


wordDFWithIndAndTransInfo['langCode1']=\
[langCode2numDict[lang2SplitDict[w][0]] for w in wordDFWithIndAndTransInfo['rawLang']]

wordDFWithIndAndTransInfo['langCode2']=\
[langCode2numDict[lang2SplitDict[w][1]] if lang2SplitDict[w][1]!=None else None for w in wordDFWithIndAndTransInfo['rawLang'] ]


langCode2FullLang={c:l for c,l in zip(GTlangCodes639_1PLUS['ISO-639-1_Code'],GTlangCodes639_1PLUS['Language '])}<jupyter_output><empty_output><jupyter_text>will read data files in a parallell way extracting song info with the help of this function<jupyter_code>#now start looking at songs:

#make these broadcasted: (hopeing for efficiency)
bvl1 = sc.broadcast(list(wordDFWithIndAndTransInfo['lang1']))
bvl2 = sc.broadcast(list(wordDFWithIndAndTransInfo['lang2']))
bvc = sc.broadcast(list(wordDFWithIndAndTransInfo['conf']))
bvlc1n = sc.broadcast(list(wordDFWithIndAndTransInfo['langCode1']))
bvlc2n = sc.broadcast(list(wordDFWithIndAndTransInfo['langCode2']))

#this will be used to extract data from each song:
def splitIntoPartsFN(nonSplit):
    # meant to be unleashed on a record of a songs 2 track id's and sparse representation of
    # the songs word count
    #
    #make a tuple of 2 id's
    #and wordIDs and frequencies
    firstSplit=[ko for ki in list(map(lambda x:x.split(","),[nonSplit])) for ko in ki]
    secondSplit=[ko for ki in list(map(lambda x:x.split(":"),firstSplit)) for ko in ki]
    id1,id2,wordId,wordCount=secondSplit[0],secondSplit[1],secondSplit[slice(2,None,2)],secondSplit[slice(3,None,2)]
    return (id1,id2,wordId,wordCount)
    
'''
#TR_data = raw_data_test.filter(lambda x: 'TR' in x).map(splitIntoPartsFN)
# mds tracks-ids all start "TR":
TR_data = raw_data_test.filter(lambda x: re.search('^TR.*', x)).map(lambda x: x.split(","))

#lookHere_rows = TR_data.collect(),
lookHere_rows = TR_data.take(3)
pprint(lookHere_rows)
#type(lookHere_rows),
#type(TR_data),
type(lookHere_rows)
type(lookHere_rows[1])
type(lookHere_rows[1][2])
pprint(lookHere_rows[1][2])
'''<jupyter_output><empty_output><jupyter_text>make use of google translates confidence in its translation in order to get as good a language label for songs as possible; accumulating weights for several language hypothesis. This cell does the job<jupyter_code>from pyspark.mllib.linalg import Vectors
import numpy as np

#accumulate confidences for languages per song
def getAccumLangConf(listInd):
    #get the languages and their coresponging confidence values
    #remember that there can be 2 languages that need to share
    #i-1 because song indices start from 1
    langConf = [(bvlc1n.value[i-1],bvc.value[i-1]/(2 if bvl2.value[i-1]==None else 1)) for i in listInd]+\
    [(int(bvlc2n.value[i-1]),bvc.value[i-1]/2) for i in listInd if bvl2.value[i-1]!=None]
    #add up the language confidences/weights
    accumLangConf=np.zeros(numLangTot)
    for k,v in langConf:
        accumLangConf[k]+=v

    return accumLangConf

#create sparse vectors
def weightVectors(nonSplit):
    #I ignore frequencies of constituents - only care about presence,
    #extract word indices:
    splitInfo=splitIntoPartsFN(nonSplit)
    listSingleInd = [int(s) for s in splitInfo[2]]

    accLanCon = getAccumLangConf(listSingleInd)

    return Vectors.sparse(numLangTot, {key:val for key,val in [(k,v) for k,v in enumerate(accLanCon) if v>0]})

#import pandas as pd
import re

wv_data = raw_data_train.filter(lambda x: re.search('^TR.*', x)).map(weightVectors).filter(lambda x:len(x.values)>0)
vec_lang_sparse=sc.runJob(wv_data, lambda part: [x for x in part])

#winningNums=[ve.indices[np.argmax(ve.values)] for i,ve in enumerate(vec_lang_sparse) if len(ve.values)>0]
#winningNums=np.asarray([ve.indices[np.argmax(ve.values)] for i,ve in enumerate(vec_lang_sparse) if len(ve.values)>0])
winningNums=np.asarray([ve.indices[np.argmax(ve.values)] for ve in vec_lang_sparse if len(ve.values)>0])

'''
#writing down winning langs to file:
fname=xtraDataPath+"lyricsLangTrain.txt"
with open(fname, "w+", encoding="utf-8") as f:
    for k in range(len(winningNums)):
        str2Write=str(winningNums[k])+','\
        +langNum2CodeDict[winningNums[k]]+'\n'
        f.write(str2Write)
'''

#investigating if the winner has competition by dividing second biggest by biggest vote
#essentially testing the assumption that whole songs tend to be in one language:

secondByfirst=np.asarray([oneandtwo[0]/(1e-6+oneandtwo[1]) for oneandtwo in [np.partition(ve,-2)[-2:] for ve in vec_lang_sparse if len(ve.values)>0]])<jupyter_output><empty_output><jupyter_text>EvaluationIllustration of language frequencies in the data:<jupyter_code>from matplotlib import pyplot as plt

n_bins = 20

fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)

ax.hist(secondByfirst, bins=n_bins)
#ax.hist(secondByfirst, bins=n_bins,cumulative=True)
#ax.set_yscale('log')
plt.show()
#plt.savefig(xtraDataPath+'2nd by 1st histogram.png')<jupyter_output><empty_output><jupyter_text>Fig1. The majority vote procedure used to assign language to song is well motivated - it is based on the assumption that lyrics tend to be in one language only. This plot shows second biggest vote divided by biggest vote for all songs in the training data and as can be seen the biggest vote tends to be much bigger.<jupyter_code>#investigating if the winner has competition by dividing second biggest by biggest vote
#also; removing English which is the dominant language from the statistics
#to see what happens
#essentially testing the assumption that whole songs tend to be in one language:

secondByfirst_nonEnglish=np.asarray([oneandtwo[0]/(1e-6+oneandtwo[1]) for oneandtwo in [np.partition(ve,-2)[-2:] for ve in vec_lang_sparse if len(ve.values)>0 and np.argmax(ve)!=20]])

n_bins = 20

fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)

ax.hist(secondByfirst_nonEnglish, bins=n_bins)
#ax.hist(secondByfirst, bins=n_bins,cumulative=True)
#ax.set_yscale('log')
plt.savefig(xtraDataPath+'2nd by 1st histogram non-English.png')
plt.show()
#plt.savefig(xtraDataPath+'2nd by 1st histogram non-English.png')<jupyter_output><empty_output><jupyter_text>Fig2. This plot shows second biggest vote divided by biggest vote but excluding those where English – the dominnt language with the most words in the dictionary. As can be seen the biggest vote still tends to be significantly bigger than the second biggest.<jupyter_code>#Motivating a removal of infrequent languages:

#this many of each:
tots=[]
langsUsed=[]
for i in list(set(winningNums)):
     print(i)
     print(langNum2CodeDict[i])
     print(langCode2FullLang[langNum2CodeDict[i]])
     print((winningNums==i).sum())
     tots.append((winningNums==i).sum())
     langsUsed.append(langCode2FullLang[langNum2CodeDict[i]])
     print('     ')

#a little visualization:

from matplotlib import pyplot as plt

'''
fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
ax.plot(np.asarray(list(range(len(tots)))),np.asarray(tots), color='blue', lw=2)
ax.plot(np.asarray(list(range(len(tots)))),np.asarray(tots), 'r*')
ax.set_yscale('log')
for i,(t,l) in enumerate(zip(tots,langsUsed)):
    ax.annotate(l, xy=(i,t)) # <--
plt.show()
'''
fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
ax.bar(np.arange(len(tots))-0.5,np.asarray(tots), 1)
ax.set_yscale('log')
for i,(t,l) in enumerate(zip(tots,langsUsed)):
    ax.annotate(l, xy=(i-1,t*1.03)) # <--
#plt.savefig(xtraDataPath+'language histogram.png')
plt.show()<jupyter_output><empty_output><jupyter_text>Fig3. English dominates but a few other languages make significant contributions to the data set<jupyter_code>#thres=10
#I will choose to remove languages with less than this many representatives:
thres=8
lInds2use=[i for i in list(set(winningNums)) if (winningNums==i).sum()>=thres]
numClasses2use=len(lInds2use)

labNDict={v:i for i,v in enumerate(lInds2use)}
labNDictInv={v:k for k,v in zip(labNDict.keys(),labNDict.values())}

#these remain:
for i in range(numClasses2use):
    print(langNum2CodeDict[labNDictInv[i]])


#set([labNDict.get(v,-1) for v in winningNums])

lInds2useLOOKUP = [labNDict.get(i,-1) for i in range(numLangTot)]
#for more efficient use:
indLU = sc.broadcast(lInds2useLOOKUP)


#this many in the train set before weeding out less frequent languages
hlp=list(set(winningNums))
hlp.sort()
ws=[(i,(winningNums==i).sum()) for i in hlp]
np.asarray([s[1] for s in ws]).sum()

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.linalg import SparseVector

#function I need:
def parseLabPs(nonSplit):
    #I ignore frequencies of constituents - only care about presence,
    #extract word indices:
    splitInfo=splitIntoPartsFN(nonSplit)
    listSingleInd = [int(s) for s in splitInfo[2]]
    #get the languages and their coresponging confidence values
    #remember that there can be 2 languages that need to share
    accLanCon = getAccumLangConf(listSingleInd)

    #encode sparsely accumulated confidence per language
    accLanConMax=indLU.value[int(np.argmax(accLanCon))] if accLanCon.sum()>0 else -1
    return LabeledPoint(accLanConMax, SparseVector(numWords, [i-1 for i in listSingleInd], len(listSingleInd)*[1.0/len(listSingleInd)]))

#now parsing out the ones that correspong to big enough languages:
parsedDataTrain = raw_data_train.filter(lambda x: re.search('^TR.*', x)).map(parseLabPs).filter(lambda x:x.label>-1)
labp=sc.runJob(parsedDataTrain, lambda part: [x for x in part])
winningNums2=np.asarray([int(l.label) for l in labp])

#this many data after removing infrequent languages
hlp2=list(set(winningNums2))
hlp2.sort()
ws2=[(i,(winningNums2==i).sum()) for i in hlp2]
np.asarray([s[1] for s in ws2]).sum()


#look which remain spelled out properly:

for i in range(numClasses2use):
    print(langCode2FullLang[langNum2CodeDict[labNDictInv[i]]])

'''
fname=xtraDataPath+"MostFreqLangs2Use.txt"
with open(fname, "w+", encoding="utf-8") as f:
    for i in range(numClasses2use):
        str2Write=str(i)+','+langCode2FullLang[langNum2CodeDict[labNDictInv[i]]]+'\n'
        f.write(str2Write)
'''

#now it's time to make a simple supervised model
# to use to get a sense of how good a simple model can be
#at saying song language
#later I'll cluster the same data - how well will it
#find this same structure that the simple logistic regression finds?
from pyspark.mllib.classification import LogisticRegressionWithLBFGS, LogisticRegressionModel

parsedDataTest = raw_data_test.filter(lambda x: re.search('^TR.*', x)).map(parseLabPs).filter(lambda x:x.label >-1)
#parsedDataTrain = raw_data_train.filter(lambda x: re.search('^TR.*', x)).map(parseLabPs).filter(lambda x:x.label>-1)

parsedDataTrain.cache()
# Build the model - will use a little regularization
#to avoid overfitting. 0.1 seems remove most overfitting
#as error is similar size on train and test sets

modellogregregu = LogisticRegressionWithLBFGS.train(parsedDataTrain,numClasses=numClasses2use,regType='l2',regParam=0.1)

# Evaluating the model on training data
labelsAndPreds = parsedDataTrain.map(lambda p: (p.label, modellogregregu.predict(p.features)))
trainErr = labelsAndPreds.filter(lambda lp: lp[0] != lp[1]).count() / float(parsedDataTrain.count())
print("Training Error = " + str(trainErr))

#=0.0
#Training Error = 0.003311305471492843 when regul

#and now testset:

parsedDataTest.cache()
# Evaluating the model on test data
labelsAndPreds = parsedDataTest.map(lambda p: (p.label, modellogregregu.predict(p.features)))
testErr = labelsAndPreds.filter(lambda lp: lp[0] != lp[1]).count() / float(parsedDataTest.count())
print("Test Error = " + str(testErr))
#Out[9]: 0.006668877344239343
#Test Error = 0.00467926752883092 #regul


#let's have a look at which languages get confused with each other:
labNpredInds2LinInd = lambda l,p,ncl:ncl*l+p
linInd2labNpredInds = lambda li,ncl:(int(np.floor(li/ncl)),np.mod(li,ncl))
lpElems = labelsAndPreds.groupBy(lambda lp: labNpredInds2LinInd(lp[0],lp[1],numClasses2use))#.count() #/ float(parsedDataTest.count())

confList=[]
for i,cf in enumerate(lpElems.collect()):
    confList.append((len([c for c in cf[1]]),linInd2labNpredInds(int(cf[0]),numClasses2use)))

for cfl in confList:
    print(str(cfl[0])+\
    #print(str(cfl[0]/nexInTest)+\
    '  '+langCode2FullLang[langNum2CodeDict[labNDictInv[cfl[1][0]]]]+\
    'to '+langCode2FullLang[langNum2CodeDict[labNDictInv[cfl[1][1]]]])

#English is by far the largest language and most errors
#are between English and another language
#besides that there are a few confusions between the latin language
#Spanish, Portugeese, Italian and French which is not surprising
#given their similarity
#+1 polish and 1 german to spanish (the 2nd largest language)
#now given this great supervised model which will be challenging
#to match with a clustering algorithm
#how high does it get on a clustering measure?
from sklearn import metrics

ls_tst = parsedDataTest.map(lambda p: p.label).collect()
ps_tst = parsedDataTest.map(lambda p: modellogregregu.predict(p.features)).collect()

metrics.adjusted_mutual_info_score(ls_tst,ps_tst)
#Out[19]: 0.9393770913689595
#Out[7]: 0.9389128244827115
#Out[10]: 0.9365355912049723
# Out[28]: 0.9365355912049723
# Out[28]: 0.9573686078169512 #regul
#Out[46]: 0.9584682946581857 #regul

#so if clustering ever manages 0.95 then it  is really doing a GREAT job!
#realistically; how close can it get?

ls_tr = parsedDataTrain.map(lambda p: p.label).collect()
#ps_tr = parsedDataTrain.map(lambda p: modellogreg.predict(p.features)).collect()
ps_tr = parsedDataTrain.map(lambda p: modellogregregu.predict(p.features)).collect()

metrics.adjusted_mutual_info_score(ls_tr,ps_tr)
#Out[19]: 1.0
#Out[35]: 0.9666974988900423 #regu
#Out[35]: 0.9664790935842319 #regu


from pyspark.mllib.clustering import KMeans, KMeansModel

#now peel off the class and use the feature only:
featuresTrain = parsedDataTrain.map(lambda p: p.features)
featuresTrain.cache()

# Build the model (cluster the data)
#doing old fashioned kmeans with no special init first:
clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initializationMode="random")
#clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initialModel = KMeansModel([tuple(centDict[i]) for i in centDict]) )


#evaluate clustering:
ls_tr = parsedDataTrain.map(lambda p: p.label).collect()
pcs_tr = parsedDataTrain.map(lambda p: clusters.predict(p.features)).collect()

metrics.adjusted_mutual_info_score(ls_tr,pcs_tr)
#Out[13]: 0.3670456909686498 #maxIterations=5, initializationMode="random")
#Out[16]: 0.3640903926470229 #maxIterations=10, initializationMode="random")
#Out[16]: 0.35939303923246807 #maxIterations=10, initializationMode="random")

#Out[15]: 0.3966480166188081 maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#Out[17]: 0.5481233780781543  maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#Out[13]: 0.7817683131779442  maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#Out[16]: 0.6577106923265921  maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#Out[36]: 0.41853894064265934  maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#Out[37]: 0.4104475203596721  maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)

#Out[37]: 0.8836244056125339 maxIterations=10, initialModel = KMeansModel([tuple(centDict[i]) for i in centDict])
#Out[37]: 0.8840288256682206 maxIterations=10, initialModel = KMeansModel([tuple(centDict[i]) for i in centDict])


#do kmean++ ;a better init!
clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
#clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initialModel = KMeansModel([tuple(centDict[i]) for i in centDict]) )
pcs_tr = parsedDataTrain.map(lambda p: clusters.predict(p.features)).collect()

metrics.adjusted_mutual_info_score(ls_tr,pcs_tr)

nRuns=20
resRandom=[]
for i in range(nRuns):
    print(i)
    clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initializationMode="random")
    ls_tr = parsedDataTrain.map(lambda p: p.label).collect()
    pcs_tr = parsedDataTrain.map(lambda p: clusters.predict(p.features)).collect()
    resRandom.append(metrics.adjusted_mutual_info_score(ls_tr,pcs_tr))

print(np.asarray(resRandom).min())
print(np.asarray(resRandom).mean())
print(np.asarray(resRandom).max())
print(np.percentile(np.asarray(resRandom),25))
print(np.percentile(np.asarray(resRandom),50))
print(np.percentile(np.asarray(resRandom),75))
print(np.asarray(resRandom).std())
'''
0.3316730542279839
0.38409542814286557
0.41446731401490056
0.369600456440365
0.3894647503075047
0.40070041241094534
0.022369174116231853
'''
nRuns=20
resPluPlu=[]
for i in range(nRuns):
    print(i)
    clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initializationMode= "k-means||", initializationSteps= 5)
    ls_tr = parsedDataTrain.map(lambda p: p.label).collect()
    pcs_tr = parsedDataTrain.map(lambda p: clusters.predict(p.features)).collect()
    resPluPlu.append(metrics.adjusted_mutual_info_score(ls_tr,pcs_tr))

print(np.asarray(resPluPlu).min())
print(np.asarray(resPluPlu).mean())
print(np.asarray(resPluPlu).max())
print(np.percentile(np.asarray(resPluPlu),25))
print(np.percentile(np.asarray(resPluPlu),50))
print(np.percentile(np.asarray(resPluPlu),75))
print(np.asarray(resPluPlu).std())
'''0
.33065672035506466
0.4780506842768837
0.7092094870396514
0.3925413633663136
0.4574918046878703
0.5401233077411065
0.10163666615298514
'''

#expect I can hijack "clusters-model" (clusters.centers[0].shape)
# and init centers (orclustercenters???) as i want
#"initialModel" input parameter
#initialModel = KMeansModel([(-1000.0,-1000.0),(5.0,5.0),(1000.0,1000.0)]))


def createCombiner(tpl):
    return (tpl[0], 1)

def mergeValue(accumulator, element):
    return (accumulator[0] + element[0], accumulator[1] + 1)

def mergeCombiner(accumulator1, accumulator2):
    return (accumulator1[0] + accumulator2[0], accumulator1[1] + accumulator2[1])

classWordCenters=parsedDataTrain.map(lambda p:(p.label,(Vectors.dense(p.features),0))).combineByKey(createCombiner, mergeValue, mergeCombiner).map(lambda t: (t[0], t[1][0]/t[1][1])).collect()

#seems this will work but not happy to make dense!!
#can i add sparse?

'''
fname=xtraDataPath+"MostFreqLangsCenteres.txt"
with open(fname, "w+", encoding="utf-8") as f:
    for i,cent in enumerate(classWordCenters):
        str2Write=str(int(cent[0]))#+','
        for elem in cent[1]:
            str2Write += ','+str(elem)#+','
        str2Write+='\n'
        f.write(str2Write)
'''
fname=xtraDataPath+"MostFreqLangsCenteres.txt"
centDict={}
with open(fname, "r") as readfile:
    for cnt,line in enumerate(readfile):
        print(cnt)
        hlp=re.split(',',line)
        centDict.update({int(hlp[0]):np.asarray([float(v) for v in hlp[1:]])})


clusters = KMeans.train(featuresTrain, numClasses2use, maxIterations=10, initialModel = KMeansModel([tuple(centDict[i]) for i in centDict]) )


#evaluate clustering:
ls_tr = parsedDataTrain.map(lambda p: p.label).collect()
pcs_tr = parsedDataTrain.map(lambda p: clusters.predict(p.features)).collect()

metrics.adjusted_mutual_info_score(ls_tr,pcs_tr)

sscc2save=np.zeros((len(clusters.centers),5000),dtype='float')
for k in range(len(clusters.centers)):
    sscc2save[k,:]=clusters.centers[k]

fname=xtraDataPath+"sscc.txt"
np.savetxt(fname,sscc2save)
#np.loadtxt(fname)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1268840692775784, 'fraction_numerical': 0.04116086753003589, 'mean_word_length': 5.863568215892054, 'pattern_counts': {'":': 2, '<': 58, '<?xml version=': 0, '>': 70, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '626754', 'license': 'Apache-2.0', 'n_tokens_mistral': 11796, 'n_tokens_neox': 10932, 'n_words': 3088, 'path': 'src/pysparking_musiXmatch_project_with_comments.ipynb', 'repo_name': 'bjornper/Examining-musiXmatch-languages-clustering-with-PySpark', 'size': '32045'}