<jupyter_start><jupyter_text>Step 1: Read the data (the json files and metadata)  Read the metadata file<jupyter_code>metadata = spark.read\
            .format("csv")\
            .option("header", "true")\
            .load(home+"/Dataset/metadata.csv")<jupyter_output><empty_output><jupyter_text>We begin by randomly sampling 10,000 papers from the pdf_json folder. We can do this through the following linux command, utilizing the `shuf` command`~  shuf -zn10000 -e document_parses/pdf_json/* | xargs -0 cp -vt random_sample/`  Read the sample<jupyter_code>papers = spark.read\
            .format("json")\
            .option("multiLine", "true")\
            .load(home+"/Dataset/random_sample/")<jupyter_output><empty_output><jupyter_text>Step 2: Explore the data  From the metadata exploration we found that many columns has nulls and some are only nulls<jupyter_code>metadata.printSchema()
metadata.show(5)
from pyspark.sql.functions import isnan, when, count, col

metadata.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in metadata.columns]).show()<jupyter_output>+--------+-----+--------+-----+-----+-----+---------+-------+--------+------------+-------+-------+------+----------------+--------+--------------+--------------+-----+-----+
|cord_uid|  sha|source_x|title|  doi|pmcid|pubmed_id|license|abstract|publish_time|authors|journal|mag_id|who_covidence_id|arxiv_id|pdf_json_files|pmc_json_files|  url|s2_id|
+--------+-----+--------+-----+-----+-----+---------+-------+--------+------------+-------+-------+------+----------------+--------+--------------+--------------+-----+-----+
|       0|77267|       0|   30|29459|71159|    32790|     41|   29300|          44|   4995|   6651|133300|          113503|  132417|         75636|         87526|11453|28314|
+--------+-----+--------+-----+-----+-----+---------+-------+--------+------------+-------+-------+------+----------------+--------+--------------+--------------+-----+-----+<jupyter_text>The json files schema is available in the dataset where we found all the info we need<jupyter_code>papers.show(10)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
|            abstract|         back_matter|         bib_entries|           body_text|            metadata|            paper_id|         ref_entries|
+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
|                  []|[[[[48,, 44, 3.37...|[, [[[H, Adams, [...|[[[], [[79, TABRE...|[[[[,,], , , Bert...|2b1cbb43a4f06e232...|[[, Chest Wall Th...|
|[[[], [], Abstrac...|[[[], [], acknowl...|[[[], , [,,,], , ...|[[[[1203,, 1200, ...|[[[[,,], , ;, Bey...|c9f4a4df803f6496b...|[[, Figure 2, fig...|
|                  []|[[[[1609,, 1586, ...|[[[[T, Aaberg, [M...|[[[[505,, 497, Ka...|              [[], ]|5c6c26e79c0824645...|[,, [, Technische...|
|                  []|[[[], [], annex, ...|[[[[Organizationa...|[[[], [], , LEARN...|[[[[University of...|[...]<jupyter_text>Now we need to process the json files and get the info we need in a simple structured dataframe  Define a function to gather all paragraphs in one text and gather abstract paragraphs and body paragraphs<jupyter_code>from pyspark.sql.functions import udf

def concatenate_text(j):
    txt_all = ""
    for a in j:
        txt_all = txt_all + " " + a['text']
    return txt_all

udf_concatenate_text = udf(concatenate_text)
papers = papers.select(papers['Paper_ID'], papers['metadata']['title'], udf_concatenate_text(papers['abstract']), udf_concatenate_text(papers['body_text']))
papers.printSchema()
papers = papers.withColumnRenamed('metadata.title', 'title')\
        .withColumnRenamed('concatenate_text(abstract)', 'abstract')\
        .withColumnRenamed('concatenate_text(body_text)', 'body')
papers.show(10)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+
|            Paper_ID|               title|            abstract|                body|
+--------------------+--------------------+--------------------+--------------------+
|2b1cbb43a4f06e232...|Level 3 guideline...|                    | S3-guideline on ...|
|c9f4a4df803f6496b...|Physicians Poster...| Background: Prom...| with Thiotepa 5 ...|
|5c6c26e79c0824645...|                    |                    | Der Weltgesundhe...|
|38c0691ee76fb1e66...|Society of Genera...|                    | LEARNING OBJECTI...|
|67c8d389c28ceef26...|        Diedrich (1)|                    | The presence of ...|
|8c93a69ce8eb2ba08...|                    |                    | July-December 20...|
|e3b74d02ad582540a...|P1-A03A IgG Subty...| Background/Case ...| Background/Case ...|
|7e62ae259b327bda2...|                    |                    | Definitions. The...|
|a78f60c67c8ac17c7...|Alterations in 14...|           [...]<jupyter_text>Seems that some fields have empty values, let's explore that<jupyter_code>from pyspark.sql.functions import isnan, when, count, col

papers.select([count(when(col(c) == "", c)).alias(c) for c in papers.columns]).show()<jupyter_output>+--------+-----+--------+----+
|Paper_ID|title|abstract|body|
+--------+-----+--------+----+
|       0| 1060|    2927|   0|
+--------+-----+--------+----+<jupyter_text>Step 3: Prepare and process the data  Now join the papers dataframe the metadata and get only the columns that we might use as features<jupyter_code>papers_meta = papers.join(metadata, papers['Paper_ID'] == metadata['sha'], how='left_outer')\
            .select(papers['Paper_ID'], papers['title'], papers['body'], metadata['publish_time'],\
                   metadata['authors'], metadata['journal'])
from pyspark.sql.functions import year, month, to_date
papers_meta = papers_meta.withColumn("publish_Year", year(to_date("publish_time")))\
                        .withColumn("publish_Month", month(to_date("publish_time")))
papers_meta.show(5)<jupyter_output>+--------------------+--------------------+--------------------+------------+--------------------+--------------------+------------+-------------+
|            Paper_ID|               title|                body|publish_time|             authors|             journal|publish_Year|publish_Month|
+--------------------+--------------------+--------------------+------------+--------------------+--------------------+------------+-------------+
|0782b4fb23ca65baf...|The population ge...| The model will b...|  2015-11-30|Tibayrenc, Michel...|        Acta Tropica|        2015|           11|
|14d8fd027f39c8311...|Drug-Induced Dela...| Drug-induced del...|  2015-05-15|Klimas, Natasha; ...|Cutaneous Drug Er...|        2015|            5|
|1e5228e3f0658479a...|Canonicalizing Kn...| User-generated c...|  2020-04-17|Fatma, Nausheen; ...|Advances in Knowl...|        2020|            4|
|1e663ac169e08ea02...|The effects of a ...| Evidence has bee...|  2006-07-31|<NAME>.; F...|Journal of Affect...[...]<jupyter_text>Define a function to detect the paper language and filter out non-english papers<jupyter_code>from langdetect import detect
def detect_lang(txt):
    try:
        return detect(txt)
    except:
        return None
udf_detect_lang = udf(detect_lang)
papers_meta = papers_meta.withColumn('Lang', udf_detect_lang(papers_meta['body']))
## some papers are indeed non English
papers_meta.select("*").where("Lang<>'en'").show(5)
papers_meta.count()
papers_meta = papers_meta.filter(papers_meta['Lang'] == 'en')
papers_meta.count()
#Drop the unneeded columns
papers_meta = papers_meta.drop('publish_time', 'Lang')
papers_meta.show(5)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+--------------------+------------+-------------+
|            Paper_ID|               title|                body|             authors|             journal|publish_Year|publish_Month|
+--------------------+--------------------+--------------------+--------------------+--------------------+------------+-------------+
|0782b4fb23ca65baf...|The population ge...| The model will b...|Tibayrenc, Michel...|        Acta Tropica|        2015|           11|
|14d8fd027f39c8311...|Drug-Induced Dela...| Drug-induced del...|Klimas, Natasha; ...|Cutaneous Drug Er...|        2015|            5|
|1e5228e3f0658479a...|Canonicalizing Kn...| User-generated c...|Fatma, Nausheen; ...|Advances in Knowl...|        2020|            4|
|1e663ac169e08ea02...|The effects of a ...| Evidence has bee...|Yip, Paul S.F.; F...|Journal of Affect...|        2006|            7|
|1f26b5e8291ea1ddc...|The Case for Labo...| The field of pat..[...]<jupyter_text>We will periodically save the dataframes we are working on in parquet format with 12 partitions (the number of logical cores in our machine) for performance improvements.<jupyter_code>## Save what we have so far
papers_meta.repartition(12).write.parquet("./Dataset/papers_meta")
#papers_meta = spark.read\
#            .format("parquet")\
#            .option("header", "true")\
#            .load("./Dataset/papers_meta")<jupyter_output><empty_output><jupyter_text>Find which columns contain null or unknown values and replace them with " " for categorical and 0 for numerical  Count Nulls and Empty values in each column to understand more about the data<jupyter_code>from pyspark.sql.functions import isnan, when, count, col
print("Count of Null")
papers_meta.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in papers_meta.columns]).show()
print("Count of Empty values")
papers_meta.select([count(when(col(c) == "", c)).alias(c) for c in papers_meta.columns]).show()
cat_cols = [item[0] for item in papers_meta.dtypes if item[1].startswith('string')] 
cat_cols
from pyspark.sql.functions import col

cat_null_cols = [column for column in cat_cols if papers_meta.where(col(column).isNull()| col(column).isin('')).count() > 0]
cat_null_cols
### Now let's fill with " "
for column in cat_null_cols:
    papers_meta = papers_meta.na.fill(" ")
num_cols = [item[0] for item in papers_meta.dtypes if item[1].startswith('int') | item[1].startswith('double')] 
num_cols
### Now let's find numerical columns with null values
num_null_cols = [column for column in num_cols if papers_meta.filter(col(column).isNull() | col(column).eqNullSafe(0)).count() > 0]
num_null_cols
### Now let's fill with 0
for column in num_null_cols:
    papers_meta = papers_meta.na.fill(0)
papers_meta.show(10)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+--------------------+------------+-------------+
|            Paper_ID|               title|                body|             authors|             journal|publish_Year|publish_Month|
+--------------------+--------------------+--------------------+--------------------+--------------------+------------+-------------+
|012b16ae54779ca1a...|Synchronized Bive...| Miniaturized imp...|<NAME>; Jo...|             Sci Rep|        2020|            2|
|01bc7fe59fc7feb0e...|Open Forum Infect...| Acute upper resp...|<NAME>; ...|Open Forum Infect...|        2018|            2|
|0240a12c9fdf6c031...|Supersize me: how...| In epidemiology,...|Kao, Rowland R.; ...|Trends in Microbi...|        2014|            5|
|03bfb747583f6b214...|The Infant Nasoph...| The human microb...|                    |                    |           0|            0|
|06e9041ff3cb1db28...|Selected Nonvacci...| A cute respirato..[...]<jupyter_text>Now drop duplicates rows (No duplicates found in our sample)<jupyter_code>papers_meta.count()
papers_meta.dropDuplicates()
papers_meta.count()
## Save what we have so far
papers_meta.repartition(12).write.parquet("./Dataset/papers_meta_cleaned")
papers_meta = spark.read\
            .format("parquet")\
            .option("header", "true")\
            .load("./Dataset/papers_meta_cleaned")<jupyter_output><empty_output><jupyter_text>Step 4 (Preprocessing)  Now let's start the Processing phase for the papers body and title<jupyter_code>from pyspark.sql.functions import lower, regexp_replace
from pyspark.ml.feature import Tokenizer, StopWordsRemover<jupyter_output><empty_output><jupyter_text>Convert the body text to lower case Remove Punctuation<jupyter_code>from pyspark.sql.functions import col

papers_meta = papers_meta.withColumn("body", lower(col('body')))
papers_meta = papers_meta.withColumn("body", regexp_replace("body", "[^a-zA-Z\\s]" , " "))
papers_meta = papers_meta.withColumn("body", regexp_replace("body", " +" , " "))
papers_meta = papers_meta.withColumn("body", regexp_replace("body", "^ +" , ""))
papers_meta = papers_meta.withColumn("title", lower(col('title')))
papers_meta = papers_meta.withColumn("title", regexp_replace("title", "[^a-zA-Z\\s]" , " "))
papers_meta = papers_meta.withColumn("title", regexp_replace("title", " +" , " "))
papers_meta = papers_meta.withColumn("title", regexp_replace("title", "^ +" , ""))
papers_meta.show(10)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+------------+------------+-------------+
|            Paper_ID|               title|                body|             authors|     journal|publish_Year|publish_Month|
+--------------------+--------------------+--------------------+--------------------+------------+------------+-------------+
|002f213aeda7ce843...|etiology and impa...|community acquire...|<NAME>; A...|J Infect Dis|        2018|            7|
|0297dd12949520da3...|optimization of p...|the localization ...|Quintana, C.; Mar...|      Micron|        1998|            8|
|09045f5964f24691f...|evaluation of tnf...|the toxic effects...|R<NAME>...|  Biotherapy|        1991|            1|
|0e3da58a0d46d88ee...|the influence of ...|increasing our un...|                    |            |           0|            0|
|103c89c60d7d24bb8...|does pathogen spi...|pathogen outbreak...|Otterstatter, Mic...|    PLoS One|        2008|       [...]<jupyter_text>Tokenize the paper body text and title text<jupyter_code>tokenizer = Tokenizer(inputCol='body', outputCol='words_token')
papers_meta = tokenizer.transform(papers_meta).select('*')
tokenizer = Tokenizer(inputCol='title', outputCol='title_token')
papers_meta = tokenizer.transform(papers_meta).select('*')
papers_meta.select('words_token', 'title_token').show(10)<jupyter_output>+--------------------+--------------------+
|         words_token|         title_token|
+--------------------+--------------------+
|[community, acqui...|[etiology, and, i...|
|[the, localizatio...|[optimization, of...|
|[the, toxic, effe...|[evaluation, of, ...|
|[increasing, our,...|[the, influence, ...|
|[pathogen, outbre...|[does, pathogen, ...|
|[immunoglobulin, ...|[immunoglobulin, ...|
|[the, novel, coro...|[center, for, med...|
|[several, graph, ...|[generalized, lat...|
|[severe, acute, r...|[sars, coronaviru...|
|[the, year, occas...|[genetic, cellula...|
+--------------------+--------------------+
only showing top 10 rows<jupyter_text>Remove stop words<jupyter_code>remover = StopWordsRemover(inputCol='words_token', outputCol='words_clean')
papers_meta = remover.transform(papers_meta).select('*')
remover = StopWordsRemover(inputCol='title_token', outputCol='title_clean')
papers_meta = remover.transform(papers_meta).select('*')
papers_meta.select('words_clean', 'title_clean').show(10)<jupyter_output>+--------------------+--------------------+
|         words_clean|         title_clean|
+--------------------+--------------------+
|[community, acqui...|[etiology, impact...|
|[localization, ch...|[optimization, ph...|
|[toxic, effects, ...|[evaluation, tnf,...|
|[increasing, unde...|[influence, clima...|
|[pathogen, outbre...|[pathogen, spillo...|
|[immunoglobulin, ...|[immunoglobulin, ...|
|[novel, coronavir...|[center, medical,...|
|[several, graph, ...|[generalized, lat...|
|[severe, acute, r...|[sars, coronaviru...|
|[year, occasioned...|[genetic, cellula...|
+--------------------+--------------------+
only showing top 10 rows<jupyter_text>Now we need to remove the custom stopwords<jupyter_code>remover2 = StopWordsRemover(inputCol='words_clean', outputCol='words_clean_custom', stopWords = ['doi', 'preprint', 'copyright', 'peer', 'reviewed', 'org',
'https', 'et', 'al', 'author', 'figure','rights', 'reserved', 'permission', 'used', 'using',
'biorxiv', 'medrxiv', 'license', 'fig', 'fig.', 'al.', 'Elsevier', 'PMC', 'CZI', 'www'])
papers_meta = remover2.transform(papers_meta).select('*')
    remover2 = StopWordsRemover(inputCol='title_clean', outputCol='title_clean_custom', stopWords = ['doi', 'preprint', 'copyright', 'peer', 'reviewed', 'org',
'https', 'et', 'al', 'author', 'figure','rights', 'reserved', 'permission', 'used', 'using',
'biorxiv', 'medrxiv', 'license', 'fig', 'fig.', 'al.', 'Elsevier', 'PMC', 'CZI', 'www'])
papers_meta = remover2.transform(papers_meta).select('*')
papers_meta.select('words_clean_custom', 'title_clean_custom').show(10)
from pyspark.sql.functions import size
papers_meta = papers_meta.withColumn("wordcount", size("words_clean_custom"))
papers_meta.printSchema()<jupyter_output>root
 |-- Paper_ID: string (nullable = true)
 |-- title: string (nullable = true)
 |-- body: string (nullable = true)
 |-- authors: string (nullable = true)
 |-- journal: string (nullable = true)
 |-- publish_Year: integer (nullable = true)
 |-- publish_Month: integer (nullable = true)
 |-- words_token: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title_token: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- words_clean: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title_clean: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- words_clean_custom: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title_clean_custom: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- wordcount: integer (nullable = false)<jupyter_text>Drop the unneeded columns<jupyter_code>#papers_meta = papers_meta.drop('title', 'body', 'words_token', 'words_clean', 'title_token', 'title_clean')
papers_meta = papers_meta.drop('body', 'words_token', 'words_clean', 'title_token')
papers_meta.show()
papers_meta.repartition(12).write.parquet("./papers_meta_processed")
#papers_meta = spark.read\
#            .format("parquet")\
#            .option("header", "true")\
#            .load("./Dataset/papers_meta_processed")<jupyter_output><empty_output><jupyter_text>Step 5 (Vectorization and prepare the features column)  Now Apply Word2Vec on the processed body text "words_clean_custom" and the title text "title_clean_custom"<jupyter_code>from pyspark.ml.feature import Word2Vec

# Learn a mapping from words to Vectors.
word2Vec = Word2Vec(vectorSize=100, minCount=0, inputCol="words_clean_custom", outputCol="word2vec_body")
model = word2Vec.fit(papers_meta)

papers_meta = model.transform(papers_meta)
word2Vec = Word2Vec(vectorSize=100, minCount=0, inputCol="title_clean_custom", outputCol="word2vec_title")
model = word2Vec.fit(papers_meta)

papers_meta = model.transform(papers_meta)
papers_meta.show(10)
papers_meta.repartition(12).write.parquet("./Dataset/papers_meta_word2vec")
papers_meta = spark.read\
            .format("parquet")\
            .option("header", "true")\
            .load("./Dataset/papers_meta_word2vec")
papers_meta.printSchema()
#Select only the needed columns
papers_meta = papers_meta.select('Paper_ID', 'authors', 'journal', 'wordcount', 'publish_Year', 'publish_Month', 'word2vec_body', 'word2vec_title')
papers_meta.show(5)<jupyter_output>+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+
|            Paper_ID|             authors|             journal|wordcount|publish_Year|publish_Month|       word2vec_body|      word2vec_title|
+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+
|012b16ae54779ca1a...|<NAME>; Jo...|             Sci Rep|     1698|        2020|            2|[0.01163073299174...|[-0.0075389318427...|
|01bc7fe59fc7feb0e...|<NAME>; ...|Open Forum Infect...|     2106|        2018|            2|[0.08144156184301...|[-0.0090094477359...|
|0240a12c9fdf6c031...|Kao, Rowland R.; ...|Trends in Microbi...|     3808|        2014|            5|[0.04915223334121...|[-0.0112262216571...|
|03bfb747583f6b214...|                    |                    |     3848|           0|            0|[0.03949146437089...|[0.00115656061[...]<jupyter_text>Now let's prepare the features vector  First, Define StringIndexers for categorical columns<jupyter_code>from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import OneHotEncoder
cat_cols = [item[0] for item in papers_meta.dtypes if item[1].startswith('string')][1:]
cat_cols
indexers = [StringIndexer(
    inputCol=column, 
    outputCol=column + '_index', 
    handleInvalid='keep') for column in cat_cols]
encoders = [OneHotEncoder(
    inputCol=column + '_index', 
    outputCol= column + '_encoded') for column in cat_cols]
from pyspark.ml import Pipeline

pipeline = Pipeline(stages=indexers + encoders)
papers_meta_transformed = pipeline.fit(papers_meta).transform(papers_meta)
papers_meta_transformed.show(10)<jupyter_output>+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+
|            Paper_ID|             authors|             journal|wordcount|publish_Year|publish_Month|       word2vec_body|      word2vec_title|authors_index|journal_index|    authors_encoded|    journal_encoded|
+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+
|012b16ae54779ca1a...|<NAME>; Jo...|             Sci Rep|     1698|        2020|            2|[0.01163073299174...|[-0.0075389318427...|       7681.0|          7.0|(8321,[7681],[1.0])|   (2750,[7],[1.0])|
|01bc7fe59fc7feb0e...|<NAME>; ...|Open Forum Infect...|     2106|        2018|            2|[0.08144156184301...|[-0.0090094477359...|       14[...]<jupyter_text>Now select the required features and apply the vector assembler<jupyter_code>requiredFeatures = [
    'wordcount',
    'publish_Year',
    'publish_Month',
    'word2vec_body',
    'word2vec_title',
    'authors_encoded',
    'journal_encoded'
]
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(inputCols=requiredFeatures, outputCol='features')
papers_meta_transformed = assembler.transform(papers_meta_transformed)
papers_meta_transformed.show(10)
papers_meta_transformed.select('features').head(1)
papers_meta_transformed.repartition(12,'features').write.parquet("./Dataset/papersmeta_transformed")
papers_meta_transformed.show(5)<jupyter_output>+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+--------------------+
|            Paper_ID|             authors|             journal|wordcount|publish_Year|publish_Month|       word2vec_body|      word2vec_title|authors_index|journal_index|    authors_encoded|    journal_encoded|            features|
+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+--------------------+
|0d3951ca998dcf8af...|<NAME>.; L...|Veterinary Clinic...|     2490|        2010|           11|[0.06525062580420...|[0.00330294785089...|       1449.0|        113.0|(8321,[1449],[1.0])| (2750,[113],[1.0])|(11274,[0,1,2,3,4...|
|10218d6165dcbaa34...|<NAME>;...|       Nat Microbiol|    [...]<jupyter_text>Step 6 (PCA and Clustering)  Apply the PCA<jupyter_code>papers_meta_transformed = spark.read\
            .format("parquet")\
            .option("header", "true")\
            .load("./Dataset/papersmeta_transformed")
from pyspark.ml.feature import PCA
from pyspark.ml.linalg import Vectors

pca = PCA(k=2, inputCol="features", outputCol="features_pca")
model = pca.fit(papers_meta_transformed)
papers_meta_transformed = model.transform(papers_meta_transformed)
model.explainedVariance
papers_meta_transformed.show(10)
papers_meta_transformed.repartition(12).write.parquet("./Dataset/papersmeta_transformed_pca")<jupyter_output><empty_output><jupyter_text>Define the clustering model  Choose number of clusters k (based on Elbow method and silhouette_score)<jupyter_code>papers_meta_transformed = spark.read\
            .format("parquet")\
            .option("header", "true")\
            .load("./Dataset/papersmeta_transformed_pca")
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator

# Calculate cost and plot
import numpy as np
import pandas as pd 

cost = np.zeros(15)
silhouette = np.zeros(15)

for k in range(2,15):
    kmeans = KMeans().setK(k).setSeed(1).setFeaturesCol('features_pca')
    model = kmeans.fit(papers_meta_transformed)
    cost[k] = model.computeCost(papers_meta_transformed)
    clusterdData = model.transform(papers_meta_transformed)
    evaluator = ClusteringEvaluator()
    silhouette[k] = evaluator.evaluate(clusterdData)

# Plot the cost
df_eval = pd.DataFrame(np.array([cost[2:].tolist(),silhouette[2:].tolist()])).transpose()
df_eval.columns = ["cost", "silhouette_score"]
new_col = [2,3,4,5,6,7,8,9,10,11,12,13,14]
df_eval.insert(0, 'cluster', new_col)

import seaborn as sns
import matplotlib.pyplot as plt
sns.set_style("whitegrid")
fig = plt.figure(figsize=(40,10))
ax1 = fig.add_subplot(1, 2, 1)
ax1.set(xticks=range(2,15))
ax2 = fig.add_subplot(1, 2, 2)
ax2.set(xticks=range(2,15))
sns.lineplot(x='cluster', y='cost', data=df_eval, ax=ax1)
sns.lineplot(x='cluster', y='silhouette_score', data=df_eval, ax=ax2)
df_eval<jupyter_output><empty_output><jupyter_text>We choose k=4 based on the plots above<jupyter_code>kmeans = KMeans().setK(4).setFeaturesCol('features_pca')
model = kmeans.fit(papers_meta_transformed)
clusterdData = model.transform(papers_meta_transformed)<jupyter_output><empty_output><jupyter_text>Step 7 Recommender System The goal here is to build a basic recommender the system that reccomends similar papers to a given title. When provided with a paper title, the recommender is only going to consider papers which belong to the same cluster. it will then run cosine similiraty between the given title and the processed paper titles in the database. it will return a dictionary with suggested paper titles and the cosine dot product with respect to the given title<jupyter_code>paper_titles_df = spark.read\
            .format("parquet")\
            .option("header", "true")\
            .load("./Dataset/papers_meta_word2vec").select('Paper_ID', 'title', 'title_clean')
clusterdData = paper_titles_df.join(clusterdData, on=['Paper_ID'], how='inner')
clusterdData.show(5)<jupyter_output>+--------------------+--------------------+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+--------------------+--------------------+----------+
|            Paper_ID|               title|         title_clean|             authors|             journal|wordcount|publish_Year|publish_Month|       word2vec_body|      word2vec_title|authors_index|journal_index|    authors_encoded|    journal_encoded|            features|        features_pca|prediction|
+--------------------+--------------------+--------------------+--------------------+--------------------+---------+------------+-------------+--------------------+--------------------+-------------+-------------+-------------------+-------------------+--------------------+--------------------+----------+
|0782b4fb23ca65baf...|the population ge...|[population, gene...|Tibayrenc, Mich[...]<jupyter_text>We will pre-calculate tf and idf values and store them in a dataframe named data for our recommender system.<jupyter_code>from pyspark.sql.functions import udf, col
df_recommender = clusterdData.select('title', 'title_clean', col('prediction').alias('cluster'))
from pyspark.ml.feature import HashingTF, IDF
hashingTF = HashingTF(inputCol="title_clean", outputCol="tf")
tf = hashingTF.transform(df_recommender)

idf = IDF(inputCol="tf", outputCol="idf_feature").fit(tf)
tfidf = idf.transform(tf)
from pyspark.ml.feature import Normalizer
normalizer = Normalizer(inputCol="idf_feature", outputCol="norm")
data = normalizer.transform(tfidf)<jupyter_output><empty_output><jupyter_text>We will define the recommendPaper function. it takes a paper title and returns N recommendations from the same cluster.<jupyter_code>from pyspark.sql.types import DoubleType

dot_udf = udf(lambda x,y: float(x.dot(y)), DoubleType())

def recommendPaper(paper_title,N, data=data):
    target_paper = data.filter(data['title'] == paper_title)
    input_cluster = target_paper.select('cluster').collect()[0].cluster
    data = data.filter(data['cluster'] == input_cluster)
    recommendations = target_paper.alias("tearget_paper").crossJoin(data.alias("right"))\
        .select(col("tearget_paper.title").alias("target_title"), 
            col("right.title").alias("recommended_title"), 
            dot_udf("tearget_paper.norm", "right.norm").alias("dot"))\
        .sort(col("dot").desc())\
        .limit(N+1)
    return {reccomendation.recommended_title:reccomendation.dot for reccomendation in recommendations.collect()[1:]}
recommendPaper('the population genetics of trypanosoma cruzi revisited in the light of the predominant clonal evolution model',4)
recommendPaper('annual update in intensive care and emergency medicine',2)
recommendPaper('use of simple laboratory features to distinguish the early stage of severe acute respiratory syndrome from dengue fever',3)
recommendPaper('virology of hepatitis c virus',10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.26965885142255, 'fraction_numerical': 0.038461538461538464, 'mean_word_length': 2.2297139210890142, 'pattern_counts': {'":': 0, '<': 102, '<?xml version=': 0, '>': 104, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '26', 'dir': 'jupyter-structured-clean-dedup', 'id': '606800', 'license': 'Unlicense', 'n_tokens_mistral': 9901, 'n_tokens_neox': 8895, 'n_words': 1797, 'path': 'Tarek AbdElRahman&Mostafa Kamal - Project4.ipynb', 'repo_name': 'mostafa-k-m/COVID-19-Scholarly-Articles-Reccomendation-Engine', 'size': '30458'}