Khởi tạo một RDD để trống

Question

Tôi có một RDD gọi

JavaPairRDD<String, List<String>> existingRDD; 

Bây giờ tôi cần phải khởi tạo này existingRDD để trống để khi tôi nhận được của RDD thực tế tôi có thể làm một liên minh với existingRDD này. Làm cách nào để khởi tạo existingRDD cho RDD trống trừ khi khởi tạo nó thành null? Đây là mã của tôi:

JavaPairRDD<String, List<String>> existingRDD; 
if(ai.get()%10==0) 
{ 
    existingRDD.saveAsNewAPIHadoopFile("s3://manthan-impala-test/kinesis-dump/" + startTime + "/" + k + "/" + System.currentTimeMillis() + "/", 
    NullWritable.class, Text.class, TextOutputFormat.class); //on worker failure this will get overwritten         
} 
else 
{ 
    existingRDD.union(rdd); 
} 

Answer 1

Tôi vẫn không chắc chắn về những gì bạn đang cố gắng để làm, nhưng bạn có thể tạo ra một RDD rỗng như sau:

// Get an RDD that has no partitions or elements. 
JavaRDD<T> emptyRDD = sc.emptyRDD() 

tôi tin tưởng bạn biết làm thế nào để sử dụng Generics, nếu không, ở đây là:

JavaRDD<Tuple2<String,List<String>>> emptyRDD = sc.emptyRDD(); 
JavaPairRDD<String,List<String>> emptyPairRDD = JavaPairRDD.fromJavaRDD(
    existingRDD 
); 

bạn cũng có thể sử dụng phương pháp mapToPair để chuyển đổi của bạn JavaRDD đến một JavaPairRDD.

Các giải pháp trong scala:

scala> val emptyRDD = sc.emptyRDD 
// emptyRDD: org.apache.spark.rdd.EmptyRDD[Nothing] = EmptyRDD[1] at ... 

Answer 2

Trong scala, tôi đã sử dụng "parallelize" lệnh.

val emptyRDD = sc.parallelize(Seq("")) 

Answer 3

@eliasah câu trả lời rất hữu ích, tôi đang cung cấp mã để tạo RDD cặp trống. Hãy xem xét một kịch bản mà trong đó nó được yêu cầu để tạo ra RDD cặp trống (khóa, giá trị). Sau mã scala minh họa cách tạo RDD cặp trống với khóa là String và giá trị là Int.

type pairRDD = (String,Int) 
var resultRDD = sparkContext.emptyRDD[pairRDD] 

RDD sẽ được tạo ra như sau:

resultRDD: org.apache.spark.rdd.EmptyRDD[(String, Int)] = EmptyRDD[0] at emptyRDD at <console>:29 

Answer 4

Trong Java, tạo RDD trống là một chút phức tạp. Tôi đã thử sử dụng scala.reflect.classTag nhưng nó không hoạt động. Sau nhiều lần kiểm tra, mã hoạt động thậm chí còn đơn giản hơn nhiều.

private JavaRDD<Foo> getEmptyJavaRdd() { 

/* this code does not compile because require <T> as parameter into emptyRDD */ 
//  JavaRDD<Foo> emptyRDD = sparkContext.emptyRDD(); 
//  return emptyRDD; 

/* this should be the solution that try to emulate the scala <T> */ 
/* but i could not make it work too */ 
//  ClassTag<Foo> tag = scala.reflect.ClassTag$.MODULE$.apply(Foo.class); 
//  return sparkContext.emptyRDD(tag); 

/* this alternative worked into java 8 */ 
    return SparkContext.parallelize(
      java.util.Arrays.asList() 
    ); 

}