Question

[Evegen55]

Hi guys!
Thank you for your tremendous efforts!
Could you explain for what purpose in class DistributedHDFSDataSet.java in constructor the class StringDataSource has been addeed? Seems like some not yet finished task. As for me it just very interesting because in an appropriate example SampleCovarianceCSR.java an object of class StringDataSource is actually created too - but there is no any issue if delete such functional.

[Evegen55]

Moreover, if save an object link as a class field, then exception will be thrown:

public class DistributedHDFSDataSet extends DistributedDataSet implements Serializable {
...
protected StringDataSource _dds; //new field

public DistributedHDFSDataSet(String filename, StringDataSource dds) {
    _filename = filename;
    this._dds = dds; //save an object link in a field
    }
...
}

ERROR:

20/01/26 17:59:14 ERROR Executor: Exception in task 1.0 in stage 0.0 (TID 1)
java.io.InvalidClassException: com.intel.daal.data_management.data_source.StringDataSource; no valid constructor
	at java.base/java.io.ObjectStreamClass$ExceptionInfo.newInvalidClassException(ObjectStreamClass.java:159)
	at java.base/java.io.ObjectStreamClass.checkDeserialize(ObjectStreamClass.java:864)
	at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2061)
	at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1594)
	at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2355)
	at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2249)
...
Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:1889)
	at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:1877)
	at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:1876)
	at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
	at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1876)
	at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:926)
	at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:926)
	at scala.Option.foreach(Option.scala:407)
	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2110)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2059)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2048)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
	at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2158)
	at org.apache.spark.rdd.RDD.$anonfun$reduce$1(RDD.scala:1035)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
	at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
	at org.apache.spark.rdd.RDD.reduce(RDD.scala:1017)
	at org.apache.spark.api.java.JavaRDDLike.reduce(JavaRDDLike.scala:385)
	at org.apache.spark.api.java.JavaRDDLike.reduce$(JavaRDDLike.scala:385)
	at org.apache.spark.api.java.AbstractJavaRDDLike.reduce(JavaRDDLike.scala:45)
	at com.intel.daal.samples.spark.SparkCovarianceCSR.reducePartialResults(SparkCovarianceCSR.java:85)
	at com.intel.daal.samples.spark.SparkCovarianceCSR.runCovariance(SparkCovarianceCSR.java:53)
	at com.intel.daal.samples.spark.SampleCovarianceCSR.main(SampleCovarianceCSR.java:58)
Caused by: java.io.InvalidClassException: com.intel.daal.data_management.data_source.StringDataSource; no valid constructor
	at java.base/java.io.ObjectStreamClass$ExceptionInfo.newInvalidClassException(ObjectStreamClass.java:159)
	at java.base/java.io.ObjectStreamClass.checkDeserialize(ObjectStreamClass.java:864)
	at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2061)
	at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1594)
	at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2355)

...

[Evegen55]

I've just tried to refactor - may be it will be helpful

[PivovarA]

Hello @Evegen55
Thanks for pointing this out. It looks like it can be removed. I don`t see that it is being used somewhere in this class.

[Evegen55]

Yep.
As we can see, the main idea of this samples is to show the main concept of distributed computation by performing job in three steps

• make a serialized entity from data presents on local node by using DAAL JavaAPI
• reduce partial results by using Spark mechanism
• compute local result on master node by using DAAL JavaAPI
  For each of this step it needs to use DAAL API directly on a node without neediness of serialization of DAAL wrappers for C++ classes

Maybe later I make a pool request with refactored code. Thank you!

[napetrov]

Java interfaces have been deprecated