exception d'étincelle lors de l'utilisation de filter() dans map()

Question

J'essaie d'utiliser filter() à l'intérieur de map() mais j'obtiens cette exception :

Les transformations et actions RDD ne peuvent être invoquées que par le pilote, et non à l'intérieur d'autres transformations ; par exemple, rdd1.map(x => rdd2.values.count() * x) n'est pas valide car la transformation des valeurs et l'action de comptage ne peuvent pas être effectuées à l'intérieur de la transformation rdd1.map. Pour plus d'informations, voir SPARK-5063.

Je sais que spark ne permet pas les transformations/actions/RDD imbriquées, donc quelqu'un peut-il me donner une suggestion sur la façon de le faire alternativement (sans transformations ou actions imbriquées), eh bien j'ai un RDD dont les tuples sont comme :

 JavaRDD< String[]> RDD

J'essaie de le mapper, en lui donnant une liste comme argument, cette liste contient des javaPairRDD tels que :

List<JavaPairRDD<String,String>> list
JavaRDD< String[]> result = RDD.map(new modifyRDD(list));

Ces lignes font référence à la fonction modifyRDD() :

public static class modifyRDD implements Function <String[], String[]> { 

    List<JavaPairRDD<String,String>> list;
    public modifyRDD (List<JavaPairRDD<String,String>> list ){ this.list=list;}

    public String [] call(String[] t) {

          String[] s = t;

          for (int i = 0; i < NB_TD; i++) {         
            int j=i; 
         // select the appropriate RDD from the RDDs_list to the current index 

            JavaPairRDD<String,String> rdd_i = list.get(i);
            String previousElement=s[j];

           JavaPairRDD<String,String> currentRDD =  rdd_i.filter(line -> line._1().equals(previousElement));

           String newElement=currentRDD.first()._2();   

           s[j]=newElement;
                }

          return   (s) ;

    }

    }

Le problème se situe donc au niveau de cette ligne

  JavaPairRDD<String,String> currentRDD =  rdd_i.filter(line -> line._1().equals(previousElement));

Je vais maintenant donner un exemple, supposons que la liste contienne 2 PairRDD.

list={PairRDD1={(a,b)(c,d)},PairRDD2={(u,v)(x,y)}..}

et mon RDD que je veux mapper contient :

 JavaRDD< String[]> RDD = {[a,u],[c,x],[a,x].....}

Le résultat que je souhaite obtenir se trouve après map() :

 JavaRDD< String[]> result = {[b,v],[d,y],[b,y].....}

Answer 1

J'ai changé le type de la liste de List> à List>> list pour éviter de traiter des RDD à l'intérieur de map(), maintenant je n'ai pas d'exception (bien sûr parce que je n'ai pas de transformations imbriquées), mais ce dont je ne suis pas sûr, c'est que mon nouveau code est efficace, En effet, la List> est grande, et pour rechercher un élément, j'ai utilisé une boucle "for" (ce qui signifie que je dois balayer toute la List> pour obtenir l'élément que je veux). Je vous demande donc, en tant qu'experts, de me faire part de vos remarques à ce sujet (utilisation de la boucle for), et de me donner des suggestions pour l'améliorer. Je vous remercie.

voici la fonction map() après modification

  public static class modifyRDD implements Function <String[], String[]> { 

    List<List<Tuple2<String,String>>> list;
    public modifyRDD (List<List<Tuple2<String,String>>> list ){ this.list=list;}

    public String [] call(String[] t) {

          String[] s = t;

          for (int i = 0; i < NB_TD; i++) {         

         // select the appropriate lookup_list 

            List<Tuple2<String,String>> list_i = list.get(i);
            String previousElement=s[i];
            String newElement="";

            for (int k = 0; k < list_i.size(); k++){

            Tuple2<String,String> sk1 = list_i.get(k);
            if (sk1._1.equals( previousElement)){  newElement=sk1._2;}

            }

           s[i]= newElement;
                }
         return(s);
                                   }