Running Tasks in Parallel

In order to illustrate how multiprocessing can help us perform machine learning on text, let’s consider an example where we would like to fit multiple models, cross-validate them, and save them to disk. We will begin by writing three functions to generate a naive Bayes model, a logistic regression, and a multilayer perceptron. Each function in turn creates three different models, defined by Pipelines, that extract text from a corpus located at a specified path. Each task also determines a location to write the model to, and reports results using the logging module (more on this in a bit):

from transformers import TextNormalizer, identity

from sklearn.pipeline import Pipeline
from sklearn.naive_bayes import MultinomialNB
from sklearn.linear_model import LogisticRegression
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.neural_network import MLPClassifier


def fit_naive_bayes(path, saveto=None, cv=12):

    model = Pipeline([
        ('norm', TextNormalizer()),
        ('tfidf', TfidfVectorizer(tokenizer=identity, lowercase=False)),
        ('clf', MultinomialNB())
    ])

    if saveto is None:
        saveto = "naive_bayes_{}.pkl".format(time.time())

    scores, delta = train_model(path, model, saveto, cv)
    logger.info((
        "naive bayes training took {:0.2f} seconds "
        "with an average score of {:0.3f}"
    ).format(delta, scores.mean()))


def fit_logistic_regression(path, saveto=None, cv=12):
    model = Pipeline([
        ('norm', TextNormalizer()),
        ('tfidf', TfidfVectorizer(tokenizer=identity, lowercase=False)),
        ('clf', LogisticRegression())
    ])

    if saveto is None:
        saveto = "logistic_regression_{}.pkl".format(time.time())

    scores, delta = train_model(path, model, saveto, cv)
    logger.info((
        "logistic regression training took {:0.2f} seconds "
        "with an average score of {:0.3f}"
    ).format(delta, scores.mean()))


def fit_multilayer_perceptron(path, saveto=None, cv=12):
    model = Pipeline([
        ('norm', TextNormalizer()),
        ('tfidf', TfidfVectorizer(tokenizer=identity, lowercase=False)),
        ('clf', MLPClassifier(hidden_layer_sizes=(10,10), early_stopping=True))
    ])

    if saveto is None:
        saveto = "multilayer_perceptron_{}.pkl".format(time.time())

    scores, delta = train_model(path, model, saveto, cv)
    logger.info((
        "multilayer perceptron training took {:0.2f} seconds "
        "with an average score of {:0.3f}"
    ).format(delta, scores.mean()))

While each of our functions can be modified and customized individually, each must also share common code. This shared functionality is defined in the train_model() function, which creates a PickledCorpusReader from the specified path. The train_model() function uses this reader to create instances and labels, compute scores using the cross_val_score utility from Scikit-Learn, fit the model, write it to disk using joblib (a specialized pickle module used by Scikit-Learn), and return the scores:

from reader import PickledCorpusReader

from sklearn.externals import joblib
from sklearn.model_selection import cross_val_score


@timeit
def train_model(path, model, saveto=None, cv=12):
    # Load the corpus data and labels for classification
    corpus = PickledCorpusReader(path)
    X = documents(corpus)
    y = labels(corpus)

    # Compute cross validation scores
    scores = cross_val_score(model, X, y, cv=cv)

    # Fit the model on entire dataset
    model.fit(X, y)

    # Write to disk if specified
    if saveto:
        joblib.dump(model, saveto)

    # Return scores as well as training time via decorator
    return scores

The documents() and labels() are helper functions that read the data from the corpus reader into a list in memory as follows:

def documents(corpus):
    return [
        list(corpus.docs(fileids=fileid))
        for fileid in corpus.fileids()
    ]


def labels(corpus):
    return [
        corpus.categories(fileids=fileid)[0]
        for fileid in corpus.fileids()
    ]

We can keep track of the time of execution using a @timeit wrapper, a simple debugging decorator that we will use to compare performance times:

import time
from functools import wraps

