Using spacy within pyspark function

[noahberhe]

Please can anyone offer some advice?

I am attempting to use spacy's entity recognition on a Spark dataframe in python.
While it works on a sample text, it fails to recognise spacy when entity recognition is wrapped inside a pyspark function, ie:

This works:

import spacy
import en_core_web_sm
nlp = en_core_web_sm.load()
nlp("Harry Potter is a great movie").ents

This doesn't:

def spacy_ner(column):
    entities = ''
    for row in column:
        for ent in nlp(row).ents:
            entities += ent.text + ' '
    return entities

ner = udf(spacy_ner)
ner = spark.udf.register("ner", spacy_ner)
test = tweets.withColumn('text_ner',ner('text_nosw'))
test.show(5)

Giving the below error:

Py4JJavaError: An error occurred while calling o904.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 55.0 failed 1 times, most recent failure: Lost task 0.0 in stage 55.0 (TID 100, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 366, in main
    func, profiler, deserializer, serializer = read_udfs(pickleSer, infile, eval_type)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 241, in read_udfs
    arg_offsets, udf = read_single_udf(pickleSer, infile, eval_type, runner_conf)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 168, in read_single_udf
    f, return_type = read_command(pickleSer, infile)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 71, in read_command
    command = serializer.loads(command.value)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py", line 587, in loads
    return pickle.loads(obj, encoding=encoding)
ModuleNotFoundError: No module named 'spacy'

	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:456)
	at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:81)
	at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:64)
	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:410)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:123)
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1891)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1879)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1878)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1878)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:927)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:927)
	at scala.Option.foreach(Option.scala:257)
	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:927)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2112)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2061)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2050)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
	at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:738)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
	at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:365)
	at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
	at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3389)
	at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2550)
	at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2550)
	at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3370)
	at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:80)
	at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:127)
	at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:75)
	at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3369)
	at org.apache.spark.sql.Dataset.head(Dataset.scala:2550)
	at org.apache.spark.sql.Dataset.take(Dataset.scala:2764)
	at org.apache.spark.sql.Dataset.getRows(Dataset.scala:254)
	at org.apache.spark.sql.Dataset.showString(Dataset.scala:291)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at py4j.Gateway.invoke(Gateway.java:282)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 366, in main
    func, profiler, deserializer, serializer = read_udfs(pickleSer, infile, eval_type)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 241, in read_udfs
    arg_offsets, udf = read_single_udf(pickleSer, infile, eval_type, runner_conf)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 168, in read_single_udf
    f, return_type = read_command(pickleSer, infile)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 71, in read_command
    command = serializer.loads(command.value)
  File "/Users/Noah/anaconda3/envs/pyspark_env/lib/python3.7/site-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py", line 587, in loads
    return pickle.loads(obj, encoding=encoding)
ModuleNotFoundError: No module named 'spacy'

	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:456)
	at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:81)
	at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:64)
	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:410)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:123)
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	... 1 more

• spaCy version: 2.2.4
• Platform: Darwin-19.3.0-x86_64-i386-64bit
• Python version: 3.7.4
• **Pyspark Version Used: 2.4.5

[BramVanroy]

This might be related to your PySpark config. Does this help? https://stackoverflow.com/questions/32336498/pyspark-module-not-found

[dave-espinosa]

Hello @BramVanroy

I have the same issue, however with different software versions:

spaCy version: 3.2.1
Platform: Ubuntu 18.04.6 LTS
Python version: 3.7.4
Pyspark Version Used: 2.4.8
Dataproc version: 2.0

Does your suggested info also apply for the setting mentioned above?

Thanks!

UPDATE: Just for the record, pip freeze displays the following list of packages installed in my environment:

