Spark (Scala) Problem Iteratoren

Question

Ich erstelle eine Funktion für Mappartitions, um das Maximum und Minimum pro Partition zu berechnen. Ich habe die Funktion in pyspark erstellt, aber ich kann sie nicht erfolgreich in scala übersetzen. Ich wende diese Funktion zweimal an und möchte im Ergebnis eine Zip-Funktion ausführen. Dies ist die errot, die ich bekomme:

result.zip (res)

Typenkonflikt;

[error] found : org.apache.spark.rdd.RDD[(Int, Int)] 
[error] required: scala.collection.GenIterable[?] 

Hier können Sie meine Funktion in Python haben:

def minmaxInt(iterator): 
    firsttime = 0 
    min = 0 
    max = 0 
    for x in iterator: 
    if(x!= '' and x!='NULL' and x is not None): 
     y=int(x)  
      if (firsttime == 0): 
        min = y; 
        max = y; 
        firsttime = 1 
      else: 
        if y > max: 
         max = y 
        if y < min: 
         min = y 
    return (min, max) 

Und hier mein Code in Scala

def minmaxInt(iterator: Iterator[String]) : Iterator[(Int,Int)]={ 

    var firsttime = 0 
    var min = 0 
    var max = 0 
    var res=List[(Int,Int)]() 
    for(x <- iterator){ 
     if(x!= "" && x!= null){ 
    var y=x.toInt 

     if(firsttime == 0){ 
      min = y 
      max = y 
      firsttime = 1} 
     else{ 
      if (y > max){ 
       max = y} 
      if (y < min){ 
       min = y} 
     } 
     } 
    } 

    res.::=(min,max) 
    return res.iterator 

} 

Vielen Dank im Voraus

Update:

Dank für dein schnelle Antwort! Der Code ist großartig, aber ich habe immer noch Probleme mit dem Zip. Ich habe zweimal die letzten Code für rdd.mapPartitions und dann ausführen ich die Zip:

[error] found : org.apache.spark.rdd.RDD[(Int, Int)] 
[error] required: scala.collection.GenIterable[?] 
[error]    result.zip(res) 

Answer 1

Hier ist eine einfacher (und mehr idiomatische) Umsetzung von minMaxInt:

def minMaxInt(iterator: Iterator[String]) : Iterator[(Int,Int)]= { 
    val tuple = iterator 
    .filter(_ != null).filter(!_.isEmpty) 
    .map(_.toInt).map(i => (i, i)) 
    .reduce[(Int, Int)] { case ((min, max), (i1, i2)) => (Math.min(min, i1), Math.max(max, i2)) } 

    Seq(tuple).iterator 
} 

, die zu einem RDD[String] angewandt werden können, wie folgt:

// some sample data 
def col = sc.parallelize(Seq("1", "4", "12", "3", "", null, "2")) 

// "use twice" and zip 
var result: RDD[(Int, Int)] = col.mapPartitions(minmaxInt) 
var res: RDD[(Int, Int)] = col.mapPartitions(minmaxInt) 

result.zip(res).foreach(println) 
// prints: 
// ((1,1),(1,1)) 
// ((2,2),(2,2)) 
// ((3,3),(3,3)) 
// ((4,12),(4,12))