Присоединение к СДР: После объединения двух разных пар СДР значение и порядок результирующего ключа СДР изменились?

У меня есть две пары RDD, скажем

 RDD1 : [(1,a),(2,b),(3,c)]    
 RDD2 : [(1,d),(2,e),(3,f)]

Теперь я присоединяюсь к этим RDD, используя join

 RDD3 = RDD1.join(RDD2);

И я отобразил элементы в RDD3 с кодом ниже

 for(Tuple2<Integer,Tuple2<String,String>> tuple : RDD3.collect()) 
                      System.out.println(tuple._1()+":"+tuple._2()._1()+","+tuple._2()._2());

я видел странные результаты, такие как

5:b,e
4:a,d 
6:c,f

где, как я и ожидал

1:a,d
1:b,e 
1:c,f

Есть ли способ получить желаемый результат, как указано выше? или же я неправильно интерпретирую поведение RDD? Пожалуйста, предложите

Редактировать :

На самом деле я читаю такие данные

JavaDoubleRDD data1 = sc.parallelizeDoubles(Arrays.asList(45.25,22.15,33.24));
JavaDoubleRDD data2 = sc.parallelizeDoubles(Arrays.asList(23.45,19.35,12.45));

а потом

JavaPairRDD<Double,Double> lat1 = data1.cartesian(data1);
JavaRDD<Double> lat2 = lat1.map(new Function<Tuple2<Double,Double>,Double>() {
    @Override
    public Double call(Tuple2<Double,Double> t) {
        return Math.pow(t._1()-t._2(),2);
    }
});
 //flag and flag1 are static variables initially equal to 1
JavaPairRDD<Integer,Double> lat3 = lat2.mapToPair(new PairFunction<Double,Integer,Double>() {
    @Override
     public Tuple2<Integer,Double> call(Double d) {
        return new Tuple2<Integer,Double>(flag++,d);
    }
});
System.out.println("Latitude values display");  
    for(Tuple2<?,?> tuple : lat3.collect()) {
                  System.out.println(tuple._1()+":"+tuple._2());
    } 
JavaPairRDD<Double,Double> long1 = data2.cartesian(data2);
JavaRDD<Double> long2 = long1.map(new Function<Tuple2<Double,Double>,Double>() {
        @Override
        public Double call(Tuple2<Double,Double> t) {
                return Math.pow(t._1()-t._2(),2);
    }
});
    JavaPairRDD<Integer,Double> long3 = long2.mapToPair(new PairFunction<Double,Integer,Double>() {
        @Override
        public Tuple2<Integer,Double> call(Double d ) {
                return new Tuple2<Integer,Double>(flag1++,d);
        }
});
System.out.println("Longitude values display"); 
    for(Tuple2<?,?> tuple : long3.collect()) {
                  System.out.println(tuple._1()+":"+tuple._2());
    }
System.out.println("latitude and longitude values join");
JavaPairRDD<Integer,Tuple2<Double,Double>> weightmatrix1 = lat3.join(long3);
System.out.println("Weightmatrix1 Display");
    for(Tuple2<?,Tuple2<?,?>> tuple : weightmatrix1.collect()) {
                  System.out.println(tuple._1()+":"+tuple._2()._1()+","+tuple._2()._2());
    }

Итак, что я делаю, так это вычисляю матрицу веса на основе значений широты и долготы.


java apache-spark join rdd
person swagath001    schedule 02.05.2015    source источник
comment
Недостаточно информации. Я подозреваю, что у вас проблема с кодом, который вы не показываете.   -  person Sean Owen    schedule 03.05.2015
comment
@Sean Owen: я добавил свой код. Мне нужно построить матрицу, похожую на данные расстояний на основе значений широты и долготы.   -  person swagath001    schedule 03.05.2015
comment
Я думаю, что проблема довольно очевидна из-за того, что вы используете глобальные статические переменные. Ваша постановка проблемы - это не то, что делает ваш код.   -  person Sean Owen    schedule 03.05.2015

Ответы (2)


arrow_upward
5
arrow_downward

Когда я делаю:

scala> val rdd1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
scala> val rdd2 = sc.parallelize(Array((1,"d"),(2,"e"),(3,"f")))
scala> val rdd3 = rdd1.join(rdd2)
scala> rdd3.toArray.foreach(println(_))

Я последовательно получаю:

(1,(a,d))
(2,(b,e))
(3,(c,f))
person David Griffin    schedule 02.05.2015

arrow_upward
0
arrow_downward

Это то, что я пробовал с ожидаемыми результатами:

val data1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
val data2 = sc.parallelize(Array((1,"d"),(2,"e"),(3,"f")))
val data3 = data1.join(data2)
data3.collect().map(tuple => tuple._1 + ":"+tuple._2._1+","+tuple._2._2).foreach(println(_))

Получение:

1:a,d
2:b,e
3:c,f

Итак, это скала. Я полагаю, в Java должен быть такой же вывод.

person Raúl Reguillo Carmona    schedule 23.09.2017