Spark-nlp: Known issue with ClassifierDLModel inside cluster

[maziyarpanahi]

The ClassifierDLModel has an issue to be used immediately right after it's trained inside a closer. It needs to be saved and then be loaded back.

While working on a fix, there is a workaround by saving and loading it back before using it inside the cluster:

from pyspark.ml import Pipeline

from sparknlp.annotator import *
from sparknlp.common import *
from sparknlp.base import *
from pyspark.sql.types import StringType
from sklearn.metrics import classification_report

# actual content is inside description column
document = DocumentAssembler()\
    .setInputCol("description")\
    .setOutputCol("document")

use = UniversalSentenceEncoder.pretrained() \
 .setInputCols(["document"])\
 .setOutputCol("sentence_embeddings")

# the classes/labels/categories are in category column
classsifierdl = ClassifierDLApproach()\
  .setInputCols(["sentence_embeddings"])\
  .setOutputCol("class")\
  .setLabelColumn("category")\
  .setMaxEpochs(5)\
  .setEnableOutputLogs(True)

pipeline = Pipeline(
    stages = [
        document,
        use,
        classsifierdl
    ])

pipelineModel = pipeline.fit(trainDataset)

# dbfs:/ or hdfs:/ if you are saving it on distributed file systems
pipelineModel.stages[-1].write().overwrite().save('tmp_classifierDL_model')

# In a new pipeline you can load it for prediction
document = DocumentAssembler()\
    .setInputCol("description")\
    .setOutputCol("document")

use = UniversalSentenceEncoder.pretrained() \
 .setInputCols(["document"])\
 .setOutputCol("sentence_embeddings")

classsifierdl = ClassifierDLModel.load("tmp_classifierDL_model") \
  .setInputCols(["sentence_embeddings"])\
  .setOutputCol("class")

pipeline = Pipeline(
    stages = [
        document,
        use,
        classsifierdl
    ])

preds = pipeline.fit(testDataset).transform(testDataset)
preds.select('category','description',"class.result").show(50, truncate=50)


preds_df = preds.select('category','description',"class.result").toPandas()
preds_df['result'] = preds_df['result'].apply(lambda x : x[0])
print (classification_report(preds_df['result'], preds_df['category']))

[IgorDimovski]

Hi @maziyarpanahi,

I've tried this approach and everything is ok with saving and loading the last stage of the pipeline, but after that, trying to run: preds = pipeline.fit(testDataset).transform(testDataset), there is an error on my Databrics cluster:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 15.0 failed 4 times, most recent failure: Lost task 0.3 in stage 15.0 (TID 264, 10.225.248.206, executor 6): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$dfAnnotate$1: (array<array<struct<annotatorType:string,begin:int,end:int,result:string,metadata:map<string,string>,embeddings:array<float>>>>) => array<struct<annotatorType:string,begin:int,end:int,result:string,metadata:map<string,string>,embeddings:array<float>>>)

and amongst all exception messages, i find this one as most relevant:

Caused by: org.tensorflow.TensorFlowException: not an sstable (bad magic number)
[[{{node save/RestoreV2}}]]
at org.tensorflow.Session.run(Native Method)
at org.tensorflow.Session.access$100(Session.java:48)
at org.tensorflow.Session$Runner.runHelper(Session.java:326)
at org.tensorflow.Session$Runner.run(Session.java:276)
at com.johnsnowlabs.ml.tensorflow.TensorflowWrapper.getSession(TensorflowWrapper.scala:64)
at com.johnsnowlabs.ml.tensorflow.TensorflowClassifier.predict(TensorflowClassifier.scala:130)
at com.johnsnowlabs.nlp.annotators.classifier.dl.ClassifierDLModel.annotate(ClassifierDLModel.scala:82)
at com.johnsnowlabs.nlp.AnnotatorModel$$anonfun$dfAnnotate$1.apply(AnnotatorModel.scala:35)
at com.johnsnowlabs.nlp.AnnotatorModel$$anonfun$dfAnnotate$1.apply(AnnotatorModel.scala:34)

[maziyarpanahi]