def timeit(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        return result, time.time() - start
    return wrapper

Python’s logging module is generally used to coordinate complex logging across multiple threads and modules. The logging configuration is at the top of the module, outside of any function, so it is executed when the code is imported. In the configuration, we can specify the %(processName)s directive, which allows us to determine which process is writing the log message. The logger is set to the module’s name so that different modules’ log statements can also be disambiguated:

import logging

# Logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="%(processName)-10s %(asctime)s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

At long last, we’re ready to actually execute our code in parallel, with a run_parallel function. This function takes a path to the corpus as an argument, the argument that is shared by all tasks. The task list is defined, then for each function in the task list, we create an mp.Process object whose name is the name of the task, target is the callable, and args and kwargs are specified as a tuple and a dictionary, respectively. To keep track of the processes we append them to a procs list before starting the process.

At this point, if we did nothing, our main process would exit as the run_parallel function is complete, which could cause our child processes to exit prematurely or to be orphaned (i.e., never terminate). To prevent this, we loop through each of our procs and join them, rejoining each to the main process. This will cause the main function to block (wait) until the processes’ join method is called. By looping through each proc, we ensure that we don’t continue until all processes have completed, at which point we can log how much total time the process took:

def run_parallel(path):
    tasks = [
        fit_naive_bayes, fit_logistic_regression, fit_multilayer_perceptron,
    ]

    logger.info("beginning parallel tasks")
    start = time.time()

    procs = []
    for task in tasks:
        proc = mp.Process(name=task.__name__, target=task, args=(path,))
        procs.append(proc)
        proc.start()

    for proc in procs:
        proc.join()

    delta = time.time() - start
    logger.info("total parallel fit time: {:0.2f} seconds".format(delta))


if __name__ == '__main__':
    run_parallel("corpus/")

Running these three tasks in parallel requires a little extra thought and a bit more work to ensure that everything is set up correctly. So what do we get out of it? In Table 11-1, we show a comparison of task and total time averaged over ten runs.

We can see that each individual task takes slightly longer when running in parallel; potentially, this extra time represents the minimal amount of overhead required to set up and manage multiprocessing. This slight increase in time is more than made up for in the total run time—roughly the length of the longest fit task and 2.6x times faster than running each task sequentially. When running a significant number of modeling tasks, multiprocessing clearly makes a difference!

While this use of multiprocessing.Process demonstrates a number of essential concepts, far more common is the use of a process pool, which we will discuss in the next section.

Process Pools and Queues

In the previous section we looked at how to use the multiprocessing.Process object to run individual tasks in parallel. The Process object makes it easy to define individual functions to run independently, closing the function when it is complete. More advanced usage might employ subclasses of the Process object, each of which must implement a run() method that defines their behavior.

In larger architectures this allows easier management of individual processes and their arguments (e.g., naming them independently or managing database connections or other per-process attributes) and is generally used for task parallel execution. With task parallelism, each task has independent input and output (or the output of one task may be the input to another task). In contrast, data parallelism requires the same task to be mapped to multiple inputs. Because the input is independent, each task can be applied in parallel. For data parallelism, the multiprocessing library provides simpler abstractions in the form of Pools and Queues, which we’ll explore in this section.

Larger workflows can be described as a directed acyclic graph (DAG), where a series of parallel steps is executed with synchronization points in between. A synchronization point ensures that all parts of the processing have completed or caught up before execution continues. Data is also generally exchanged at synchronization points, sent out to parallel tasks from the main task, or retrieved by the main task.

Data parallel execution is still appropriate for the Python multiprocessing library, however, some additional considerations are required. The first is how many processes to use and how. For simple operations, it is inefficient to run a process for each input and then tear the process down (or worse, run one process per input, which would swamp your operating system). Instead, a fixed number of processes is instantiated in a multiprocessing.Pool, each of which read input, apply the operation, then send output until the input data is exhausted.

This leads to the second consideration: How do you safely send and receive data from a process, ensuring no duplication or corruption? For this, you need to use a multiprocessing.Queue, a data structure that is both thread- and multiprocessing-safe because operations are synchronized with locks to ensure that only one process or thread has access to the queue at a time. A process can safely put(item) an item on the queue and another process can safely get() an item from the queue in a first-in, first-out (FIFO) fashion.

A common architecture for this type of processing is shown in Figure 11-2. The Pool forks n processes, each of which gets to work reading an input queue and sending their data to an output queue. The main process continues by enqueuing input data into the input queue. Generally once it’s done enqueuing the input data, the main process also enqueues n semaphores, flags that tell the processes in the process pool that there is no more data and they can terminate. The main process then can join the pool waiting for all processes to complete, or begin work immediately fetching data from the output queue for final processing.

Figure 11-2. Process pool and queues

If this all sounds like a lot of work to set up each data structure and maintain the code for enqueuing and processing, don’t worry, the multiprocessing library provides simple methods on the Pool object to perform this work. The methods, apply, map, imap, and starmap each take a function and arguments and send them to the pool for processing, blocking until returning a result. These methods also have an _async counterpart—for example, apply_async does not block but instead returns an AsynchronousResult object that is filled in when the task is done, or can call callbacks on success or error when completed. We’ll see how to use apply_async in the next section.

Parallel Corpus Preprocessing

Adapting a corpus reader to use multiprocessing can be fairly straightforward when you consider that each document can be independently processed for most tasks, particularly for things like frequency analysis, vectorization, and estimation. In these cases, all multiprocessing requires is a function whose argument is a path on disk, and the fileids read from the corpus can be mapped to a process pool.

Probably the most common and time-consuming task applied to a corpus, however, is preprocessing the corpus from raw text into a computable format. Preprocessing, discussed in Chapter 3, takes a document and converts it into a standard data structure: a list of paragraphs that are lists of sentences, which in turn are lists of (token, part of speech) tuples. The final result of preprocessing is usually saving the document as a pickle, which is both usually more compact than the original document as well as easily loaded into Python for further processing.

In Chapter 3, we created a Preprocessor class that wrapped a CorpusReader object so that a method called process was applied to each document path in the corpus. The main entry point to run the preprocessor was a transform method that kicked off transforming documents from the corpus and saving them into a target directory.

Here we will extend that class, which gives us the ability to use apply_async with a callback that saves state. In this case we create a self.results list to store the results as they come back from the process() method, but it is easy to adapt on_result() to update a process or do logging.

Next, we modify the transform method to count the cores available on the local machine using mp.cpu_count(). We then create the process pool, enqueuing the tasks by iterating over all the fileids and applying them to the pool (which is where the callback functionality comes into play to modify the state). Finally we close the pool (with pool.close()), meaning that no additional tasks can be applied and the child processes will join when done, and we wait for them to complete (with pool.join()):

class ParallelPreprocessor(Preprocessor):

    def on_result(self, result):
        self.results.append(result)

    def transform(self, tasks=None):
        [...]

        # Reset the results
        self.results = []

        # Create a multiprocessing pool
        tasks = tasks or mp.cpu_count()
        pool  = mp.Pool(processes=tasks)

        # Enqueue tasks on the multiprocessing pool and join
        for fileid in self.fileids():
            pool.apply_async(
                self.process, (fileid,), callback=self.on_result
            )

        # Close the pool and join
        pool.close()
        pool.join()

        return self.results

Although this introduction to threads and multiprocessing was brief and high-level hopefully it gives you a sense of the challenges and opportunities provided by parallelism. Many text analysis tasks can be sped up and scaled linearly simply by using the multiprocessing module and applying already existing code. By running multiprocessing code on a modern laptop or by using a large compute-optimized cloud instance, it is relatively simple to take advantage of multiprocessing without the overhead of setting up a cluster. Compared to cluster computations, programs written with multiprocessing are also easier to reason about and manage.

Anatomy of a Spark Job

Spark is an execution engine for distributed programs whose primary advantage is support for in-memory computing. Because Spark applications can be written quickly in Java, Scala, Python, and R it has become synonymous with Big Data science. Several libraries built on top of Spark such as Spark SQL and DataFrames, MLlib, and GraphX mean that data scientists used to local computing in notebooks with these tools feel comfortable very quickly in the cluster context. Spark has allowed applications to be developed upon datasets previously inaccessible to machine learning due to their scope or size; a category that many text corpora fall into. In fact, cluster computing frameworks were originally developed to handle text data scraped from the web.

Spark can run in two modes: client mode and cluster mode. In cluster mode, a job is submitted to the cluster, which then computes independently. In client mode, a local client connects to the cluster in an interactive fashion; jobs are sent to the cluster and the client waits until the job is complete and data is returned. This makes it possible to interact with the cluster using PySpark, an interactive interpreter similar to the Python shell, or in a Juypter notebook. For dynamic analysis, client mode is perfect for quickly getting answers to questions on smaller datasets and corpora. For more routine or longer running jobs, cluster mode is ideal.

In this section we will briefly explore how to compose Python programs using Spark’s version 2.x API. The code can be run locally in PySpark or with the spark-submit command.

With PySpark:

$ pyspark
Python 3.6.3 (v3.6.3:2c5fed86e0, Oct  3 2017, 00:32:08)
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.3.0
      /_/

Using Python version 3.6.3 (v3.6.3:2c5fed86e0, Oct  3 2017 00:32:08)
SparkSession available as 'spark'.
>>> import hello

With spark-submit:

$ spark-submit hello.py

In practice we would use a series of flags and arguments in the above spark-submit command to let Spark know, for example, the URL for the cluster (with --master), the entry point for the application (with --class), the location for the driver to execute in the deployment environment (with --deploy-mode), etc.

A connection to the cluster, called the SparkContext, is required. Generally, the SparkContext is stored in a global variable, sc, and if you launch PySpark in either a notebook or a terminal, the variable will immediately be available to you. If you run Spark locally, the SparkContext is essentially what gives you access to the Spark execution environment, whether that’s a cluster or a single machine. To create a standalone Python job, creating the SparkContext is the first step, and a general template for Spark jobs is as follows:

from pyspark import SparkConf, SparkContext

APP_NAME = "My Spark Application"


def main(sc):
    # Define RDDs and apply operations and actions to them.


if __name__ == "__main__":
    # Configure Spark
    conf = SparkConf().setAppName(APP_NAME)
    sc   = SparkContext(conf=conf)

    # Execute Main functionality
    main(sc)

Now that we have a basic template for running Spark jobs, the next step is to load data from disk in a way that Spark can use, as we will explore in the next section.

Distributing the Corpus

Spark jobs are often described as directed acyclic graphs (DAGs), or as acyclic data flows. This refers to a style of programming that envisions data loaded into one or more partitions, and subsequently transformed, merged, or split until some final state is reached. As such, Spark jobs begin with resilient distributed datasets (RDDs), collections of data partitioned across multiple machines, which allows safe operations to be applied in a distributed manner.

The simplest way to create RDDs is to organize data similarly to how we organized our corpus on disk—directories with document labels, and each document as its own file on disk. For example, to load the hobbies corpus introduced in “Loading Yellowbrick Datasets”, we use sc.wholeTextFiles to return an RDD of (filename, content) pairs. The argument to this method is a path that can contain wildcards and point to a directory of files, a single file, or compressed files. In this case, the syntax "hobbies/*/*.txt" looks for any file with a .txt extension under any directory in the hobbies directory:

corpus = sc.wholeTextFiles("hobbies/*/*.txt")
print(corpus.take(1))

The take(1) action prints the first element in the corpus RDD, allowing you to visualize that the corpus is a collection of tuples of strings. With the hobbies corpus, the first element shows as follows:

[('file:/hobbies/books/56d62a53c1808113ffb87f1f.txt',
"\r\n\r\nFrom \n\n to \n\n, Oscar voters can't get enough of book
adaptations. Nowhere is this trend more obvious than in the Best
Actor and Best Actress categories.\n\n\r\n\r\nYes, movies have
been based on books and true stories since the silent film era,
but this year represents a notable spike...")]

Saving data to disk and parsing it can be applied to a variety of formats including other text formats like JSON, CSV, and XML or binary formats like Avro, Parquet, Pickle, Protocol Buffers, etc., which can be more space-efficient. Regardless, once preprocessing has been applied (either with Spark or as described in “Parallel Corpus Preprocessing”), data can be stored as Python objects using the RDD.saveAsPickleFile method (similar to how we stored pickle files for our preprocessed corpus) and then loaded using sc.pickleFile.

Now that we’ve loaded our hobbies corpus into RDDs, the next step is to apply transformations and actions to them, which we will discuss more in the next section.

RDD Operations

There are two primary types of operations in a Spark program: transformations and actions. Transformations are operations that manipulate data, creating a new RDD from an existing one. Transformations do not immediately cause execution to occur on the cluster, instead they are described as a series of steps applied to one or more RDDs. Actions, on the other hand, do cause execution to occur on the cluster, causing a result to be returned to the driver program (in client mode) or data to be written to disk or evaluated in some other fashion (in cluster mode).

The three most common transformations—map, filter, and flatMap—each accept a function as their primary argument. This function is applied to each element in the RDD and each returned value is used to create the new RDD.

For instance, we can use the Python operator module to extract the hobbies subcategory for each document. We create a parse_label function that can extract the category name from the document’s filepath. As in the example we looked at before, the data RDD of (filename, content) key-value pairs is created by loading whole text files from the specified corpus path. We can then create the labels RDD by mapping itemgetter(0), an operation that selects only filenames from each element of the data RDD, then mapping the parse_label function to each:

import os
from operator import itemgetter

def parse_label(path):
    # Returns the name of the directory containing the file
    return os.path.basename(os.path.dirname(path))


data = sc.wholeTextFiles("hobbies/*/*.txt")
labels = data.map(itemgetter(0)).map(parse_label)

Note that at this point, nothing is executed across the cluster because we’ve only defined transformations, not actions. Let’s say we want to get the count of the documents in each of the subcategories.

While our labels RDD is currently a collection of strings, many Spark operations (e.g., groupByKey and reduceByKey) work only on key-value pairs. We can create a new RDD called label_counts, which is first transformed by mapping each label into a key-value pair, where the key is the label name, and the value is a 1. We can then reduceByKey with the add operator, which will sum all the 1s by key, giving us the total count of documents per category. We then use the collect action to execute the transformations across the cluster, loading the data, creating the labels and label_count RDD, and returning a list of (label, count), tuples, which can be printed in the client program:

from operator import add

label_count = labels.map(lambda l: (l, 1)).reduceByKey(add)
for label, count in label_count.collect():
    print("{}: {}".format(label, count))

The result is as follows:

books: 72
cinema: 100
gaming: 128
sports: 118
cooking: 30

As we’ve discussed in previous sections, Spark applications are defined by data flow operations. We first load data into one or more RDDs, apply transformations to those RDDs, join and merge them, then apply actions and save the resulting data to disk or aggregate data and bring it back to a driver program. This is a powerful abstraction that allows us to think about data as a collection rather than as a distributed computation, enabling cluster computing without requiring much effort on the analyst’s part. More complex usage of Spark involves creating DataFrames and Graphs, a little of which we’ll see in the next section.

NLP with Spark

Natural language processing is a special interest of the distributed systems community. This is not only because some of the largest datasets are text (in fact, Hadoop was designed specifically to parse HTML documents for search engines), but also because cutting-edge, language-aware applications require especially large corpora to be effective. As a result, Spark’s machine learning library, MLLib,⁴ boasts many tools for intelligent feature extraction and vectorization of text similar to the ones discussed in Chapter 4, including utilities for frequency, one-hot, TF–IDF, and word2vec encoding.

Machine learning with Spark starts with a collection of data similar to an RDD, the SparkSQL DataFrame. Spark DataFrames are conceptually equivalent to relational database tables or their Pandas counterparts, coordinating data by row and column as a table. However, they add rich optimizations that take advantage of the SparkSQL execution engine, and are quickly becoming the standard for distributed data science.

If we wanted to use SparkMLLib on our hobbies corpus, we would first transform the corpus RDD using a SparkSession (exposed as the global variable spark in PySpark) to create a DataFrame from the collection of tuples, identifying each column by name:

# Load data from disk
corpus = sc.wholeTextFiles("hobbies/*/*.txt")

# Parse the label from the text path
corpus = corpus.map(lambda d: (parse_label(d[0]), d[1]))

# Create the dataframe with two columns
df = spark.createDataFrame(corpus, ["category", "text"])

The SparkSession is the entry point to SparkSQL and the Spark 2.x API. To use this in a spark-submit application, you must first build it, similar to how we constructed the the SparkContext in the previous section. Adapt the Spark program template by adding the following lines of code:

from pyspark.sql import SparkSession
from pyspark import SparkConf, SparkContext

APP_NAME = "My Spark Text Analysis"


def main(sc, spark):
    # Define DataFrames and apply ML estimators and transformers


if __name__ == "__main__":
    # Configure Spark
    conf = SparkConf().setAppName(APP_NAME)
    sc   = SparkContext(conf=conf)

    # Build SparkSQL Session
    spark = SparkSession(sc)

    # Execute Main functionality
    main(sc, spark)

With our corpus structured as a Spark DataFrame, we can now get to the business of fitting models and transforming datasets. Luckily, Spark’s API is very similar to the Scikit-Learn API, so transitioning from Scikit-Learn or using Scikit-Learn and Spark in conjunction is not difficult.

from pyspark.ml.feature import RegexTokenizer

# Create the RegexTokenizer
tokens = RegexTokenizer(
    inputCol="text", outputCol="tokens",
    pattern="\\w+", gaps=False, toLowercase=True)

# Transform the corpus
corpus = tokens.transform(corpus)

def make_vectorizer(stopwords=True, tfidf=True, n_features=5000):
    # Creates a vectorization pipeline that starts with tokenization
    stages = [
        Tokenizer(inputCol="text", outputCol="tokens"),
    ]

    # Append stopwords to the pipeline if requested
    if stopwords:
        stages.append(
            StopWordsRemover(
                caseSensitive=False, outputCol="filtered_tokens",
                inputCol=stages[-1].getOutputCol(),
            ),
        )

    # Create the Hashing term frequency vectorizer
    stages.append(
        HashingTF(
            numFeatures=n_features,
            inputCol=stages[-1].getOutputCol(),
            outputCol="frequency"
        )
    )

    # Append the IDF vectorizer if requested
    if tfidf:
        stages.append(
            IDF(inputCol=stages[-1].getOutputCol(), outputCol="tfidf")
        )

    # Return the completed pipeline
    return Pipeline(stages=stages)

vectors = make_vectorizer().fit(corpus)
corpus = vectors.transform(corpus)
corpus[['label', 'tokens', 'tfidf']].show(5)

+-----+--------------------+--------------------+
|label|              tokens|               tfidf|
+-----+--------------------+--------------------+
|books|[, name, :, ian, ...|(5000,[15,24,40,4...|
|books|[, written, by, k...|(5000,[8,177,282,...|
|books|[, last, night,as...|(5000,[3,9,13,27,...|
|books|[, a, sophisticat...|(5000,[26,119,154...|
|books|[, pools, are, so...|(5000,[384,569,60...|
+------+--------------------+--------------------+
only showing top 5 rows

from tabulate import tabulate
from pyspark.ml import Pipeline
from pyspark.ml.clustering import BisectingKMeans
from pyspark.ml.feature import Word2Vec, Tokenizer

# Create the vector/cluster pipeline
pipeline = Pipeline(stages=[
    Tokenizer(inputCol="text", outputCol="tokens"),
    Word2Vec(vectorSize=7, minCount=0, inputCol="tokens", outputCol="vecs"),
    BisectingKMeans(k=10, featuresCol="vecs", maxIter=10),
])

# Fit the model
model = pipeline.fit(corpus)
corpus = model.transform(corpus)

# Retrieve stages
bkm = model.stages[-1]
wvec = model.stages[-2]

# Evaluate clustering
cost = bkm.computeCost(corpus)
sizes = bkm.summary.clusterSizes

# Get the text representation of each cluster
table = [["Cluster", "Size", "Terms"]]
for ci, c in enumerate(bkm.clusterCenters()):
    ct = wvec.findSynonyms(c, 7)
    size = sizes[ci]
    terms = " ".join([row.word for row in ct.take(7)])
    table.append([ci, size, terms])

# Print the results
print(tabulate(table))
print("Sum of square distance to center: {:0.3f}".format(cost))

Cluster  Size  Terms
-------  ----  ---------------------------------------------------------------
    0      81  the"soros caption,"bye markus henkes clarity. presentation,elon
    1       3  novak hiatt  veered monopolists,then,would clarity. cilantro.
    2       2  publics. shipmatrix. shiri flickr groupon,meanwhile,has sleek!
    3       2  barrymore 8,2016 tips? muck glorifies tags between,earning
    4     265  getting sander countervailing officers,ohio,then voter. dykstra
    5     550  back peyton's condescending embryos racist,any voter. nebraska
    6     248  maxx,and davan think'i smile,i 2014,psychologists thriving.
    7     431  ethnography akhtar problem,and studies,taken monica,california.
    8     453  instilled wife! pnas,the ideology,with prowess,pride
    9     503  products,whereas attacking grouper sets,facebook flushing,
Sum of square distance to center: 39.750

from pyspark.ml.feature import StringIndexer

# Create the vectorizer
vector = make_vectorizer().fit(corpus)

# Index the labels of the classification
labelIndex = StringIndexer(inputCol="label", outputCol="indexedLabel")
labelIndex = labelIndex.fit(corpus)

# Split the data into training and test sets
training, test = corpus.randomSplit([0.8, 0.2])

from pyspark.ml.classification import LogisticRegression

model = Pipeline(stages=[
    vector, labelIndex, clf
]).fit(training)

# Make predictions
predictions = model.transform(test)
predictions.select("prediction", "indexedLabel", "tfidf").show(5)

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

evaluator = MulticlassClassificationEvaluator(
    labelCol="indexedLabel",
    predictionCol="prediction",
    metricName="f1"
)

score = evaluator.evaluate(predictions)
print("F1 Score: {:0.3f}".format(score))

# Vectorize the corpus on the cluster
vector = make_vectorizer().fit(corpus)
corpus = vector.transform(corpus)

# Get the sample from the dataset
sample = corpus.sample(False, 0.1).collect()
X = [row['tfidf'] for row in sample]
y = [row['label'] for row in sample]

# Train a Scikit-Learn Model
clf = AdaBoostClassifier()
clf.fit(X, y)

# Broadcast the Scikit-Learn Model to the cluster
clf = sc.broadcast(clf)

# Create accumulators for correct vs incorrect
correct = sc.accumulator(0)
incorrect = sc.accumulator(1)

def make_accuracy_closure(model, correct, incorrect):
    # model should be a broadcast variable
    # correct and incorrect should be accumulators
    def inner(rows):
        X = []
        y = []

        for row in rows:
            X.append(row['tfidf'])
            y.append(row['label'])

        yp = model.value.predict(X)
        for yi, ypi in zip(y, yp):
            if yi == ypi:
                correct.add(1)
            else:
                incorrect.add(1)
    return inner

# Create the accuracy closure
accuracy = make_accuracy_closure(clf, incorrect, correct)

# Compute the number incorrect and correct
corpus.foreachPartition(accuracy)

accuracy = float(correct.value) / float(correct.value + incorrect.value)
print("Global accuracy of model was {}".format(accuracy))