使用Spark从HDFS读取时如何避免出现“非文件”异常

Question

我在初始EMR步骤中将一棵文件树从S3复制到具有S3DistCP的 HDFS 。hdfs dfs -ls -R hdfs:///data_dir显示预期的文件，如下所示：

/data_dir/year=2015/
/data_dir/year=2015/month=01/
/data_dir/year=2015/month=01/day=01/
/data_dir/year=2015/month=01/day=01/data01.12345678
/data_dir/year=2015/month=01/day=01/data02.12345678
/data_dir/year=2015/month=01/day=01/data03.12345678

“目录”被列出为零字节文件。

然后，我运行一个启动步骤，该步骤需要读取这些文件。因此，加载代码为：

sqlctx.read.json('hdfs:///data_dir, schema=schema)

作业失败，出现Java异常

java.io.IOException: Not a file: hdfs://10.159.123.38:9000/data_dir/year=2015

我曾经（也许很幼稚）假设spark会递归地下降“ dir tree”并加载数据文件。如果我指向S3，它将成功加载数据。

我误会了HDFS吗？我可以告诉spark忽略零字节文件吗？我可以使用S3DistCp压扁树吗？

Answer 1

在当前Spark上下文的Hadoop配置中，先配置Hadoop InputFormat的“递归”读取，以获取sql ctx

val hadoopConf = sparkCtx.hadoopConfiguration
hadoopConf.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

这将为“不是文件”提供解决方案。接下来，要读取多个文件：

Hadoop作业从多个目录获取输入文件

或将文件列表合并为单个数据框：

使用Spark从目录中读取多个文件