fs.rename（new Path（rawFileName）， new Path（Process FileName））不工作

Question

我正在研究基于Scala的Apache Spark实现，用于从远程位置到HDFS的数据载入，然后研究从HDFS到Hive表的数据摄取。

使用我的第一份spark工作，我已在某个位置将数据/文件加载到HDFS中，例如-

hdfs://sandbox.hortonworks.com:8020/data/analytics/raw/文件夹

让我们考虑一下，在入职CT_Click_Basic.csv和CT_Click_Basic1.csv.gz文件后，我在HDFS中有以下文件[共享位置的文件名将是此处的文件夹名称，其内容将出现在part xxxxx文件中]：

[root@sandbox~]#hdfs dfs-ls/数据/分析/原始/*/找到3项

-rw-r--r--3 chauhan。bhupesh hdfs 0 2017-07-27 15:02/数据/分析/原始/CT\u Click\u基本。csv/\u成功

-rw-r--r--3 chauhan。bhupesh hdfs 8383 2017-07-27 15:02/数据/分析/原始/CT\u Click\u基本。csv/part-00000

-rw-r--r--3chauhan.bhupeshhdfs8395 2017-07-27 15:02 /data/analytics/raw/CT_Click_Basic.csv/part-00001

找到2个项目

-rw-r--r--3 chauhan。bhupesh hdfs 0 2017-07-27 15:02/数据/分析/原始/CT\u点击基础1。csv。gz/\u成功

-rw-r--r--3 chauhan。bhupesh hdfs 16588 2017-07-27 15:02/数据/分析/原始/CT\u点击基础1。csv。gz/part-00000

现在使用我的另一个Spark作业，我想根据每个阶段执行的任务，将这些文件原样从/raw文件夹移动到/process，然后最终移动到HDFS中的/archive文件夹。

为此，我首先使用以下代码获取/raw文件夹下存在的所有文件的列表：

    def listAllFilesFolderInDir(filePath:String,recursiveTraverse:Boolean,filePaths: ListBuffer[Path]) : ListBuffer[Path] = {
val files = GlobalContext.hdfs.listStatus(new Path(filePath))
files.foreach { fileStatus => {
           if(!fileStatus.isDirectory()) {
                filePaths+=fileStatus.getPath()      
            }
            else {
                listAllFilesFolderInDir(fileStatus.getPath().toString(), recursiveTraverse, filePaths)
            }
        }
  }   
  filePaths
}

然后使用以下代码行，我试图重命名/移动文件夹中 /raw文件到 /process文件夹：

var inputDir = "/data/analytics/raw"
var outputDir = "/data/analytics/process"
var filePaths = new ListBuffer[Path]()
var pathArray = listAllFilesFolderInDir(inputDir, true, filePaths)
val fs= <Getting hdfs FileSystem Instance Here>
for(path<-pathArray){
   var pathSplit = path.toString().split("/")
   var pathSplitSize = pathSplit.size
   val rawFileName = inputDir + "/" + pathSplit(pathSplitSize-2) + "/" + pathSplit(pathSplitSize-1)
   val processFileName = outputDir + "/" + pathSplit(pathSplitSize-2) + "/" + pathSplit(pathSplitSize-1)
   fs.rename(new Path(rawFileName), new Path(processFileName))
 }

但我无法使用上述编写的代码移动/重命名这些文件。我试着调试代码，发现fs。rename（）返回“false”。

请注意：当我在/数据/分析/原始文件夹ex-CT中手动复制任何文件时，我能够实现文件重命名/移动。csv[或任何其他文件]，然后运行fs。rename（），但它不适用于Part-xxxxx文件。

有什么是我遗漏的吗？

任何快速的帮助将不胜感激。

你好Bhupesh

Answer 1

如果new Path（rawFileName）不存在，重命名可以返回false。
在fs.rename检查文件是否存在之前：

if (fs.exists(somePath)) {
 fs.rename...
}

另一个原因可能是您尝试重命名的文件被某人使用。或者如果您尝试重命名目录，其中的某些文件可能会被某人使用。
要确保这是根本原因，请尝试重命名代码中的其他文件：

var inputDir = "/data/analytics/raw"
var outputDir = "/data/analytics/process"
var filePaths = new ListBuffer[Path]()
var pathArray = listAllFilesFolderInDir(inputDir, true, filePaths)
val fs= <Getting hdfs FileSystem Instance Here>
for(path<-pathArray){
   var pathSplit = path.toString().split("/")
   var pathSplitSize = pathSplit.size
   val rawFileName = inputDir + "/" + pathSplit(pathSplitSize-2) + "/" + pathSplit(pathSplitSize-1)
   val processFileName = outputDir + "/" + pathSplit(pathSplitSize-2) + "/" + pathSplit(pathSplitSize-1)
   fs.rename(new Path("**/TESTDIR1**"), new Path("**/TESTDIR2**"))
 }

如果这次重命名成功，根本原因是真的处于竞争状态。

Answer 2

最后，我得到了这个问题。实际上，我正试图重命名/data/analytics/raw/folder中的文件。csv/part-xxxxx收件人/数据/分析/流程/文件夹。csv/part-xxxxx，其中HDFS中存在/数据/分析/流程，但不存在“folder.csv”；因此，它在重命名时返回了false。我在代码中添加了以下行，效果很好

var inputDir = "/data/analytics/raw"
var outputDir = "/data/analytics/process"
var filePaths = new ListBuffer[Path]()
var pathArray = listAllFilesFolderInDir(inputDir, true, filePaths)
val fs= <Getting hdfs FileSystem Instance Here>
for(path<-pathArray){
   var pathSplit = path.toString().split("/")
   var pathSplitSize = pathSplit.size

   val rawFileName = inputDir + "/" + pathSplit(pathSplitSize-2) + "/" + pathSplit(pathSplitSize-1)

   var processFolderName = outputDir + "/" + pathSplit(pathSplitSize-2)
   var processFolderPath = new Path(processFolderName)
   if(!(fs.exists(processFolderPath)))
         fs.mkdirs(processFolderPath)
   val processFileName = processFolderName + "/" + pathSplit(pathSplitSize-1)
   fs.rename(new Path(rawFileName), new Path(processFileName))
 }