<jupyter_start><jupyter_text>Part 4: Scale up some more with Pandas and RayTODO: Summarize parts 1-3![First version of our processing pipeline](images/pipeline_v1.png)<jupyter_code>import market_intelligence as mi
articles = mi.maybe_download_articles()
articles<jupyter_output><empty_output><jupyter_text>Baseline: Process documents with a simple `for` loop. At the end of Part 3, we showed how to run our pipeline with a simple `for` loop.<jupyter_code>%%time

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)

def steps_1_through_4(doc_html: str) -> pd.DataFrame:
    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_2_results["person"],
                                                   spacy_language_model)
    step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
    return step_4_results

# Repeat steps 1-4 on every document
dataframes_to_stack = [
    steps_1_through_4(doc_html) for doc_html in articles["html"]
]

# Step 5: Merge the results across documents
step_5_results = pd.concat(dataframes_to_stack)
step_5_results<jupyter_output>CPU times: user 1min 30s, sys: 1.26 s, total: 1min 31s
Wall time: 4min 34s<jupyter_text>This code ran in about 4 1/2 minutes, a 3x improvement from the start of Part 3.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. Let's use [Ray](https://ray.io) to parallelize our document processing steps.TODO: Intro to Rayas you tune the accuracy of an NLP application, you'll move between different scalesfrom examining results on individual documents on your laptopto processing dozens of documents on your laptopto processing thousands of documents on a serverto processing millions of documents on a clusterand back again to your laptop for the next round of tuningRay lets you code up your processing pipeline once and have it work well across this wide variety of scales.The first step is to start up a Ray cluster by calling `ray.init()`.<jupyter_code>def reboot_ray():
    if ray.is_initialized():
        ray.shutdown()
    ray.init()
    
reboot_ray()<jupyter_output>2021-05-18 17:03:31,861	INFO services.py:1174 -- View the Ray dashboard at [1m[32mhttp://127.0.0.1:8265[39m[22m<jupyter_text>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:<jupyter_code># NOTE: The blog version of this cell should show this code side-by-side with the 
#  original code and highlight the changes.

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)

# @ray.remote decorator defines a Ray task
@ray.remote
def steps_1_through_4(doc_html: str) -> pd.DataFrame:
    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_2_results["person"],
                                                   spacy_language_model)
    step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
    return step_4_results<jupyter_output><empty_output><jupyter_text>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:<jupyter_code>doc_html = articles.iloc[1]["html"]

future = steps_1_through_4.remote(doc_html)
step_4_results = ray.get(future)<jupyter_output><empty_output><jupyter_text>oops, can't pickle a connection to Watson NLU -- it has open sockets and lock objectsalso can't pickle the language model inside the SpaCy dependency parser(as of SpaCy version 3.0).and even if you could, it's a 500 megabyte model.To make this work, we'll need to pull the initialization code inside the Ray task:<jupyter_code>@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_2_results["person"],
                                                   spacy_language_model)
    step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
    return step_4_results<jupyter_output><empty_output><jupyter_text>Now our Ray task works on a single document:<jupyter_code>step_4_results = ray.get(steps_1_through_4.remote(doc_html))
step_4_results<jupyter_output><empty_output><jupyter_text>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:<jupyter_code>reboot_ray()
%%time

# NOTE: The blog version of this listing should highlight what has changed 
#  relative to the original code.

@ray.remote
def steps_1_through_4(doc_html: str) -> pd.DataFrame:
    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)
    spacy_language_model = spacy.load("en_core_web_trf")
    
    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_2_results["person"],
                                                   spacy_language_model)
    step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
    return step_4_results

# Repeat steps 1-4 on every document, at the same time
dataframes_to_stack = ray.get([
    steps_1_through_4.remote(doc_html) for doc_html in articles["html"]
])