This one will be very hard to debug, I just tested the same workaround on a Databricks cluster couple of times. Could you please provide the entire code that results in this error so I see if I can reproduce it?

[IgorDimovski]

it's the same code above, as in the original post of this issue

[maziyarpanahi]

The only thing different in that code from the one I ran couple of times on a cluster is tmp_classifierDL_model actually dbfs:/tmp_classifierDL_model. The rest is the same. Are you saving it on DBFS?

[IgorDimovski]

yes, the files are there, i can see them

[maziyarpanahi]

I run it one more time on a clean cluster to see if it crashes, meanwhile if you can remove any existing model you saved and do the same that could be helpful.

[IgorDimovski]

I tried that and everything is ok now. Previously I saved the model not directly to the root, but in the folder i've created.

Thanks and sorry for bothering.

[maziyarpanahi]

No bother at all! Thanks for finding the bug and confirming the work around, much appreciate it.
Please continue if something comes up or a any new issues.

[tommasogiannantonio]

Hi,
sorry the workaround is still causing me problems...

I follow your suggested steps on Databricks, and when I try to save the model with
use_pipelineModel.stages[-1].write().overwrite().save('dbfs:/tmp_classifierDL_model')
, I get the following error:

An error occurred while calling o14301.save. : java.io.IOException: org.tensorflow.TensorFlowException: Unsuccessful TensorSliceReader constructor: Failed to find any matching files for /local_disk0/tmp/6cd48a845085_ner6081631778754106369/variables [[{{node save/RestoreV2}}]]

Part of the model files are written on dbfs:
dbfs:/tmp_classifierDL_model/fields/ dbfs:/tmp_classifierDL_model/metadata/
but the variables folders is somehow written on the local fs, and only as a temp file, i.e. as
/local_disk0/tmp/d15179e7a61c_ner6357169637809964979/variables_temp_ca197b34f2a0452b9ce8a8d4f4c004bc, then it is looked up as /local_disk0/tmp/d15179e7a61c_ner6357169637809964979/variables and not found...

So the model is only partially written and then the process
use_pipelineModel.stages[-1].write().overwrite().save('dbfs:/tmp_classifierDL_model')
crashes.

Do you have any insight about what could be happening? Thanks!

[maziyarpanahi]

Hi @tommasogiannantonio
This might be about what actually sits behind your dbfs:/tmp_classifierDL_model. Depending on the backend, this can be a local file system or object storage etc.

Could you please try to save the model in a local file system as follow:

use_pipelineModel.stages[-1].write().overwrite().save('/FileStore/tmp_classifierDL_model')

[ajhepburn]

Py4JJavaError: An error occurred while calling o655.count.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4.0 failed 1 times, most recent failure: Lost task 0.0 in stage 4.0 (TID 26, localhost, executor driver): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$dfAnnotate$1: (array,embeddings:array>>>) => array,embeddings:array>>)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificMutableProjection.apply(Unknown Source)
at org.apache.spark.sql.execution.python.EvalPythonExec$$anonfun$doExecute$1$$anonfun$5.apply(EvalPythonExec.scala:124)
at org.apache.spark.sql.execution.python.EvalPythonExec$$anonfun$doExecute$1$$anonfun$5.apply(EvalPythonExec.scala:122)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
at scala.collection.Iterator$GroupedIterator.takeDestructively(Iterator.scala:1074)
at scala.collection.Iterator$GroupedIterator.go(Iterator.scala:1089)
at scala.collection.Iterator$GroupedIterator.fill(Iterator.scala:1127)
at scala.collection.Iterator$GroupedIterator.hasNext(Iterator.scala:1130)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:224)
at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.writeIteratorToStream(PythonUDFRunner.scala:50)
at org.apache.spark.api.python.BasePythonRunner$WriterThread$$anonfun$run$1.apply(PythonRunner.scala:346)
at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1945)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:195)
Caused by: java.lang.NullPointerException
at com.johnsnowlabs.nlp.annotators.tokenizer.wordpiece.BasicTokenizer.isPunctuation(BasicTokenizer.scala:33)
at com.johnsnowlabs.nlp.annotators.tokenizer.wordpiece.BasicTokenizer.tokenize(BasicTokenizer.scala:94)
at com.johnsnowlabs.nlp.embeddings.BertEmbeddings$$anonfun$tokenize$1.apply(BertEmbeddings.scala:234)
at com.johnsnowlabs.nlp.embeddings.BertEmbeddings$$anonfun$tokenize$1.apply(BertEmbeddings.scala:233)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at com.johnsnowlabs.nlp.embeddings.BertEmbeddings.tokenize(BertEmbeddings.scala:233)
at com.johnsnowlabs.nlp.embeddings.BertEmbeddings.annotate(BertEmbeddings.scala:251)
at com.johnsnowlabs.nlp.AnnotatorModel$$anonfun$dfAnnotate$1.apply(AnnotatorModel.scala:35)
at com.johnsnowlabs.nlp.AnnotatorModel$$anonfun$dfAnnotate$1.apply(AnnotatorModel.scala:34)
... 17 more

