스파크 예외지도에서 필터를()()를 사용할 때

Question

내가지도 내부 필터를()()를 사용하려고하지만 난이 스파크 예외가 :

RDD 변환과 행동 드라이버 만 호출 할 수

다른 변형의 내부가 아닙니다. 예를 들어 rdd1.map (x => rdd2.values.count() * x)은 rdd1.map 변환에서 값 변환 및 개수 작업을 수행 할 수 없기 때문에 유효하지 않습니다. 자세한 내용은 SPARK-5063을 참조하십시오.

아니라 내가 그 튜플은 RDD이다, 나는 그 불꽃이 중첩 변환/액션/RDDs을 허용하지 않습니다 알고, 그래서 어떤 사람은 나에게 (중첩 변환이나 행동하지 않고) 다른 방법을 수행하는 방법 제안을 제공 할 수 있습니다 같은 :

내가 인수로 그것을 목록을주고, 그것을 매핑 할

JavaRDD< String[]> RDD 

이이 목록은 javaPairRDDs 포함

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

이 라인은 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) ; 

    } 


    } 

때문에, 문제가

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

지금 내가 예를 들어 줄이 라인에, 그 목록은 내가 포함 매핑 할 2 PairRDDs

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

내 RDD 포함되어있는 경우를 가정 해 봅시다 :

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

내가 원하는 결과는 map()입니다.

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

Answer 1

내가 지금은 (BCZ offcourse 내가 더 중첩 된 변형이없는)도 예외가 없다)지도 (내부 RDDs 다루는 피하기 위해 >> 목록을 나열하려면> 목록에서 목록의 유형을 변경,하지만 나는 그것이 새로운 코드가 효율적인지, bcz List>가 크고, 엘리먼트를 검색하기 위해 "for"루프를 사용했다는 것을 확신하지 못한다. (내가 전체 엘리먼트를 스윕해야 함을 의미한다.) 그래서 나는 (루프를 사용하여) 그것에 관해서 나에게 비평을하고, 그것을 개량하기위한 제안을하기 위해 전문가로서 당신에게 묻는다. 당신이 수정 후지도() 함수

감사

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); 
            }