Wie man Funken erlaubt, um fehlende Eingabedateien zu ignorieren?

Question

Ich möchte einen Funken-Job (Funken v1.5.1) über einige generierte S3-Pfade mit Avro-Dateien ausführen. Ich lade sie mit:

val avros = paths.map(p => sqlContext.read.avro(p)) 

Einige der Pfade werden jedoch nicht existieren. Wie kann ich Funken sprühen, um diese leeren Pfade zu ignorieren? Zuvor habe ich this answer verwendet, aber ich bin mir nicht sicher, wie man das mit der neuen Dataframe-API benutzt.

Hinweis: Ich suche idealerweise nach einem ähnlichen Ansatz für die verknüpfte Antwort, die Eingabepfade nur optional macht. Ich möchte nicht explizit auf die Existenz von Pfaden in S3 prüfen müssen (da das mühsam ist und die Entwicklung schwierig machen könnte), aber ich denke, das ist mein Fallback, wenn es keinen sauberen Weg gibt, dies jetzt zu implementieren.

Answer 1

Ich würde die Scala Try Typ verwenden, um die Möglichkeit des Fehlers beim Lesen eines Verzeichnisses von AVRO-Dateien zu behandeln. Mit ‚Try‘ können wir die Möglichkeit des Scheiterns explizit in unserem Code machen, und es in funktioneller Weise handhaben:

object Main extends App { 

    import scala.util.{Success, Try} 
    import org.apache.spark.{SparkConf, SparkContext} 
    import com.databricks.spark.avro._ 

    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("example")) 
    val sqlContext = new org.apache.spark.sql.SQLContext(sc) 

    //the first path exists, the second one doesn't 
    val paths = List("/data/1", "/data/2") 

    //Wrap the attempt to read the paths in a Try, then use collect to filter 
    //and map with a single partial function. 
    val avros = 
    paths 
     .map(p => Try(sqlContext.read.avro(p))) 
     .collect{ 
     case Success(df) => df 
     } 
    //Do whatever you want with your list of dataframes 
    avros.foreach{ df => 
    println(df.collect()) 
    } 
    sc.stop() 
}