Konvertieren Klassenobjekt in Java RDD

Question

Ich habe eine Pojo-AnalyticsModel und ich habe eine Schleife, wo ich ein Objekt für den Pojo erstellen. Jedes Mal, wenn ich das Objekt erstelle, speichere ich in einer Liste namens AnalyticsModelList. Und nachdem ich die Schleife verlassen habe, konvertiere ich die Liste in eine RDD. Unten ist mein Code:

do {  
     AnalyticsModel analyticsObj = new AnalyticsModel(time, columnName, aggrResult); 
     analyticsList.add(analyticsObj); 
    }while (cal.getTimeInMillis() <= endDate); 

    JavaRDD<AnalyticsModel> analyticsRdd = sc.parallelize(analyticsList); 

Statt zu einer Liste hinzuzufügen und dann zu einem RDD Umwandlung, ist es trotzdem, ich direkt das Objekt als RDD haben kann. Kann mir bitte jemand dabei helfen? Ich bin ziemlich neu in Spark.

Danke

Answer 1

Es gibt einen Weg. Aber was Sie tun, funktioniert gut, wenn Sie wenige AnalyticsModel Objekte haben (Sie müssen nicht Ihr Leben durch die folgende Lösung komplex zu machen). Wenn Sie eine große Anzahl von AnalyticsModel Objekte erstellen möchten, dann betrachten Sie unten Lösung.

Diese Klasse generiert eine AnalyticsModel Instanz pro Aufruf.

public class AnalyticsModelGenerator implements FlatMapFunction<String, AnalyticsModel> { 

    private final Long endDate; 
    private final Calendar cal; 

    public AnalyticsModelGenerator(Calendar cal, Long endDate) { 
     this.cal = cal; 
     this.endDate = endDate; 
    } 
    @Override 
    public Iterable<AnalyticsModel> call(String dummyInput) throws Exception { 
     return new Iterable<AnalyticsModel>() { 
      @Override 
      public Iterator<AnalyticsModel> iterator() { 
       return new AMIterator(cal, endDate); 
      } 
     } 
    } 

    private class AMIterator implements Iterator<AnalyticsModel> { 
     private final Long endDate; 
     private final Calendar cal; 

     public AMIterator(Calendar cal, Long endDate) { 
      this.cal = cal; 
      this.endDate = endDate; 
     } 

     @Override 
     public boolean hasNext() { 
      return (cal.getTimeInMillis() <= endDate); 
     } 

     @Override 
     public AnalyticsModel next() { 
      return new AnalyticsModel(time, columnName, aggrResult); 
     } 
    } 
} 

Im Folgenden Code erstellt eine RDD von AnalyticsModel Objekte:

JavaRDD<String> initJavaRDD = jSparkContext.parallelize(new ArrayList<String>(){{add("dummyInput");}}); 
JavaRDD<AnalyticsModel> amJavaRDD = initJavaRDD.flatMap(new AnalyticsModelGenerator(cal, endDate));