I get this when trying to use pre-trained BERT embeddings, but weirdly only on my test set? So I've a feeling it might be to do with it trying to annotate entries with no tokens.

[demongolem]

Hi @tommasogiannantonio
This might be about what actually sits behind your dbfs:/tmp_classifierDL_model. Depending on the backend, this can be a local file system or object storage etc.

Could you please try to save the model in a local file system as follow:

use_pipelineModel.stages[-1].write().overwrite().save('/FileStore/tmp_classifierDL_model')

@maziyarpanahi I want to follow up on this one because I never saw a reply from @tommasogiannantonio . It seems I am having pretty much the identical problem. Let me recap:

There is a problem with ClassifierDL transforming immediately after fitting a model and I use embeddings so perhaps this occurs when embeddings is part of the pipeline. I get the Unsuccessful TensorSliceReader constructor message.

Then, a workaround was suggested to kick off this thread. When applied, I get the same error message. I am trying to save to dbfs:/ as above as I am working within a Databricks notebook.

Then you suggested to try a slightly different path and we never get a reply on that. Well, when I do it, again I get the same sort of error message.

The error message I get is java.io.IOException: org.tensorflow.TensorFlowException: Unsuccessful TensorSliceReader constructor: Failed to find any matching files for /local_disk0/tmp/0bac4e26604c_ner6828786924328354543/variables so as has been discussed variable and local storage which are not being found.

So is there a next step to debugging this? Have there been any insights or fixes in the last 6 months to address this issue?

[maziyarpanahi]

Hi @tommasogiannantonio
This might be about what actually sits behind your dbfs:/tmp_classifierDL_model. Depending on the backend, this can be a local file system or object storage etc.
Could you please try to save the model in a local file system as follow:

use_pipelineModel.stages[-1].write().overwrite().save('/FileStore/tmp_classifierDL_model')

I want to follow up on this one because I never saw a reply from @tommasogiannantonio . It seems I am having pretty much the identical problem. Let me recap:

There is a problem with ClassifierDL transforming immediately after fitting a model and I use embeddings so perhaps this occurs when embeddings is part of the pipeline. I get the Unsuccessful TensorSliceReader constructor message.

Then, a workaround was suggested to kick off this thread. When applied, I get the same error message. I am trying to save to dbfs:/ as above as I am working within a Databricks notebook.

Then you suggested to try a slightly different path and we never get a reply on that. Well, when I do it, again I get the same sort of error message.

The error message I get is java.io.IOException: org.tensorflow.TensorFlowException: Unsuccessful TensorSliceReader constructor: Failed to find any matching files for /local_disk0/tmp/0bac4e26604c_ner6828786924328354543/variables so as has been discussed variable and local storage which are not being found.

So is there a next step to debugging this? Have there been any insights or fixes in the last 6 months to address this issue?

Hi,

Sorry, I don't have any history of your issue. Could you please create a new issue, complete the template with code snippets so we can reproduce it? Please don't leave any details out as they are important and may not necessarily be related to this issue.

[jfernandrezj]

This is just happening in Azure for me