|
18 | 18 | package org.apache.spark.sql.hive |
19 | 19 |
|
20 | 20 | import org.apache.hadoop.conf.Configuration |
21 | | -import org.apache.hadoop.fs.{Path, PathFilter} |
| 21 | +import org.apache.hadoop.fs.{FileSystem, Path, PathFilter} |
22 | 22 | import org.apache.hadoop.hive.conf.HiveConf |
23 | 23 | import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants._ |
24 | 24 | import org.apache.hadoop.hive.ql.exec.Utilities |
@@ -68,6 +68,8 @@ class HadoopTableReader( |
68 | 68 | math.max(sc.hiveconf.getInt("mapred.map.tasks", 1), sc.sparkContext.defaultMinPartitions) |
69 | 69 | } |
70 | 70 |
|
| 71 | + @transient private lazy val fs = FileSystem.get(sc.hiveconf) |
| 72 | + |
71 | 73 | // TODO: set aws s3 credentials. |
72 | 74 |
|
73 | 75 | private val _broadcastedHiveConf = |
@@ -218,11 +220,10 @@ class HadoopTableReader( |
218 | 220 | * returned in a single, comma-separated string. |
219 | 221 | */ |
220 | 222 | private def applyFilterIfNeeded(path: Path, filterOpt: Option[PathFilter]): Option[String] = { |
221 | | - if (path.getFileSystem(sc.hiveconf).exists(path)) { |
| 223 | + if (fs.exists(path)) { |
222 | 224 | // if the file exists |
223 | 225 | filterOpt match { |
224 | 226 | case Some(filter) => |
225 | | - val fs = path.getFileSystem(sc.hiveconf) |
226 | 227 | val filteredFiles = fs.listStatus(path, filter).map(_.getPath.toString) |
227 | 228 | if (filteredFiles.length > 0) { |
228 | 229 | Some(filteredFiles.mkString(",")) |
|
0 commit comments