# Step 5: Merge the results across documents
step_5_results = pd.concat(dataframes_to_stack)
step_5_results<jupyter_output>[2m[36m(pid=17001)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17001)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17000)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17000)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=16999)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=16999)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=16991)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=16991)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17000)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17000)[0m Request failed 1 times; retrying in 1 sec
CPU times: user 4.65 s, sys: 1.44 s, total: 6.09 s
Wall time: 2min 20s<jupyter_text>This version actually works, and it runs in 1:35 on an 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) ERROR:root:Too Many Requests  (pid=8210) Traceback (most recent call last):  (pid=8210)   File "/Users/freiss/pd/blogs-market/env/lib/python3.7/site-packages/ibm_cloud_sdk_core/base_service.py", line 225, in send  (pid=8210)     response.status_code, http_response=response)  (pid=8210) ibm_cloud_sdk_core.api_exception.ApiException: Error: Too Many Requests, Code: 429 , X-global-transaction-id: ded4dd39-a617-4ce2-9d54-26aa7a82ab95  (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:```python@ray.remotedef 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:<jupyter_code>@ray.remote
class ParserModelActor(object):
    def __init__(self, spacy_model_name: str):
        self._language_model = spacy.load(spacy_model_name)
        self._nlu_api = ibm_watson.NaturalLanguageUnderstandingV1(version="2021-01-01", 
                authenticator=ibm_cloud_sdk_core.authenticators.IAMAuthenticator(api_key))
        self._nlu_api.set_service_url(service_url)

    def steps_1_through_4(self, doc_html: str) -> pd.DataFrame:
        step_1_results = mi.extract_named_entities_and_semantic_roles(doc_html, 
                                                                      self._nlu_api)
        step_2_results = mi.identify_persons_quoted_by_name(step_1_results)   
        step_3_results = mi.perform_targeted_dependency_parsing(
                                                       step_2_results["person"],
                                                       self._language_model)
        step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
        return step_4_results<jupyter_output><empty_output><jupyter_text>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:<jupyter_code>reboot_ray()
actor = ParserModelActor.remote("en_core_web_trf")
future = actor.steps_1_through_4.remote(doc_html)
step_4_results = ray.get(future)
step_4_results<jupyter_output><empty_output><jupyter_text>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.<jupyter_code>reboot_ray()
%%time

# This listing should NOT appear in the blog version.
# Run a pool with the actor defined above over the entire document collection, just
# to make sure it works and to determine the running time.

num_cpus = multiprocessing.cpu_count() // 2

actors = ray.util.ActorPool([ParserModelActor.remote("en_core_web_trf")
                             for i in range(num_cpus)])

# Repeat steps 1-4 on every document
dataframes_to_stack = actors.map_unordered(
    lambda actor, value: actor.steps_1_through_4.remote(value), articles["html"]
)

# Step 5: Merge the results across documents
step_5_results = pd.concat(dataframes_to_stack)
step_5_results<jupyter_output>[2m[36m(pid=17182)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17182)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17182)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17182)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17193)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17193)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17193)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17193)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17189)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17189)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17189)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17189)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17192)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17192)[0m Request failed 1 times; retrying in 1 sec
[2m[36m(pid=17192)[0m Request failed 1 times; retrying in 1[...]<jupyter_text>Using actors to manage request rateThe 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.We can use mutable actor state to track of how quickly our application is sending requests to the Watson Natural Language Understanding web service. One simple way to track the request rate is to remember the 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.<jupyter_code>import importlib
mi = importlib.reload(mi)
@ray.remote
class NLUClientActor(mi.RateLimitedActor):
    """
    Threaded actor to handle multiple simulatenous requests to the IBM Watson
    Natural Language Understanding service while respecting an upper bound on the
    number of requests per second.
    """
    def __init__(self, requests_per_sec: float, 
                 api_key: str, service_url: str):
        super().__init__(requests_per_sec)
        # One instance of the Python API for all threads
        self._nlu_api = ibm_watson.NaturalLanguageUnderstandingV1(
            version="2021-01-01", 
            authenticator=ibm_cloud_sdk_core.authenticators.IAMAuthenticator(api_key))
        self._nlu_api.set_service_url(service_url)
    
    def process_internal(self, doc_html: str) -> Any:
        return mi.extract_named_entities_and_semantic_roles(doc_html, self._nlu_api)
@ray.remote
class ParserModelActor(object):
    def __init__(self, spacy_model_name: str, nlu_client: NLUClientActor):
        self._language_model = spacy.load(spacy_model_name)
        self._nlu_client = nlu_client

    def steps_1_through_4(self, doc_html: str) -> pd.DataFrame:
        step_1_results = ray.get(self._nlu_client.process.remote(doc_html))
        step_2_results = mi.identify_persons_quoted_by_name(step_1_results)   
        step_3_results = mi.perform_targeted_dependency_parsing(
                                                       step_2_results["person"],
                                                       self._language_model)
        step_4_results = mi.extract_titles_of_persons(step_2_results, step_3_results)
        return step_4_results

reboot_ray()
%%time

num_cpus = multiprocessing.cpu_count() // 2

nlu_client = NLUClientActor.options(max_concurrency=5).remote(5.0, api_key, service_url)
actors = ray.util.ActorPool([ParserModelActor.remote("en_core_web_trf", nlu_client)
                             for i in range(num_cpus)])

dataframes_to_stack = actors.map_unordered(
    lambda actor, value: actor.steps_1_through_4.remote(value), 
    articles["html"]
)

step_5_results = pd.concat(dataframes_to_stack)
step_5_results<jupyter_output>CPU times: user 1.59 s, sys: 575 ms, total: 2.16 s
Wall time: 55.1 s<jupyter_text>Now we run in about 45 seconds without hitting the rate limit.And now the 5 documents per second rate limit of our Lite instance the chief bottleneck.We can remove that bottleneck by switching to a Standard instance of Watson Natural Languague Understanding with a simple one-line change.<jupyter_code># Don't show this cell in the blog version
standard_api_key = os.environ.get("STANDARD_API_KEY")
standard_service_url = os.environ.get("STANDARD_SERVICE_URL")  
reboot_ray()
%%time

num_cpus = multiprocessing.cpu_count()

# The blog version of this code listing should only show how the next line changes.
nlu_client = NLUClientActor.options(max_concurrency=num_cpus).remote(
    80.0, standard_api_key, standard_service_url)


# Note that this call to remote() will start asynchronously loading the language models.
actors = ray.util.ActorPool([ParserModelActor.remote("en_core_web_trf", nlu_client)
                             for i in range(num_cpus)])

dataframes_to_stack = actors.map_unordered(
    lambda actor, value: actor.steps_1_through_4.remote(value), 
    articles["html"]
)

step_5_results = pd.concat(dataframes_to_stack)
step_5_results<jupyter_output>CPU times: user 2.02 s, sys: 629 ms, total: 2.65 s
Wall time: 50 s<jupyter_text>Now our running time is down to 31 seconds, from our original running time of 852 seconds.That's a performance improvement of 27x!And with the 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.  Running time comparisonLet's compare the running times across the different ways of performing our task.<jupyter_code>import matplotlib.pyplot as plt
time_data = {
    # All times in seconds
    "for loop": 852,
    "Pandas optimizations": 283,
    "+Ray task": 90,
    "+Ray actors": 45, 
    "+standard\nNLU\ninstance": 30,   
    # Separate run on iRIS machine with 56 cores.
    "+cloud VM": 15,
}

plt.bar(time_data.keys(), time_data.values())
plt.ylabel("Running time (sec)")
plt.show()
num_docs = len(articles.index)
docs_per_sec = {
    k: num_docs/v for k, v in time_data.items()
}
plt.bar(docs_per_sec.keys(), docs_per_sec.values())
plt.ylabel("Documents per second")
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07537399309551208, 'fraction_numerical': 0.03222094361334868, 'mean_word_length': 4.212946763309173, 'pattern_counts': {'":': 6, '<': 51, '<?xml version=': 0, '>': 59, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '87825', 'license': 'Apache-2.0', 'n_tokens_mistral': 6798, 'n_tokens_neox': 6246, 'n_words': 2019, 'path': 'tutorials/market/Market_Intelligence_Part4.ipynb', 'repo_name': 'ZachEichen/text-extensions-for-pandas', 'size': '20856'}