スパーク：パスが存在する場合のみファイルを読み込みます。

Question

スカラーのパスのSequenceにあるファイルを読み込もうとしています。以下はサンプル（疑似）コードです：

ここで、上記のシーケンスでは、いくつかのパスが存在するのに対し、いくつかのパスは存在しません。 parquetファイルを読み取っている間に、欠落しているパスを無視する方法はありますか（org.apache.spark.sql.AnalysisException: Path does not existを避けるため）？

私は以下試してみました、それが動作しているようだが、その後、私は私がやって避けたいものです二度同じパスを読み終わる：

val filteredPaths = paths.filter(p => Try(spark.read.parquet(p)).isSuccess) 

私はoptions方法を確認DataFrameReaderでも、それにはignore_if_missingに似たオプションはありません。

また、これらのパスはhdfsまたはs3ことができます（このSeqは、メソッドの引数として渡されます）と読んでいる間、私はパスがs3かhdfsであるかどうかわからない程度にs3またはhdfs特定のAPIを使用することはできません存在を確認してください。例えば

paths.filter(f => new java.io.File(f).exists) 

：

Seq("/tmp", "xx").filter(f => new java.io.File(f).exists) 
// res18: List[String] = List(/tmp) 

Answer 1

。 sparkでは、そうする最善の方法は、内部スパークハーフープ設定を使用することです。スパークセッション変数が「スパーク」と呼ばれる場合、次の操作を実行できます。

import org.apache.hadoop.fs.FileSystem 
import org.apache.hadoop.fs.Path 

val hadoopfs: FileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration) 

def testDirExist(path: String): Boolean = { 
    val p = new Path(path) 
    hadoopfs.exists(p) && hadoopfs.getFileStatus(p).isDirectory 
} 
val filteredPaths = paths.filter(p => testDirExists(p)) 
val dataframe = spark.read.parquet(filteredPaths: _*) 

Answer 2

あなたは、@のPsidomの答えのように無関係なファイルを除外できpaths firstly`をフィルタリングについてどのように

Answer 3

多分このようなものがあなたに役立つでしょうか？

def read(path: Seq[String]): Try[DataFrame] = Try(spark.read.parquet(p)) 


read("somePath") match { 
    case Success(df) => df.show() 
    case Failure(_) => Unit 
}