Jak umożliwić iskrowi ignorowanie brakujących plików wejściowych?

Question

Chcę uruchomić zadanie iskrzenia (iskra v1.5.1) na niektórych wygenerowanych ścieżkach S3 zawierających pliki avro. Ładuję je:

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

Niektóre ścieżki nie istnieją. Jak mogę uzyskać iskrę, aby zignorować te puste ścieżki? Poprzednio używałem this answer, ale nie jestem pewien, jak tego użyć z nowym interfejsem API danych.

Uwaga: Idealnie szukam podobnego podejścia do połączonej odpowiedzi, która sprawia, że ​​ścieżki wejściowe są opcjonalne. Nie chcę szczególnie wyraźnie sprawdzać istnienia ścieżek w S3 (ponieważ jest to uciążliwe i może utrudniać programowanie), ale myślę, że to jest mój powrót, jeśli nie ma żadnego prostego sposobu na zaimplementowanie tego.

Answer 1

Chciałbym użyć typu scala Try w celu obsługi możliwości awarii podczas odczytu katalogu plików avro. Dzięki "Try" możemy w naszym kodzie jawnie ujawnić możliwość wystąpienia błędu i obsłużyć go w sposób funkcjonalny:

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() 
}