Spark-nlp: java.lang.NoClassDefFoundError when using Dataproc cluster on GCP

[johanmoen]

Description

I'm trying to run a simple example using a pre-trained pipeline from the Spark NLP library. I get an error when I'm downloading the pipeline. I'm using python 3, spark-nlp 2.6.0, and appache spark 2.4.6. The code is running on a Dataproc cluster on GCP.

Error

Py4JJavaError Traceback (most recent call last)
in
----> 1 pipeline = PretrainedPipeline('explain_document_dl', lang='en')

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/pretrained.py in __init__(self, name, lang, remote_loc, parse_embeddings, disk_location)
89 def __init__(self, name, lang='en', remote_loc=None, parse_embeddings=False, disk_location=None):
90 if not disk_location:
---> 91 self.model = ResourceDownloader().downloadPipeline(name, lang, remote_loc)
92 else:
93 self.model = PipelineModel.load(disk_location)

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/pretrained.py in downloadPipeline(name, language, remote_loc)
49 def downloadPipeline(name, language, remote_loc=None):
50 print(name + " download started this may take some time.")
---> 51 file_size = _internal._GetResourceSize(name, language, remote_loc).apply()
52 if file_size == "-1":
53 print("Can not find the model to download please check the name!")

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/internal.py in __init__(self, name, language, remote_loc)
190 def __init__(self, name, language, remote_loc):
191 super(_GetResourceSize, self).__init__(
--> 192 "com.johnsnowlabs.nlp.pretrained.PythonResourceDownloader.getDownloadSize", name, language, remote_loc)
193
194

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/internal.py in __init__(self, java_obj, *args)
127 super(ExtendedJavaWrapper, self).__init__(java_obj)
128 self.sc = SparkContext._active_spark_context
--> 129 self._java_obj = self.new_java_obj(java_obj, *args)
130 self.java_obj = self._java_obj
131

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/internal.py in new_java_obj(self, java_class, *args)
137
138 def new_java_obj(self, java_class, *args):
--> 139 return self._new_java_obj(java_class, *args)
140
141 def new_java_array(self, pylist, java_class):

/usr/lib/spark/python/pyspark/ml/wrapper.py in _new_java_obj(java_class, args)
65 java_obj = getattr(java_obj, name)
66 java_args = [_py2java(sc, arg) for arg in args]
---> 67 return java_obj(java_args)
68
69 @staticmethod

/opt/conda/anaconda/lib/python3.7/site-packages/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:

