相关疑难解决方法(0)

我正在尝试阅读Sequencescala中Paths中的文件.下面是示例(伪)代码:

val paths = Seq[String] //Seq of paths
val dataframe = spark.read.parquet(paths: _*)

现在,在上面的序列中,存在一些路径而一些路径不存在.有没有办法在读取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,以检查是否存在.

我正在尝试创建或打开一个文件来存储HDFS中的一些输出,但是当我exists在下面的代码片段的倒数第二行调用方法时,我得到一个NullPointerException :

DistributedFileSystem dfs = new DistributedFileSystem();
Path path = new Path("/user/hadoop-user/bar.txt");
if (!dfs.exists(path)) dfs.createNewFile(path);
FSDataOutputStream dos = dfs.create(path);

这是堆栈跟踪:

java.lang.NullPointerException
        at org.apache.hadoop.dfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:390)
        at org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:667)
        at ClickViewSessions$ClickViewSessionsMapper.map(ClickViewSessions.java:80)
        at ClickViewSessions$ClickViewSessionsMapper.map(ClickViewSessions.java:65)
        at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:47)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:227)
        at org.apache.hadoop.mapred.TaskTracker$Child.main(TaskTracker.java:2209)

问题是什么？

任何人都可以建议在pyspark中检查文件存在的最佳方法。

目前正在使用以下方法进行检查，请指教。

def path_exist(path):

try:
    rdd=sparkSqlCtx.read.format("orc").load(path)
    rdd.take(1)
    return True

except Exception as e:
    return False