Jak korzystać z RangePartitioner w Spark

Question

Chcę używać RangePartitioner w mojej aplikacji Java Spark, ale nie mam pojęcia, jak ustawić dwa parametry scala: scala.math.Ordering<K> evidence$1 i scala.reflect.ClassTag<K> evidence$2. Czy ktoś może dać mi przykład?

Oto link do JavaDoc of RangePartitioner (było żadnej pomocy dla mnie, bo jestem nowy Spark i Scala ...):

mój kod faktycznie wygląda:

JavaPairRDD<Integer, String> partitionedRDD = rdd.partitionBy(new RangePartitioner<Integer, String>(10, rdd, true, evidence$1, evidence$2)); 

Answer 1

Jeśli patrzysz na api dla OrderedRDDFunctions jest przykład, jak ustawić domyślne porządkowanie swojego klucza.

import org.apache.spark.SparkContext._ 

val rdd: RDD[(String, Int)] = ... 
implicit val caseInsensitiveOrdering = new Ordering[String] { 
    override def compare(a: String, b: String) = a.toLowerCase.compare(b.toLowerCase) 
} 

znam jej fragment od apis zapłonie scala, ale można przynajmniej wywnioskować, jak przekazać swój parametr zamówieniu. W przypadku typu ClassTag sugeruję sprawdzenie ogólnego dokumentu lub forum. Dodanie znacznika scala do pytania.

Answer 2

Można tworzyć zarówno Ordering i ClassTag, wywołując metody na swoich obiektach towarzyszących.

Są one określane w java tak: ClassName$.MODULE$.functionName()

Jedno dodatkowe zmarszczka jest to, że konstruktor wymaga Scala RDD, a nie jeden java. Można uzyskać RDD scala z java PairRDD wywołując rdd.rdd()

final Ordering<Integer> ordering = Ordering$.MODULE$.comparatorToOrdering(Comparator.<Integer>naturalOrder()); 
    final ClassTag<Integer> classTag = ClassTag$.MODULE$.apply(Integer.class); 
    final RangePartitioner<Integer, String> partitioner = new RangePartitioner<>(
      10, 
      rdd.rdd(), //note the call to rdd.rdd() here, this gets the scala RDD backing the java one 
      true, 
      ordering, 
      classTag); 
    final JavaPairRDD<Integer, String> partitioned = rdd.partitionBy(partitioner);