Closed jrwiebe closed 8 years ago
That makes sense.
If NER3Classifier were an object, we could pass the classifier file path to every instance of the classifier
method. But it would be loaded many times then..
I've changed the NER3Classifier from an object to a class to be passed around. The driver 'creates' the NER3Classifier class with a specific classifier, but the classifier file is loaded 'locally' by worker nodes (when classifier.classify() is first called). Would this work?
Try this?
import org.warcbase.spark.matchbox.NER3Classifier
val classifier = new NER3Classifier("hdfs:///user/jrwiebe/english.all.3class.distsim.crf.ser.gz")
val entities = sc.textFile("hdfs:///user/jrwiebe/cpp.text-greenparty/part-m-00000")
.map(_.split("\t"))
.map(e => (e(0), e(1), classifier.classify(e(2))))
val entities = rdd.map(r => (r._1, r._2, classifier_.classify(r._3)))
entities.take(3).foreach(println)
I also tried this yesterday, making NER3Classifier a class extending Serializable -- although I didn't think to move the NERClassType object definition and the ObjectMapper instantiation. It still isn't working for me, though. Running the above code (minus the second-last line), I'm still getting an error loading the classifier. You'll notice it's a FileNotFoundException, although the file is definitely where it's specified.
WARN TaskSetManager - Lost task 0.0 in stage 0.0 (TID 0, trantor04.umiacs.umd.edu): java.lang.ExceptionInInitializerError: Unable to load classifier java.io.FileNotFoundException: hdfs:/user/jrwiebe/english.all.3class.distsim.crf.ser.gz (No such file or directory)
at org.warcbase.spark.matchbox.NER3Classifier.classify(NER3Classifier.scala:78)
at $line15.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(<console>:27)
at $line15.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(<console>:27)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$33.apply(RDD.scala:1177)
at org.apache.spark.rdd.RDD$$anonfun$33.apply(RDD.scala:1177)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1503)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1503)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
at org.apache.spark.scheduler.Task.run(Task.scala:64)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
...
Hmm.. that's really odd. It works for me, although the classifier is local, and not on HDFS.
Does CRFClassifier.getClassifier("file")
also return the FileNotFoundException?
Also, my bad about the second-last line. Need to copy/paste more carefully.
Sorry for dragging you into this, Alice. Your original code worked fine; the issue was with my code, and not understanding how the workers handle files. Instead of trying to make the workers load the classifier from "hdfs://...", the thing to do is to use sc.addFile()
.
So, the following works, using your earlier code where NER3Classifier is an object (cc274ed):
sc.addFile("hdfs:///user/jrwiebe/english.all.3class.distsim.crf.ser.gz")
val entities = sc.textFile("hdfs:///user/jrwiebe/cpp.text-greenparty/part-m-00000")
.map(_.split("\t"))
.map{ e =>
NER3Classifier.apply("english.all.3class.distsim.crf.ser.gz")
(e(0), e(1), NER3Classifier.classify(e(2)))
}
Better, since we don't want to initialize the classifier for every string, is this:
sc.addFile("hdfs:///user/jrwiebe/english.all.3class.distsim.crf.ser.gz")
val entities = sc.textFile("hdfs:///user/jrwiebe/cpp.text-greenparty/part-m-00000")
.mapPartitions(iter => {
NER3Classifier.apply("english.all.3class.distsim.crf.ser.gz")
iter.map(_.split("\t"))
.map(e => (e(0), e(1), NER3Classifier.classify(e(2))))
})
Without modification, the updated class version of NER3Classifier (582b21a) doesn't work with mapPartitions(), though it does work with straight map(). I suggest we revert to the object version.
@jrwiebe Dunno if this helps, but the reason why your original code didn't work was this: in your original code, you initialized something in the shell, which is local, and then you referred to it inside map
, which runs in parallel in each executor (possibly remotely over a large cluster). The only way that could possible work is if Spark could "package up" the NER val and send to the executors, but it didn't know how...
It was fixed in the later code because you had the NER initialized from within the closure inside the map. Yes, this means that each executor has to separately initialize the NER, but there's no way around that --- remember, in principle, the map could be running on hundreds of nodes... make sense?
Thanks, @lintool. I do understand all of this now, and that was my basic understanding when I opened this issue, but I couldn't figure out how to serialize the NER classifier, and I couldn't manage to initialize a classifier within the closure either (the problem here being trying to access an hdfs-path).
For my part, I think we can close this issue. You can ignore my comment above about the class version of NER3Classifier not working. I just tested it again, and it works fine. I wonder if I polluted the namespace in the REPL by testing first with the NER3Classifier object and then loading the class definition. Anyway, this works with the current (class) code:
import org.warcbase.spark.matchbox.NER3Classifier
sc.addFile("hdfs:///user/jrwiebe/english.all.3class.distsim.crf.ser.gz")
val entities = sc.textFile("hdfs:///user/jrwiebe/cpp.text-greenparty/part-m-00000")
.mapPartitions(iter => {
val classifier = new NER3Classifier("english.all.3class.distsim.crf.ser.gz")
iter.map(_.split("\t"))
.map(e => (e(0), e(1), classifier.classify(e(2))))
})
@jrwiebe Can you please document this knowledge appropriately in the wiki?
Closing this issue.
Done.
On Sun, Nov 22, 2015 at 3:09 PM, Jimmy Lin notifications@github.com wrote:
@jrwiebe https://github.com/jrwiebe Can you please document this knowledge appropriately in the wiki?
Closing this issue.
— Reply to this email directly or view it on GitHub https://github.com/lintool/warcbase/issues/167#issuecomment-158794331.
This issue refers to @aliceranzhou's ner3classifier branch. Also, I'm a novice at Scala and Spark, so forgive me if what I've labelled an issue can be chalked up to my ignorance.
I tried writing a straight port of my
extract-entities-from-scrape-text.pig
script to test NER3Classifier.scala on the cluster. It failed, it would appear because the NER3Classifier object is not serializable.I ran the following in the Spark shell:
I received the following error messages:
I eventually concluded that the problem was with evaluation of the second map closure. This seemed to be confirmed when I pasted
NER3Classifier.scala
(minus thepackage
declaration) into the Spark shell, replacing the import statement in my script. In this instance a "Task not serializable" exception was thrown instead of a NullPointerException. (I assume the difference is due to the information available to the interpreter, as opposed to what the bytecode produced by the compiler.)I tried modifying NER3Classifier, but I didn't get anywhere. It's not even clear to me that I'm taking the right approach. It seems to me that rather than the driver creating the NER3Classifier object, loading the classifier file, and somehow getting serialized and then passed to the worker nodes, each worker node should rather get a function that would create the NER3Classifier object 'locally'. I tried this by passing the map a function, but I got the same "Task not serializable" error.