<jupyter_start><jupyter_text>Spark preproc pipeline<jupyter_code>def set_regextokenizer(inputCol, outputCol):
    "Use Spark to perform custom Regex tokenization"
    tokenizer = RegexTokenizer(
        inputCol=inputCol,
        outputCol=outputCol,
        pattern=r"[A-Za-z0-9\-]{3,50}",  # only alphanumeric hyphenated text with 3 or more chars
        gaps=False
    )
    return tokenizer


def set_stopword_remover(stopword, inputCol, outputCol):
    "use Spark to drop stopwords from a sequence of tokens"
    stopwords = sc.textFile(stopwordfile).collect()
    remover = (StopWordsRemover(
        inputCol=inputCol, outputCol=outputCol).setStopWords(stopwords))
    return remover
def run_spark_preproc_pipeline(df):
    """Create a Spark preprocessing that transforms the input DataFrame to produce a
       final output column storing each document as a list of tokens with stopwords removed.
    """
    preprocPipeline = Pipeline(stages=[
        set_regextokenizer("content", "words"),
        set_stopword_remover(stopwordfile, "words", "words_filtered")
    ])
    wordsDF = preprocPipeline.fit(df).transform(df)
    # Concatenate list of words into a single word for downstream Spark-NLP pipeline
    preprocDF = wordsDF.withColumn("words_joined", f.concat_ws(" ", "words_filtered"))
    return preprocDF

