Part 5: Scale up some more with Pandas and Ray¶

Let's start by summarizing our progress so far.

In Part 1, we defined an NLP pipeline that used Watson Natural Language Understanding and Text Extensions for Pandas to identify the names of executives in corporate press releases.

In Part 2, we extended our NLP pipeline by using SpaCy and Text Extensions for Pandas to associate each executive name with a job title.

If we separate the expensive NLP model evaluations from the rest of our code, we can think of this processing pipeline as having five distinct stages, as shown in the following diagram.

In Part 3, we applied the semijoin trick with Text Extensions for Pandas to reduce the cost of running the SpaCy dependency parser. This change decreased the cost of parsing by a factor of 9 and improved end-to-end running time by a factor of 3.

In Part 4, we showed how you can use Ray to rate-limit requests to a remote web service such as Watson Natural Language Understanding.

In this final part of the tutorial, we use Ray to take the performance improvements from Part 3 to the next level by applying parallel processing. By the time we're done, we'll have a version of our pipeline that runs 300 times faster than the pipeline we started with at the beginning of Part 3.

Baseline: Process documents with a simple for loop.¶

Parts 1 and 2 of this tutorial defined a processing pipeline with two parts: First find places where the document quotes a person by name, then associate those names with job titles.

If we break out the expensive model evaluations from these two parts, we get a four-step pipeline. Tack on the final step of merging these results across our entire collection of documents, and you will have five steps.

Here we have a version of the code from the end of Part 2, refactored slightly to break out those five distinct steps. Let's see how long it takes to run through our collection of 191 IBM press releases.

The exact running time for the above cell varies depending on which machine you use to run the notebook and what other processes are running in the background. On a 2020 Macbook Pro, it takes around 850 seconds. Let's see if we can improve on that running time.

Recap: Accelerating the baseline pipeline with the semijoin trick¶

In Part 3, we showed how to improve the end-to-end performance of this pipeline by applying the semijoin trick to the expensive dependency parsing step. This change made parsing 9x faster, improving the end-to-end performance by a factor of 3. Here's what our five-stage pipeline looks like after that change.

This code runs in about 280 seconds, a 3x improvement from the initial baseline version.

Now it's time to deploy parallel processing and improve that performance some more.

First version: Wrap the entire processing pipeline in a @ray.remote decorator.¶

Ray is a system for parallel processing that is designed to work at a wide variety of scales. As you tune the accuracy of an NLP application, you'll move between different sized inputs. You may start out by examining results on individual documents on your laptop, then switch to processing dozens of documents on your laptop. Later you may run through thousands of documents on a server. Eventually, you could deploy to production and process millions of documents on a cluster. And when there's a problem in production you might find yourself back again to working on your laptop for the next round of tuning.

Ray lets you code up your processing pipeline once and have it work well across this wide variety of scales. That way you can use the same code at every point throughout this iterative process.

In the remainder of this part of the tutorial, we're going to show how to parallelize our end-to-end pipeline using Ray. Once we have it running in parallel on a laptop, we'll take the same code over to a large server to get even more parallel speedup.

The first step is to start up a Ray cluster by calling ray.init().

Now let's try the simplest approach possible: Wrap our document processing code in a Ray remote function. To create a remote function, we just need to define a Python function and add the @ray.remote decorator to the function:

Now we can push the processing of a document to the Ray cluster by spawning a copy of the remote function. To spawn a copy of the function, we call the remote function's remote() method, which starts running the function in the background and returns a future -- a placeholder for the result that the function will produce when it completes.

If we pass the future to ray.get(), then Ray will block until the function has completed, download the result to the calling process, and return the result:

Oops, that didn't work! Ray tasks can only operate over Python objects that can be serialized. A connection to Watson NLU can't be serialized because the connection object has open sockets and lock objects

We also can't serialize the language model inside the SpaCy dependency parser (as of SpaCy version 3.0), because that Python object contains locks. Even if you could serialize it, doing so would result in Ray shipping a 500 megabyte model to every copy of the task, which would lead to underwhelming performance.

To make this work, we'll need to pull the initialization code for both models inside the Ray task:

Now our Ray task works on a single document:

To process our entire collection of documents, we modify our for loop so that it spawns background tasks for each of the documents instead of processing the documents locally. Then we wrap the for loop in a call to ray.get(), which tells Ray to wait until all the background tasks have completed and collect the results:

