Comment obtenir de l'élément à l'Index dans l'Étincelle RDD (Java)

Je sais que la méthode de rdd.(première) qui me donne le premier élément dans un EDR.

Il y a aussi la méthode de rdd.prendre(num) Qui me donne le premier "num" éléments.

Mais n'est-il pas une possibilité pour obtenir un élément à l'index?

Grâce.

InformationsquelleAutor progNewbie | 2014-11-09
  1. 57

    Cela devrait être possible, par la première indexation de la RDD. La transformation zipWithIndex fournit une stabilité de l'indexation, la numérotation de chaque élément dans l'ordre original.

    Donné: rdd = (a,b,c)

    val withIndex = rdd.zipWithIndex //((a,0),(b,1),(c,2))

    À la recherche d'un élément à l'index, ce formulaire n'est pas utile. Nous avons d'abord besoin d'utiliser l'index de clé:

    val indexKey = withIndex.map{case (k,v) => (v,k)}  //((0,a),(1,b),(2,c))

    Maintenant, il est possible d'utiliser le lookup action dans PairRDD de trouver un élément clé:

    val b = indexKey.lookup(1) //Array(b)

    Si vous vous attendez à utiliser lookup souvent sur le même CA, je vous recommande pour mettre en cache les indexKey RDD pour améliorer les performances.

    Comment faire cela en utilisant la API Java est un exercice laissé au lecteur.

    • Rappelez-vous aussi que le CA de la commande n'est pas défini, sauf si elle est triée. Il peut changer à chaque fois c'est évaluée autrement. Vous pouvez également map(_.swap) pour passer des clés et des valeurs.
    • Ce que vous de la gauche "comme un exercice pour le lecteur" est la principale complexité de cette question.
    • Quelles seraient, selon vous manque? zipWithIndex et lookup sont les mêmes en Java, de sorte que seules les map fonction nécessite un petit effort pour permuter les valeurs de la (clé, valeur) dans (valeur, clé) ce n'est pas de base pour cette question.
    • Vieux thread, mais j'ai ajouté un Java de réponse (bien que la recherche d'un éventail de dossiers pas un enregistrement individuel).
    • Vous voulez vraiment indexKey être partitionné.
    InformationsquelleAutor maasg
  2. 2

    J'ai essayé cette classe pour aller chercher un élément par son indice. Tout d'abord, lorsque vous construisez new IndexedFetcher(rdd, itemClass), il compte le nombre d'éléments dans chaque partition de la RDD. Ensuite, lorsque vous appelez indexedFetcher.get(n), il exécute un travail uniquement sur la partition qui contient l'indice.

    Remarque que j'avais besoin de le compiler à l'aide de Java 1.7 au lieu de 1.8; comme d'Étincelle 1.1.0, le logiciel org.objectweb.asm dans com.esotericsoftware.reflectasm ne peut pas lire Java 1.8 classes encore (throws exception IllegalStateException lorsque vous essayez de runJob Java 1.8 fonction).

    import java.io.Serializable;
import org.apache.spark.SparkContext;
import org.apache.spark.TaskContext;
import org.apache.spark.rdd.RDD;
import scala.reflect.ClassTag;
public static class IndexedFetcher<E> implements Serializable {
private static final long serialVersionUID = 1L;
public final RDD<E> rdd;
public Integer[] elementsPerPartitions;
private Class<?> clazz;
public IndexedFetcher(RDD<E> rdd, Class<?> clazz){
this.rdd = rdd;
this.clazz = clazz;
SparkContext context = this.rdd.context();
ClassTag<Integer> intClassTag = scala.reflect.ClassTag$.MODULE$.<Integer>apply(Integer.class);
elementsPerPartitions = (Integer[]) context.<E, Integer>runJob(rdd, IndexedFetcher.<E>countFunction(), intClassTag);
}
public static class IteratorCountFunction<E> extends scala.runtime.AbstractFunction2<TaskContext, scala.collection.Iterator<E>, Integer> implements Serializable {
private static final long serialVersionUID = 1L;
@Override public Integer apply(TaskContext taskContext, scala.collection.Iterator<E> iterator) {
int count = 0;
while (iterator.hasNext()) {
count++;
iterator.next();
}
return count;
}
}
static <E> scala.Function2<TaskContext, scala.collection.Iterator<E>, Integer> countFunction() {
scala.Function2<TaskContext, scala.collection.Iterator<E>, Integer> function = new IteratorCountFunction<E>();
return function;
}
public E get(long index) {
long remaining = index;
long totalCount = 0;
for (int partition = 0; partition < elementsPerPartitions.length; partition++) {
if (remaining < elementsPerPartitions[partition]) {
return getWithinPartition(partition, remaining);
}
remaining -= elementsPerPartitions[partition];
totalCount += elementsPerPartitions[partition];
}
throw new IllegalArgumentException(String.format("Get %d within RDD that has only %d elements", index, totalCount));
}
public static class FetchWithinPartitionFunction<E> extends scala.runtime.AbstractFunction2<TaskContext, scala.collection.Iterator<E>, E> implements Serializable {
private static final long serialVersionUID = 1L;
private final long indexWithinPartition;
public FetchWithinPartitionFunction(long indexWithinPartition) {
this.indexWithinPartition = indexWithinPartition;
}
@Override public E apply(TaskContext taskContext, scala.collection.Iterator<E> iterator) {
int count = 0;
while (iterator.hasNext()) {
E element = iterator.next();
if (count == indexWithinPartition)
return element;
count++;
}
throw new IllegalArgumentException(String.format("Fetch %d within partition that has only %d elements", indexWithinPartition, count));
}
}
public E getWithinPartition(int partition, long indexWithinPartition) {
System.out.format("getWithinPartition(%d, %d)%n", partition, indexWithinPartition);
SparkContext context = rdd.context();
scala.Function2<TaskContext, scala.collection.Iterator<E>, E> function = new FetchWithinPartitionFunction<E>(indexWithinPartition);
scala.collection.Seq<Object> partitions = new scala.collection.mutable.WrappedArray.ofInt(new int[] {partition});
ClassTag<E> classTag = scala.reflect.ClassTag$.MODULE$.<E>apply(this.clazz);
E[] result = (E[]) context.<E, E>runJob(rdd, function, partitions, true, classTag);
return result[0];
}
}
    InformationsquelleAutor yonran
  3. 2

    Je suis coincé sur ce pendant un certain temps, donc de s'étendre sur les Maasg réponse mais répondant à regarder pour une plage de valeurs de l'indice pour Java (vous aurez besoin de définir les 4 variables en haut):

    DataFrame df;
SQLContext sqlContext;
Long start;
Long end;
JavaPairRDD<Row, Long> indexedRDD = df.toJavaRDD().zipWithIndex();
JavaRDD filteredRDD = indexedRDD.filter((Tuple2<Row,Long> v1) -> v1._2 >= start && v1._2 < end);
DataFrame filteredDataFrame = sqlContext.createDataFrame(filteredRDD, df.schema());

    Rappelez-vous que lorsque vous exécutez ce code de votre cluster de Java 8 (comme une expression lambda est en cours d'utilisation).

    Aussi, zipWithIndex est probablement cher!

    • salut, pouvez-vous s'il vous plaît guide solution similaire pour Java 7 ?
    • code rdd.filtre(la nouvelle Fonction<Tuple2<Ligne, Longue>, Boolean>(){ public Boolean appel(Tuple2<Ligne, Longue> v1){ return v1._2 >= start && v1._2 < end; }); code
    InformationsquelleAutor Luke W