Using dask with Stanford CoreNLP server in Stanza

Viewed 166

I am trying to optimize Stanza's CoreNLPClient with dask by trying to run it in parallel.

For example: consider the following sample DF

import pandas as pd
df = pd.DataFrame(["Eating an apple is good for your health"]*100, columns = ["Sample"])

Parallelize Code:

from dask import dataframe as dask_dd
from dask.multiprocessing import get
import stanza
from stanza.server import CoreNLPClient, CorefChain

def parallel_func(text):
    
    with CoreNLPClient(
        annotators=['tokenize','ssplit','pos','lemma','ner', 'parse', 'depparse','coref'],
        timeout=30000,endpoint="http://localhost:9000",
        memory='16G') as client:
        
        ann = client.annotate(text)
    return ann.corefChain

def main_func(data):
    
    data["chain"] = dask_dd.from_pandas(data, npartitions=4).map_partitions(
    lambda t: t.apply(lambda m: parallel_func(m["Sample"]), axis = 1)).compute(scheduler="threads")

Error:

Unable to start the CoreNLP server on port 9000 (possible something is already running there)

Attempt 1 to rectify this error:

Since the port is occupied I tried to get different_partitions run on different_port by manually creating a index for the DF and randomly select a number to be passed as the port, assuming each_sub_partition created will have different_slice of the DF so the randomly selected index will be different

Code:

def parallel_func(text, port):
    
    with CoreNLPClient(
        annotators=['tokenize','ssplit','pos','lemma','ner', 'parse', 'depparse','coref'],
        timeout=30000,endpoint="http://localhost:900"+str(port),
        memory='16G') as client:
        
        ann = client.annotate(text)
    return ann.corefChain

def main_func(data):

   data=data.reset_index(inplace=False)
   data["chain"] = dask_dd.from_pandas(data, npartitions=4).map_partitions(
   lambda t: t.apply(lambda m: parallel_func(m["Sample"], m["index]), axis = 
  1)).compute(scheduler="threads")

But the same index ===>1 is passed to the port parameter and I am getting the same error as mentioned above. Any suggestions on how to solve this would be helpful

0 Answers
Related