yugagarin / ctakesspark Goto Github PK

Inspired by UIMA collection processing engine...

We would need to make each RDD/DataFrame partition to be processed as a single collection by the pipeline similar how UIMA collection processing engine UI works now so that dictionaries and other resources are loaded once only per collection. Need to find sample code for this and add to CtakesFlatMapFunction call method.

Possible flow:

1. Implement CollectionReader over Spark partition (eg java.util.Iterator partition );
2. Run pipeline over the CollectionReader

public static void runPipeline(CollectionReader reader,
AnalysisEngine... engines)
throws UIMAException,
IOException
(http://uima.apache.org/d/uimafit-2.0.0/api/org/apache/uima/fit/pipeline/SimplePipeline.html )

There seem to be differences in output when submitting job to yarn cluster and running in local[ * ] modes. local[ * ] mode is closer to real data, while yarn cluster mode seem to be lossy...

/usr/bin/spark-submit --class org.poc.ctakes.spark.CtakesSparkMain --master yarn --deploy-mode cluster --conf spark.executor.extraClassPath=/tmp/ctakesdependencies/ --conf spark.driver.extraClassPath=/tmp/ctakesdependencies/ --conf spark.driver.memory=5g --executor-memory 10g spark-ctakes-0.1-shaded.jar

/usr/bin/spark-submit --class org.poc.ctakes.spark.CtakesSparkMain --master local[*] --conf spark.executor.extraClassPath=/tmp/ctakesdependencies/ --conf spark.driver.extraClassPath=/tmp/ctakesdependencies/ --conf spark.driver.memory=5g --executor-memory 10g spark-ctakes-0.1-shaded.jar

Hint: 50/50 - on 32 files sample there seems that half of results have one size, and the other half of the results have another size

Hi - how well was this working?
Why did development stop? It seems like a really neat idea to use stakes with spark, so why was it abandoned?

I followed the instruction , but when I submit the job. It fails with NPE at below

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6, wn0-sparkc.cloudapp.net, executor 3): java.lang.NullPointerException
        at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:1792)
        at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:1769)
        at org.apache.commons.io.IOUtils.copy(IOUtils.java:1744)
        at org.apache.commons.io.FileUtils.copyInputStreamToFile(FileUtils.java:1512)
        at org.apache.ctakes.dictionary.lookup.ae.UmlsDictionaryLookupAnnotator.createAnnotatorDescription(UmlsDictionaryLookupAnnotator.java:131)
        at org.apache.ctakes.clinicalpipeline.ClinicalPipelineFactory.getDefaultPipeline(ClinicalPipelineFactory.java:71)
        at org.poc.ctakes.spark.CtakesFlatMapFunction.setup(CtakesFlatMapFunction.java:67)
        at org.poc.ctakes.spark.CtakesFlatMapFunction.readObject(CtakesFlatMapFunction.java:74)

I am setting both driver and executor extra classpath parameters to the spark submit. Please let me know, if I am missing on anything.