aiohttp==3.8.1
aiosignal==1.2.0
alabaster==0.7.12
alembic==1.7.5
anaconda-client==1.7.2
anaconda-navigator==1.9.7
anaconda-project==0.8.3
ansiwrap==0.8.4
appdirs==1.4.4
arrow==1.2.2
asn1crypto==1.0.1
astroid==2.3.1
astropy==3.2.2
async-generator==1.10
async-timeout==4.0.2
asyncio==3.4.3
asynctest==0.13.0
atomicwrites==1.3.0
attrs==21.4.0
Babel==2.7.0
backcall==0.1.0
backports.functools-lru-cache==1.5
backports.os==0.1.1
backports.shutil-get-terminal-size==1.0.0
backports.tempfile==1.0
backports.weakref==1.0.post1
bcolz==1.2.1
beautifulsoup4==4.8.0
binaryornot==0.4.4
bitarray==1.0.1
bkcharts==0.2
black==22.6.0
bleach==3.1.0
blinker==1.4
blis==0.7.8
bokeh==1.3.4
boto==2.49.0
Bottleneck==1.2.1
bs4==0.0.1
cachetools==5.2.0
catalogue==2.0.8
certifi==2019.9.11
certipy==0.1.3
cffi==1.12.3
chardet==3.0.4
charset-normalizer==2.1.0
click==8.1.3
cloudpickle==1.2.2
clyent==1.2.2
colorama==0.4.1
conda==4.7.12
conda-build==3.18.9
conda-package-handling==1.6.0
conda-verify==3.4.2
confuse==1.4.0
contextlib2==0.6.0
cookiecutter==1.7.3
cryptography==2.7
cycler==0.10.0
cymem==2.0.6
Cython==0.29.13
cytoolz==0.10.0
dask==2.5.2
decorator==4.4.0
defusedxml==0.6.0
distlib==0.3.5
distributed==2.5.2
docker==5.0.3
docutils==0.15.2
entrypoints==0.3
et-xmlfile==1.0.1
fairing==0.5.3
fastcache==1.1.0
filelock==3.0.12
findspark==2.0.1
Flask==1.1.1
frozenlist==1.3.0
fsspec==2022.5.0
future==0.17.1
gcsfs==2022.5.0
gevent==1.4.0
gitdb==4.0.9
GitPython==3.1.27
glob2==0.7
gmpy2==2.0.8
google-api-core==1.32.0
google-api-python-client==1.11.0
google-auth==2.9.1
google-auth-httplib2==0.1.0
google-auth-oauthlib==0.5.2
google-cloud-bigquery==1.26.1
google-cloud-bigquery-storage==1.0.0
google-cloud-core==1.4.4
google-cloud-dataproc==1.1.3
google-cloud-datastore==1.15.5
google-cloud-language==1.3.2
google-cloud-logging==1.15.3
google-cloud-spanner==1.18.0
google-cloud-storage==1.31.2
google-cloud-translate==2.0.4
google-crc32c==1.3.0
google-resumable-media==1.3.3
googleapis-common-protos==1.56.4
greenlet==0.4.15
grpc-google-iam-v1==0.12.4
grpcio==1.31.0
h5py==2.9.0
HeapDict==1.0.1
html5lib==1.0.1
htmlmin==0.1.12
httplib2==0.18.1
idna==2.8
ImageHash==4.2.1
imageio==2.6.0
imagesize==1.1.0
importlib-metadata==0.23
ipykernel==5.1.2
ipyparallel==6.3.0
ipython==7.8.0
ipython-genutils==0.2.0
ipython-sql==0.4.1
ipywidgets==7.5.1
isort==4.3.21
itsdangerous==1.1.0
jdcal==1.4.1
jedi==0.15.1
jeepney==0.4.1
Jinja2==2.10.3
jinja2-time==0.2.0
joblib==0.13.2
json5==0.8.5
jsonschema==3.0.2
jupyter==1.0.0
jupyter-client==5.3.5
jupyter-console==6.0.0
jupyter-contrib-core==0.4.0
jupyter-contrib-nbextensions==0.5.1
jupyter-core==4.6.3
jupyter-gcs-contents-manager==0.0.1
jupyter-highlight-selected-word==0.2.0
jupyter-http-over-ws==0.0.8
jupyter-latex-envs==1.4.6
jupyter-nbextensions-configurator==0.5.0
jupyterhub==1.0.0
jupyterlab==1.1.4
jupyterlab-git==0.11.0
jupyterlab-server==1.0.6
keyring==18.0.0
kiwisolver==1.1.0
kubernetes==24.2.0
langcodes==3.3.0
lazy-object-proxy==1.4.2
libarchive-c==2.8
lief==0.9.0
llvmlite==0.29.0
locket==0.2.0
lxml==4.4.1
Mako==1.1.4
Markdown==3.2.2
MarkupSafe==1.1.1
matplotlib==3.1.1
mccabe==0.6.1
metakernel==0.27.5
missingno==0.4.3
mistune==0.8.4
mkl-fft==1.0.14
mkl-random==1.1.0
mkl-service==2.3.0
mock==3.0.5
more-itertools==7.2.0
mpmath==1.1.0
msgpack==0.6.1
multidict==6.0.2
multipledispatch==0.6.0
murmurhash==1.0.7
mypy-extensions==0.4.3
navigator-updater==0.2.1
nbclient==0.6.6
nbconvert==5.6.0
nbdime==1.0.7
nbformat==4.4.0
nest-asyncio==1.5.5
networkx==2.6.3
nltk==3.4.5
nose==1.3.7
notebook==6.0.3
numba==0.45.1
numexpr==2.7.0
numpy==1.21.6
numpydoc==0.9.1
oauth2client==4.1.3
oauthlib==3.1.0
olefile==0.46
opencv-python==4.2.0.34
openpyxl==3.0.0
packaging==21.3
pamela==1.0.0
pandas==1.2.5
pandas-profiling==2.9.0
pandocfilters==1.4.2
papermill==2.1.3
parso==0.5.1
partd==1.0.0
path.py==12.0.1
pathlib2==2.3.5
pathspec==0.9.0
pathy==0.6.2
patsy==0.5.1
pep8==1.7.1
pexpect==4.7.0
phik==0.12.0
pickleshare==0.7.5
Pillow==8.3.2
Pillow-SIMD==7.0.0.post4
pkginfo==1.5.0.1
platformdirs==2.5.2
pluggy==0.13.0
ply==3.11
portalocker==2.3.0
poyo==0.5.0
preshed==3.0.6
prettytable==0.7.2
prometheus-client==0.7.1
prompt-toolkit==2.0.10
protobuf==3.13.0
psutil==5.6.3
ptyprocess==0.6.0
py==1.8.0
py4j==0.10.7
pyarrow==1.0.0
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycodestyle==2.5.0
pycosat==0.6.3
pycparser==2.19
pycrypto==2.6.1
pycurl==7.43.0.3
pydantic==1.8.2
pydot==1.4.2
pyflakes==2.1.1
Pygments==2.4.2
PyJWT==2.0.1
pylint==2.4.2
pyodbc==4.0.27
pyOpenSSL==19.0.0
pyparsing==2.4.2
pyrsistent==0.15.4
PySocks==1.7.1
# Editable install with no version control (pyspark==2.4.8)
-e /usr/lib/spark/python
pytest==5.2.1
pytest-arraydiff==0.3
pytest-astropy==0.5.0
pytest-doctestplus==0.4.0
pytest-openfiles==0.4.0
pytest-remotedata==0.3.2
python-dateutil==2.8.0
python-slugify==6.1.2
pytz==2019.3
PyWavelets==1.0.3
PyYAML==5.1.2
pyzmq==18.1.0
QtAwesome==0.6.0
qtconsole==4.5.5
QtPy==1.9.0
requests==2.28.1
requests-oauthlib==1.3.1
rope==0.14.0
rsa==4.6
ruamel-yaml==0.15.46
scikit-image==0.15.0
scikit-learn==0.21.3
scipy==1.3.1
seaborn==0.9.0
SecretStorage==3.1.1
Send2Trash==1.5.0
simplegeneric==0.8.1
singledispatch==3.4.0.3
six==1.15.0
smart-open==5.2.1
smmap==5.0.0
snowballstemmer==2.0.0
sortedcollections==1.1.2
sortedcontainers==2.1.0
soupsieve==1.9.3
spacy==3.2.1
spacy-legacy==3.0.9
spacy-loggers==1.0.3
sparkmonitor-s==0.0.22
Sphinx==2.2.0
sphinxcontrib-applehelp==1.0.1
sphinxcontrib-devhelp==1.0.1
sphinxcontrib-htmlhelp==1.0.2
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-qthelp==1.0.2
sphinxcontrib-serializinghtml==1.1.3
sphinxcontrib-websupport==1.1.2
spyder==3.3.6
spyder-kernels==0.5.2
spylon==0.3.0
spylon-kernel==0.4.1
SQLAlchemy==1.3.9
sqlparse==0.4.2
srsly==2.4.4
statsmodels==0.10.1
sympy==1.4
tables==3.5.2
tangled-up-in-unicode==0.2.0
tblib==1.4.0
tenacity==7.0.0
terminado==0.8.2
testpath==0.4.2
text-unidecode==1.3
textwrap3==0.9.2
thinc==8.0.17
tomli==2.0.1
toolz==0.10.0
tornado==6.0.3
tqdm==4.64.0
traitlets==4.3.3
typed-ast==1.5.4
typer==0.4.2
typing-extensions==3.10.0.2
unicodecsv==0.14.1
uritemplate==3.0.1
urllib3==1.24.2
virtualenv==20.0.35
visions==0.5.0
wasabi==0.10.1
wcwidth==0.1.7
webencodings==0.5.1
websocket-client==1.3.3
Werkzeug==0.16.0
widgetsnbextension==3.5.1
wrapt==1.11.2
wurlitzer==1.0.3
xlrd==1.2.0
XlsxWriter==1.2.1
xlwt==1.3.0
yarl==1.7.2
zict==1.0.0
zipp==0.6.0

[radiatechs-himanshu]

Its not clear from the above stack overflow post