RDDの2つの異なるタイプをマージする

Question

RDD [Vector]とRDD [Int]をRDD [Vector]に組み合わせたいのですが これは私が何をしたのか、私はクラスタを予測するためにKmeansを使用しました。前の各ベクトルiの結果は、私が

[0.0,0.0,0.0,1.0] 
    [0.1,0.1,0.1,1.0] 
    [0.2,0.2,0.2,1.0] 
    [9.0,9.0,9.0,0.0] 
    [9.1,9.1,9.1,0.0] 
    [9.2,9.2,9.2,0.0] 

をしたいと思います

([0.0,0.0,0.0],0) 
([0.1,0.1,0.1],0) 
([0.2,0.2,0.2],0) 
([9.0,9.0,9.0],1) 
([9.1,9.1,9.1],1) 
([9.2,9.2,9.2],1) 

出力され

val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate() 
val data = spark.sparkContext.textFile("C:/spark/data/mllib/kmeans_data.txt") 
//Cluster the data into two classes using KMeans 
val numClusters = 2 
val numIterations = 20 
val parsedData = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble))).cache()//RDD[vector] 
val clusters = KMeans.train(parsedData, numClusters, numIterations) 
val resultatOfprediction=clusters.predict(parsedData)//RDD[int] 
val finalData=parsedData.zip(resultatOfprediction) 
finalData.collect().foreach(println) 

をやったcorrespondantクラスタ.Here

最終的なRDD [ベクトル]をtxtファイルに保存してグリッドに表示したいのですが、あなたが提供する結果はRDDではありません[ベクトル]

Answer 1

必要な結果を得るにはこれら2つのRDDを圧縮する。ここでは、あなたがそれを行う方法である

val parsedData = spark.sparkContext.parallelize(Seq(1.0,1.0,1.0,0.0,0.0,0.0)) 

val resultatOfprediction = spark.sparkContext.parallelize(Seq(
    (0.0,0.0,0.0), 
    (0.1,0.1,0.1), 
    (0.2,0.2,0.2), 
    (9.0,9.0,9.0), 
    (9.1,9.1,9.1), 
    (9.2,9.2,9.2) 
)) 

resultatOfprediction.zip(parsedData) 

それがタプルを返すので、あなたがなり得ることができるよう、あなたが@ラーフル・Sukla resultatOfprediction.zip(parsedData) .map(t => t._1.productIterator.toList.map(_.asInstanceOf[Double]) :+ t._2)

希望で推薦してfollingとして行うことができますダイナミックについて

resultatOfprediction.zip(parsedData) 
     .map(t => (t._1._1, t._1._2, t._1._3, t._2)) 

これは役に立ちます！