This version runs correctly, and it runs in about 90 seconds on our 8-core laptop -- an additional performance improvement of almost 3x on top of the 3x improvement from Part 3.

But there's room for two kinds of additional improvement:

• This code loads a 500 megabyte model for every document, which incurs a signficant additional cost. The 8 cores on our test machine are fully occupied, but we only see a speedup of 3. So we're using more than twice as many CPU cycles per document.
• This code produces several screens of scary log messages like:

  (pid=8210) Request failed 2 times; retrying in 2 sec
  

  because we're exceeding the request rate limit of our free Lite instance of Watson Natural Language Understanding. The limit for a Lite instance is 5 requests per second. We should be able to finish our 190 documents in 38 seconds while staying below the limit. But instead we're taking twice as long while continually bumping up against the limit and having to retry requests.

We can use Ray actors to address both of these problems.

Using actors to avoid repeated model startup¶

A Ray actor is persistent Python or Java object that lives in a Ray worker process. Actors can maintain arbitrary amounts of state, both mutable and immutable. If we attach the SpaCy language model to an actor, then the model will be initialized once in the actor's constructor instead of being loaded each time we process a document.

Turning our steps_1_through_4() function into a Ray Actor is a simple matter of restructuring the code slightly.

Here's the code before using actor:

@ray.remote
def steps_1_through_4(doc_html: str) -> pd.DataFrame:
    spacy_language_model = spacy.load("en_core_web_trf")
    nlu_api = ibm_watson.NaturalLanguageUnderstandingV1(version="2021-01-01", 
            authenticator=ibm_cloud_sdk_core.authenticators.IAMAuthenticator(api_key))
    nlu_api.set_service_url(service_url)

    step_1_results = mi.extract_named_entities_and_semantic_roles(doc_html, nlu_api)
    step_2_results = mi.identify_persons_quoted_by_name(step_1_results)   
    step_3_results = mi.perform_targeted_dependency_parsing(
                                                   step_1_results["analyzed_text"],
                                                   spacy_language_model)
    step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
    return step_4_results

And here's a version that uses an actor:

Compared with the previous Ray task, this version makes two important changes:

• The model initializaiton in first line of steps_1_through_4 moves to the class's constructor.
• We use the actor's local copy, self._language_model, as the input to the processing for step 3.

Invoking a Ray actor is a two-step process. First, you create an instance of the actor. Then you tell that instance to perform tasks.

Here's some code that creates an instance of the ParserModelActor Actor, invokes the actor's steps_1_through_4 task, blocks until the task completes, and returns the results:

This code runs, performs the computation in the background, and produces the correct result. And the actor will remain active as long as the Python variable actor is still in scope. We can pass additional documents to the steps_1_through_4 task without incurring the overhead of loading a language model each time.

As a Python class, this actor can only process one request at a time. To process documents in parallel, we can define a Ray actor pool, a group of multiple copies of our actor that can process documents in parallel.

Using actors to manage request rate¶

The large model that we attached to our ParserModelActor actor is an example of immutable actor state. Ray actors can also have mutable state that changes in response to tasks that the actor performs.

For this application, we can use mutable actor state to track of how quickly our application is sending requests to the Watson Natural Language Understanding web service.

Part 4 of this tutorial described in detail how to add such rate-limiting logic with a Ray actor. Here's a quick summary of how it works: We track the request rate is by tracking how much time has elapsed since the most recent request. With that information in hand, our actor can throttle new requests if they would exceed the rate limit. We put the logic for managing this state into an abstract base class, RateLimitedActor, the code for which can be found in market_intelligence.py. With that base class in place, we can define a Ray actor that sends documents to the Watson Natural Language Understanding web service while respecting a request rate limit. Because the Python API for Watson Natural Language Understanding is thread-safe, we can use a multithreaded Python actor to track multiple simulataneous requests.

Now our pipeline runs in about 45 seconds without hitting the rate limit.

And now the 5 documents per second rate limit of our Lite instance of Watson Natural Language Understanding the chief performance bottleneck. We can remove that bottleneck by switching to a Standard instance of the service with a simple one-line change.

Now our running time is down to 31 seconds, from our original running time of 852 seconds. That's a performance improvement of 27x!

Our 8-core laptop is now the bottleneck. We can further improve running time by switching to a larger machine or a cluster of machines, with no code changes.

On a larger machine from the IBM cloud, we can process these documents in 15 seconds, which is 57 times faster than the code we started with!

Running time comparison¶

Here's a detailed chart that compares the running times across the different ways of performing our task.