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