/usr/lib/spark/python/pyspark/sql/utils.py in deco(a, *kw)
61 def deco(a, *kw):
62 try:
---> 63 return f(a, *kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()

/opt/conda/anaconda/lib/python3.7/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(

Py4JJavaError: An error occurred while calling z:com.johnsnowlabs.nlp.pretrained.PythonResourceDownloader.getDownloadSize.
: java.lang.NoClassDefFoundError: org/apache/spark/ml/util/MLReadable$class
at com.johnsnowlabs.nlp.DocumentAssembler$.(DocumentAssembler.scala:178)
at com.johnsnowlabs.nlp.DocumentAssembler$.(DocumentAssembler.scala)
at com.johnsnowlabs.nlp.pretrained.PythonResourceDownloader$.(ResourceDownloader.scala:431)
at com.johnsnowlabs.nlp.pretrained.PythonResourceDownloader$.(ResourceDownloader.scala)
at com.johnsnowlabs.nlp.pretrained.PythonResourceDownloader.getDownloadSize(ResourceDownloader.scala)
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: java.lang.ClassNotFoundException: org.apache.spark.ml.util.MLReadable$class
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
... 16 more

[maziyarpanahi]

Please complete the template, it is required in order to narrow down the issue. (especially the OS)

PS: On a separate note, I would appreciate it if you can test explain_document_ml just as a test and then also check the firewall rules on your GCP. The last test would be downloading the pipeline directly from here and use load instead (to make sure the network connectivity is not an issue): https://github.com/JohnSnowLabs/spark-nlp-models#english---pipelines

After extract:

pipeline = PretrainedPipeline.from_disk('/tmp/explain_document_dl_en_2.4.3_2.4_1584626657780')

[johanmoen]

Thanks for your prompt reply. Regarding your suggestions:

• explain_document_ml produces the same error
• the firewall on the virtual machines allows HTTP and HTTPS traffic
• manually downloading and calling the pipeline also produces an error

Below is an overview of my environment and the error message I get when loading the pipeline locally. Let me know if you require any additional information.

Environment

• Spark NLP version: 2.6.0
• Apache Spark version: 2.4.6
• Java version (java -version):
  
  
  
  • openjdk version "1.8.0_265"
  
  
  • OpenJDK Runtime Environment (AdoptOpenJDK)(build 1.8.0_265-b01)
  
  
  • OpenJDK 64-Bit Server VM (AdoptOpenJDK)(build 25.265-b01, mixed mode)
  
  
• Setup and installation:
  
  
  
  • conda 4.8.4
  
  
  • Python 3.7.4
  
  
  • pip 19.2.3
  
  
• Operating System and version:
  
  
  
  • Dataproc Image version: 1.5.14-debian10
  
  

Error (manual download)

Py4JJavaError Traceback (most recent call last)
in
1 #pipeline = PretrainedPipeline.from_disk('/tmp/explain_document_dl_en_2.4.3_2.4_1584626657780')
----> 2 pipeline = PretrainedPipeline.from_disk('gs://measuring-automation/notebooks/jupyter/tmp/explain_document_dl_en_2.4.3_2.4_1584626657780')

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/pretrained.py in from_disk(path, parse_embeddings)
96 @staticmethod
97 def from_disk(path, parse_embeddings=False):
---> 98 return PretrainedPipeline(None, None, None, parse_embeddings, path)
99
100 def annotate(self, target, column=None):

/opt/conda/anaconda/lib/python3.7/site-packages/sparknlp/pretrained.py in __init__(self, name, lang, remote_loc, parse_embeddings, disk_location)
91 self.model = ResourceDownloader().downloadPipeline(name, lang, remote_loc)
92 else:
---> 93 self.model = PipelineModel.load(disk_location)
94 self.light_model = LightPipeline(self.model, parse_embeddings)
95

/usr/lib/spark/python/pyspark/ml/util.py in load(cls, path)
360 def load(cls, path):
361 """Reads an ML instance from the input path, a shortcut of read().load(path)."""
--> 362 return cls.read().load(path)
363
364

/usr/lib/spark/python/pyspark/ml/pipeline.py in load(self, path)
240 metadata = DefaultParamsReader.loadMetadata(path, self.sc)
241 if 'language' not in metadata['paramMap'] or metadata['paramMap']['language'] != 'Python':
--> 242 return JavaMLReader(self.cls).load(path)
243 else:
244 uid, stages = PipelineSharedReadWrite.load(metadata, self.sc, path)

/usr/lib/spark/python/pyspark/ml/util.py in load(self, path)
298 if not isinstance(path, basestring):
299 raise TypeError("path should be a basestring, got type %s" % type(path))
--> 300 java_obj = self._jread.load(path)
301 if not hasattr(self._clazz, "_from_java"):
302 raise NotImplementedError("This Java ML type cannot be loaded into Python currently: %r"

/opt/conda/anaconda/lib/python3.7/site-packages/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:

/usr/lib/spark/python/pyspark/sql/utils.py in deco(a, *kw)
61 def deco(a, *kw):
62 try:
---> 63 return f(a, *kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()

/opt/conda/anaconda/lib/python3.7/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(

Py4JJavaError: An error occurred while calling o221.load.
: java.lang.NoClassDefFoundError: Could not initialize class com.johnsnowlabs.nlp.DocumentAssembler$
at com.johnsnowlabs.nlp.DocumentAssembler.read(DocumentAssembler.scala)
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 org.apache.spark.ml.util.DefaultParamsReader$.loadParamsInstance(ReadWrite.scala:652)
at org.apache.spark.ml.Pipeline$SharedReadWrite$.$anonfun$load$1(Pipeline.scala:274)
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)
at scala.collection.TraversableLike.map(TraversableLike.scala:238)
at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:198)
at org.apache.spark.ml.Pipeline$SharedReadWrite$.load(Pipeline.scala:272)
at org.apache.spark.ml.PipelineModel$PipelineModelReader.load(Pipeline.scala:348)
at org.apache.spark.ml.PipelineModel$PipelineModelReader.load(Pipeline.scala:342)
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)

[maziyarpanahi]

Thanks, the second error when you are loading it offline is different. : java.lang.NoClassDefFoundError: Could not initialize class com.johnsnowlabs.nlp.DocumentAssembler$

This suggests the JAR/Java part which comes witch sparknlp.start() didn't load properly. Could you please make sure:

• The process/kernel in your jupyter is restarted correctly and everything is clear
• In the first like try spark.version just to be sure there is no such thing as spark or SparkSession. Sometimes those notebooks come with SparkSession which the second time will just reuse it. (this should fail)
• instead of sparknlp.start(), please run the following:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Spark NLP")\
    .master("local[4]")\
    .config("spark.driver.memory","16G")\
    .config("spark.driver.maxResultSize", "2G") \
    .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.11:2.6.1")\
    .config("spark.kryoserializer.buffer.max", "1000M")\
    .getOrCreate()

There is a place that you can see the logs for the Jupyter notebook, in there you shouldn't see any error. I am guessing it fails in spark.jars.packages or fails to download one of the dependencies. In case it failed or you see any error in the logs while you where create SparakSession, please restart the kernel and try the offline:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Spark NLP")\
    .master("local[4]")\
    .config("spark.driver.memory","16G")\
    .config("spark.driver.maxResultSize", "2G") \
    .config("spark.jars", "/tmp/spark-nlp-assembly-2.6.1.jar")\
    .config("spark.kryoserializer.buffer.max", "1000M")\
    .getOrCreate()

You can download the fat JAR from https://s3.amazonaws.com/auxdata.johnsnowlabs.com/public/spark-nlp-assembly-2.6.1.jar

PS: you can see the SparkSession, by default it uses local as a master. So if you are in the cluster you should follow the correct setup here, if you are in a single machine you can ignore this.

[johanmoen]

I have verified that there is nothing defined as spark or SparkSession after I restart the kernel. When I run the code you provided, the code does not fail. However, the same errors occur when I try to load the pipeline (both for the direct and manual download).

I also tried to restart the kernel and create the session using the offline approach. Again, the code does not fail but I get the same errors as before when I try to load the pipeline. Any thoughts on what can be causing this?

[maziyarpanahi]

@johanmoen Thanks for the update. Could you please provide the full errors especially when you are creating the SparkSession with offline JAR? (if you can also look into the console logs/runtime logs to see if there is an error during SparkSession)

There are two different errors in this issue so I am not sure which one will happen when. Unfortunately, if the JAR is missing or there is an error in some packages SparkSession will not fail. It gives the impression that everything went well but it only logs it in the log as what really happens. (it's a bit permissive and hard to be sure the JAR was really accessible and used in the session).

We do have users from Dataproc I remember on Slack and they don't have this issue. Your full error stack and logs from creating the SparkSession would really help to narrow this down

[johanmoen]

It sounds like the runtime logs are quite central here. Unfortunately I'm a bit of a newbie when it comes to Dataproc; I'm not sure how to access the logs from the Jupyter notebook (the cloud logging module is mainly concerned with hadoop/spark logs and the terminal of the vm does not display any activity). Do you have any idea about how to access the logs?

[maziyarpanahi]

May I ask how you create the cluster and your associated bucket on GCP? I remember this article was also helpful in that regard: https://medium.com/google-cloud/apache-spark-and-jupyter-notebooks-made-easy-with-dataproc-component-gateway-fa91d48d6a5a

[johanmoen]

Indeed, I did use that article as a reference when setting up the cluster. I did some adjustments to include spark-nlp and to add a service account for authorization. For my current cluster I used the following setup:

gcloud beta dataproc clusters create ${CLUSTER_NAME} \
 --region=${REGION} \
 --service-account=<some_account>@project.iam.gserviceaccount.com \
 --image-version=1.5 \
 --metadata 'PIP_PACKAGES=google-cloud-storage spark-nlp' \
 --master-machine-type=n1-standard-4 \
 --worker-machine-type=n1-standard-4 \
 --bucket=${BUCKET_NAME} \
 --initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh \
 --optional-components=ANACONDA,JUPYTER \
 --enable-component-gateway 

I'm using the same bucket that I've used with some other virtual machines. I'm able to load files from the bucket, so I don't think there are any issues with that.

[maziyarpanahi]

Hi @johanmoen

I think we have users on Slack who can guide you easily in how to first create your cluster including spark-nlp first. This one pops up today: https://spark-nlp.slack.com/archives/CA118BWRM/p1600436541009500?thread_ts=1600434933.008200&cid=CA118BWRM

If you search you can find more or ask a new question so other users can help.

[johanmoen]

Hi @maziyarpanahi
That post was quite useful. When I follow the setup of this guy it works smoothly. Thanks a lot!

[maziyarpanahi]

Fantastic, thanks for the update. I am leaving the post here just in case for the future users:

Hey thanks for the speedy response I fixed the issue. This is what I did. I built spark NLP when creating cluster from cloud-shell.

• Image version 1.4 with debian to address lack of scala in class path

Just to be safe I also built dependency at runtime this way

spark = SparkSession.builder \
  .appName('Jupyter BigQuery Storage').config('gs://spark-lib/bigquery/spark-bigquery-latest.jar')\
    .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.11:2.6.1,com.google.cloud.spark:spark-bigquery-with-dependencies_2.11:0.15.1-beta")\
  .getOrCreate()