preprocDF = run_spark_preproc_pipeline(df)
preprocDF.show(5)<jupyter_output>+----------+--------------------+--------------------+--------------------+--------------------+
|      date|             content|               words|      words_filtered|        words_joined|
+----------+--------------------+--------------------+--------------------+--------------------+
|2016-06-30|Stellar pitching ...|[stellar, pitchin...|[stellar, pitchin...|stellar pitching ...|
|2016-06-30|Mayor Bill de Bla...|[mayor, bill, bla...|[mayor, bill, bla...|mayor bill blasio...|
|2016-06-30|In the early morn...|[the, early, morn...|[early, morning, ...|early morning hou...|
|2016-06-30|It was the Apple ...|[was, the, apple,...|[apple, store, ne...|apple store new y...|
|2016-06-30|The United States...|[the, united, sta...|[united, states, ...|united states oly...|
+----------+--------------------+--------------------+--------------------+--------------------+
only showing top 5 rows<jupyter_text>Spark NLP pipeline<jupyter_code>def set_document_assembler(inputCol):
    "Spark NLP document assembler"
    return DocumentAssembler().setInputCol(inputCol)


def set_tokenizer(inputCol, outputCol):
    "Tokenize text for input to the lemmatizer"
    tokenizer = (Tokenizer()
        .setInputCols([inputCol])
        .setOutputCol(outputCol)
    )
    return tokenizer


def set_lemmatizer(inputCol, outputCol):
    "Retrieve root lemmas out of the input tokens"
    # Use default SparkNLP English pretrained lemmatizer for now
    lemmatizer = (LemmatizerModel.pretrained(name="lemma_antbnc", lang="en")
        .setInputCols([inputCol])
        .setOutputCol(outputCol)
    )
    return lemmatizer


def set_finisher(finishedCol):
    "Finisher transform for Spark NLP pipeline"
    finisher = (Finisher()
        .setInputCols([finishedCol])
        .setIncludeMetadata(False)
    )
    return finisher


def run_sparknlp_pipeline(df):
    """Create a SparkNLP pipeline that transforms the input DataFrame to procude a final output
       column storing each document as a sequence of lemmas (root words).
    """
    nlpPipeline = Pipeline(stages=[
        set_document_assembler("words_joined"),
        set_tokenizer("document", "token"),
        set_lemmatizer("token", "lemma"),
        set_finisher("lemma")
    ])
    nlpPipelineDF = (nlpPipeline.fit(df)
        .transform(df)
        .withColumnRenamed('finished_lemma', 'allTokens')
    )
    return nlpPipelineDF
nlpPipelineDF = run_sparknlp_pipeline(preprocDF)
nlpPipelineDF.select('allTokens').show(5)
nlpPipelineDF.persist()<jupyter_output><empty_output><jupyter_text>Spark ML Pipeline<jupyter_code>def set_countvectorizer(inputCol, outputCol, params):
    countvectorizer = CountVectorizer(
        inputCol=inputCol,
        outputCol=outputCol,
        vocabSize=params['vocabsize'],
        minDF=params['minDF'],
        maxDF=params['maxDF'],
        minTF=1.0
    )
    return countvectorizer


def set_idf(inputCol, outputCol):
    return IDF(inputCol="features", outputCol="idf")


def set_lda_model(params):
    lda = LDA(
        k=params['num_topics'],
        maxIter=params['iterations'],
        optimizer="online",
        seed=1,
        learningOffset=100.0,  # If high, early iterations are downweighted during training
        learningDecay=0.51,    # Set between [0.5, 1) to guarantee asymptotic convergence
    )
    return lda


def run_ml_pipeline(nlpPipelineDF, params):
    """Create a Spark ML pipeline and transform the input NLP-transformed DataFrame 
       to produce a trained LDA topic model for the given data.
    """
    mlPipeline = Pipeline(
        stages=[
            set_countvectorizer("allTokens", "features", params),
            set_idf("features", "idf"),
            set_lda_model(params)
        ]
    )
    mlModel = mlPipeline.fit(nlpPipelineDF)
    ldaModel = mlModel.stages[2]
    # Calculate upper bound on model perplexity
    mlPipelineDF = mlModel.transform(nlpPipelineDF)
    ldaPerplexity = ldaModel.logPerplexity(mlPipelineDF)
    return mlModel, ldaPerplexity
mlModel, ldaPerplexity = run_ml_pipeline(nlpPipelineDF, params)
print("Upper bound on perplexity: {}".format(ldaPerplexity))<jupyter_output>Upper bound on perplexity: 7.251443234084913<jupyter_text>Describe Topics<jupyter_code>def describe_topics(mlModel):
    """Obtain topic words and weights from the LDA model.
       Returns: topics -> List[Dict[str, float]]
       A list of mappings between the top 15 topic words (str) and their weights
       (float) for each topic. The length of the list equals the number of topics.
    """
    # Store vocab from CountVectorizer
    vocab = mlModel.stages[0].vocabulary
    # Store LDA model part of pipeline
    ldaModel = mlModel.stages[2]

    # Take top 15 words in each topic
    topics = ldaModel.describeTopics(15)
    topics_rdd = topics.rdd

    topic_words = topics_rdd \
        .map(lambda row: row['termIndices']) \
        .map(lambda idx_list: [vocab[idx] for idx in idx_list]) \
        .collect()

    topic_weights = topics_rdd \
        .map(lambda row: row['termWeights']) \
        .collect()

    # Store topic words and weights as a list of dicts
    topics = [dict(zip(words, weights))
              for words, weights in zip(topic_words, topic_weights)]
    return topics
topics = describe_topics(mlModel)
topics<jupyter_output><empty_output><jupyter_text>Plot topic wordclouds<jupyter_code>def plot_wordclouds(topics, colormap="cividis"):
    cloud = WordCloud(
        background_color='white',
        width=600,
        height=400,
        colormap=colormap,
        prefer_horizontal=1.0,
    )

    num_topics = len(topics)
    fig_width = min(ceil(0.6 * num_topics + 6), 20)
    fig_height = min(ceil(0.65 * num_topics), 20)
    fig = plt.figure(figsize=(fig_width, fig_height))

    for idx, word_weights in tqdm(enumerate(topics), total=num_topics):
        ax = fig.add_subplot((num_topics / 5) + 1, 5, idx + 1)
        wordcloud = cloud.generate_from_frequencies(word_weights)
        ax.imshow(wordcloud, interpolation="bilinear")
        ax.set_title('Topic {}'.format(idx + 1))
        ax.set_xticklabels([])
        ax.set_yticklabels([])
        ax.tick_params(length=0)

    plt.tick_params(labelsize=14)
    plt.subplots_adjust(wspace=0.1, hspace=0.1)
    plt.margins(x=0.1, y=0.1)
    st = fig.suptitle("LDA Topics", y=0.92)
    fig.savefig("pyspark-topics.png", bbox_extra_artists=[st], bbox_inches='tight')
plot_wordclouds(topics)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1435303107488538, 'fraction_numerical': 0.015792154865002548, 'mean_word_length': 3.0107252298263534, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 20, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '166389', 'license': 'MIT', 'n_tokens_mistral': 2592, 'n_tokens_neox': 2471, 'n_words': 562, 'path': 'src/notebooks/3_pyspark_lda.ipynb', 'repo_name': 'prrao87/topic-modelling', 'size': '7852'}