SparkのDataFrameReaderに渡すパスから存在しないものを予め除外する

SparkのDataFrameReaderの json メソッドには、HDFSやS3に対するglobパターンを複数指定できます。

sqlContext.read.json(
  "s3://mybucket/data/year=2016/month=11/day=*/",
  "s3://mybucket/data/year=2016/month=12/day=*/",
  "s3://mybucket/data/year=2017/month=01/day=*/"
)

ただし、マッチするオブジェクトが一つも存在しないようなパスが含まれているとエラーになります。

org.apache.spark.sql.AnalysisException: Path does not exist: s3://mybucket/data/year=2017/month=01/day=*/;

今のところエラーを無視して存在するパスだけ読み出すようなオプションは用意されていないようです。
このため、予めglobパターンに対してマッチするオブジェクトが存在するかを確認する必要があります。
これは org.apache.hadoop.fs パッケージを利用すると実現できます。

import java.net.URI
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkContext
import org.apache.spark.deploy.SparkHadoopUtil

def checkPathExists(path: String): Boolean = {
  val fs = FileSystem.get(new URI(path), sc.hadoopConfiguration)
  val p = new Path(path)
  val qualified = p.makeQualified(fs.getUri, fs.getWorkingDirectory)
  val glob = SparkHadoopUtil.get.globPathIfNecessary(qualified)
  glob.nonEmpty && fs.exists(glob.head)
}

checkPathExists("s3://mybucket/data/year=2017/month=01/day=*/")
// => false

参考

• https://github.com/apache/spark/blob/v2.0.2/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala#L380-L391