将RDD初始化为空

Question

我叫RDD

JavaPairRDD<String, List<String>> existingRDD; 

现在我需要将这个existingRDD初始化为空，这样当我得到实际的rdd时，我可以与这个existingRDD建立联盟。如何将existingRDD初始化为空RDD，除非将其初始化为null？这是我的代码：

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

要在Java中创建一个空的RDD，您只需执行以下操作：

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

我相信你知道如何使用泛型，否则，对于你的情况，你需要：

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

您还可以使用mapToPair方法将JavaRDD转换为JavaPairRDD。

在斯卡拉：

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

Answer 2

val emptyRdd=sc.emptyRDD[String]

上面的语句将使用String Type创建空的RDD

来自SparkContext类：

获取没有分区或元素的RDD

def emptyRDD[T: ClassTag]: EmptyRDD[T] = new EmptyRDD[T] (this)

Answer 3

在scala中，我使用了“parallelize”命令。

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

Answer 4

@eliasah回答非常有用，我提供代码来创建空对RDD。考虑一种需要创建空对RDD（键，值）的场景。以下scala代码说明了如何创建空对RDD，其中键为String，值为Int。

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

RDD将创建如下：

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

Answer 5

在Java中，创建空RDD有点复杂。我尝试使用scala.reflect.classTag，但它也不起作用。经过多次测试后，工作的代码更加简单。

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

}

Answer 6

在Java中，创建空对RDD，如下所示：

JavaPairRDD<T, T> emptyPairRDD = JavaPairRDD.fromJavaRDD(SparkContext.emptyRDD());