スパークClassCastException：JavaRDDをorg.apache.spark.mllib.linalg.Vectorにキャストできません

Question

Javaを使用してARIMA時系列を実装したいと思います。 私は、Scalaのコードを次ています

object SingleSeriesARIMA { 
    def main(args: Array[String]): Unit = { 
    // The dataset is sampled from an ARIMA(1, 0, 1) model generated in R. 
    val lines = scala.io.Source.fromFile("../data/R_ARIMA_DataSet1.csv").getLines() 
    val ts = Vectors.dense(lines.map(_.toDouble).toArray) 
    val arimaModel = ARIMA.fitModel(1, 0, 1, ts) 
    println("coefficients: " + arimaModel.coefficients.mkString(",")) 
    val forecast = arimaModel.forecast(ts, 20) 
    println("forecast of next 20 observations: " + forecast.toArray.mkString(",")) 
    } 
} 

を、私は以下のソリューション試してみました：

public class JavaARIMA { 

public static void main(String args[]) 
     { 
    System.setProperty("hadoop.home.dir", "C:/winutils"); 
    SparkConf conf = new SparkConf().setAppName("Spark-TS Ticker Example").setMaster("local").set("spark.sql.warehouse.dir", "file:///C:/Users/devanshi/Downloads/Spark/sparkdemo/spark-warehouse/"); 
    JavaSparkContext context = new JavaSparkContext(conf); 

    JavaRDD<String> lines = context.textFile("path/inputfile"); 

    JavaRDD<Vector> ts = lines.map(
       new Function<String, Vector>() { 
       public Vector call(String s) { 
        String[] sarray = s.split(","); 
        double[] values = new double[sarray.length]; 
        for (int i = 0; i < sarray.length; i++) { 
        values[i] = Double.parseDouble(sarray[i]); 
        } 
        return Vectors.dense(values); 
       } 
       } 
      ); 
    double[] total = {1.0,0.0,1.0}; 
    //DenseVector dv = new DenseVector(total); 
    //convert(dv,toBreeze()); 
    //ARIMAModel arimaModel = ARIMA.fitModel(1, 0, 1, dv, true, "css-cgd", null); 
    ARIMAModel arimaModel = ARIMA.fitModel(1, 0, 1, (Vector) ts, false, "css-cgd", total); 

    // arimaModel = ARIMA.fitModel(1, 0, 1, ts); 
    System.out.println("coefficients: " + arimaModel.coefficients()); 
    Vector forcst = arimaModel.forecast((Vector) ts,20); 
    System.out.println("forecast of next 20 observations: " + forcst); 
} 
} 

をしかし、私が得た：

Exception in thread "main" java.lang.ClassCastException: 
org.apache.spark.api.java.JavaRDD cannot be cast to 
org.apache.spark.mllib.linalg.Vector 

可能な場合は、私を助けてください。

Answer 1

JavaRDDをVectorにキャストすることはできません。代わりに、rdd.foreachを使用して個々のVectorを取得する必要があります。だから、コードはこのようになる可能性があります。

ts.foreach(new VoidFunction<Vector>() { 
    @Override 
    public void call(Vector v) throws Exception { 
     double[] total = { 1.0, 0.0, 1.0 }; 
     ARIMAModel arimaModel = ARIMA.fitModel(1, 0, 1, (Vector) v, false, "css-cgd", total); 

     System.out.println("coefficients: " + arimaModel.coefficients()); 
     Vector forcst = arimaModel.forecast((Vector) v, 20); 
     System.out.println("forecast of next 20 observations: " + forcst); 
    } 
});