java - hadoop 当前租户正在尝试重新创建文件

Question

我的 MapReduce 作业按日期处理数据，需要将输出写入特定的文件夹结构。目前的期望是生成以下结构的输出:

天 => 天/月/文件

reducer 类:

public static class Reducerclass  extends Reducer<Text,Text,Text,Text> {
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        private MultipleOutputs<Text, Text> multipleOutputs;
        public void reduce(Text rkey, Iterable<Text> rvalue, Context context) throws IOException, InterruptedException {
            multipleOutputs = new MultipleOutputs<Text, Text>(context);
            for(Text value : rvalue) {              

                Date date = new Date(Long.parseLong(value.toString()));     
                String datestring = dateFormat.format(date); 
                //              context.write(new Text(rkey + ", "),new Text(", "+value +", "+datestring));


                String[] parts = datestring.split(" ");
                String[] dateparts = parts[0].split("-");

                String day = dateparts[2]; 
                String month = dateparts[1]; 
                String year = dateparts[0]; 


                String baseFileName ="Days";  
                String dayFolderpath = baseFileName + day + "/" + month + "/" + year;

                String outputFileName = baseFileName + "/" + month + "/" + day; 
                if(datestring.equals("2017-10-26 20:30:42") || datestring.equals("2017-10-26 20:29:32")){

                    multipleOutputs.write(new Text(rkey + ", "), new Text(", "+value +", "+datestring),outputFileName) ;
                }
                else{   

                multipleOutputs.write(new Text(rkey + ", "), new Text(", "+value +", "+datestring),generateFileName(new Text(rkey),new Text(value) )) ;
                }

            }   
        }
        String generateFileName(Text key, Text value){
            return key.toString() + "_" + value.toString();     
        }

        @Override
        public void setup(Context context){
            multipleOutputs = new MultipleOutputs<Text, Text>(context);
        }

        @Override
        public void cleanup(final Context context) throws IOException, InterruptedException{
            multipleOutputs.close();
        }
    }

当我运行 jar 时收到以下错误消息

17/11/14 18:30:17 INFO mapreduce.Job: Task Id : attempt_1510547616164_0054_r_000000_0, Status : FAILED
Error: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException): failed to create file /user/hdfs/OutputFile/_temporary/1/_temporary/attempt_1510547616164_0054_r_000000_0/Days/10/26-r-00000 for DFSClient_attempt_1510547616164_0054_r_000000_0_757297700_1 for client 10.0.2.15 because current leaseholder is trying to recreate file.
    at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.recoverLeaseInternal(FSNamesystem.java:3171)

Answer 1

我相信你错过的是:

job.setReducerClass(ReducerFormatMultiOutput.class);

请贴出驱动代码，找出具体原因。