java - Inicializar un RDD para vaciar

apache-spark

Question

Tengo un RDD llamado

JavaPairRDD<String, List<String>> existingRDD; 

Ahora necesito inicializar este DRDD existingRDD para que esté vacío, de modo que cuando obtenga los rdd reales pueda hacer una unión con este DRDD existingRDD . ¿Cómo inicializo RDD existingRDD en un RDD vacío, excepto en inicializarlo en nulo? Aquí está mi código:

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

En Java, crear el RDD vacío era un poco complejo. Intenté usar el scala.reflect.classTag pero tampoco funciona. Después de muchas pruebas, el código que funcionó fue aún más simple.

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 2

En Java, cree un par de RDD vacío de la siguiente manera:

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

Answer 3

En Scala, usé el comando "paralelizar".

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

Answer 4

La respuesta de @eliasah es muy útil, estoy proporcionando código para crear un RDD de par vacío. Considere un escenario en el que se requiere crear un RDD de par vacío (clave, valor). El siguiente código de Scala ilustra cómo crear un RDD de par vacío con clave como Cadena y valor como Int.

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

RDD se crearía de la siguiente manera:

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

Answer 5

Para crear un RDD vacío en Java , solo debes hacer lo siguiente:

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

Confío en que sepa cómo usar los genéricos, de lo contrario, para su caso, necesitará:

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

También puede usar el método mapToPair para convertir su JavaRDD a un JavaPairRDD .

En Scala :

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

Answer 6

val emptyRdd=sc.emptyRDD[String]

La declaración anterior creará un RDD vacío con String tipo de String

De la clase SparkContext:

Obtener un RDD que no tiene particiones o elementos

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