具有 FileAlreadyExistsException 的 Reducer 中的 Hadoop MultipleOutputs

Question

我在 reducer 中使用 MultipleOutputs。多重输出会将文件写入名为 NewIdentities 的文件夹。代码如下所示:

private MultipleOutputs<Text,Text> mos;
@Override
public void reduce(Text inputKey, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        ......
        // output to change report
        if (ischangereport.equals("TRUE")) {
            mos.write(new Text(e.getHID()), new Text(changereport.deleteCharAt(changereport.length() - 1).toString()), "NewIdentities/");
        }
    }
}

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

@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
    mos.close();
}

它可以在以前运行。但是当我今天运行它时，它会抛出如下异常。我的 hadoop 版本是 2.4.0。

错误:org.apache.hadoop.fs.FileAlreadyExistsException:/CaptureOnlyMatchIndex9/TEMP/ChangeReport/NewIdentities/-r-00000 for client 192.168.71.128 already exists at org.apache.hadoop.hdfs.server.namenode.FSNamesystem. org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInt(FSNamesystem.java:2225) 在 org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFile(FSNamesystem. java:2178) 在 org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.create(NameNodeRpcServer.java:520) 在 org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.create(ClientNamenodeProtocolServerSideTranslatorPB.java:354) 在 org .apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java) 在 org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:585) 在 org.apache。 hadoop.ipc.RPC$Server.call(RPC.java:928) 在 org.apach e.hadoop.ipc.Server$Handler$1.run(Server.java:2013) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2009) 在 java.security.AccessController.doPrivileged( native 方法)在 javax.security.auth.Subject.doAs(Subject.java:415) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1548) 在 org.apache.hadoop.ipc.Server$ Handler.run(Server.java:2007) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) 在 sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java: 45) 在 java.lang.reflect.Constructor.newInstance(Constructor.java:526) 在 org.apache.hadoop.ipc.RemoteException.instantiateException(RemoteException.java:106) 在 org.apache.hadoop.ipc.RemoteException.unwrapRemoteException (RemoteException.java:73) 在 org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:1604) 在 org.apache。 hadoop.hdfs.DFSClient.create(DFSClient.java:1465) 在 org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1390) 在 org.apache.hadoop.hdfs.DistributedFileSystem$6.doCall(DistributedFileSystem.java :394) 在 org.apache.hadoop.hdfs.DistributedFileSystem$6.doCall(DistributedFileSystem.java:390) 在 org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) 在 org.apache.hadoop.hdfs .DistributedFileSystem.create(DistributedFileSystem.java:390) 在 org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:334) 在 org.apache.hadoop.fs.FileSystem.create(FileSystem.java:906) 在org.apache.hadoop.fs.FileSystem.create(FileSystem.java:887) 在 org.apache.hadoop.fs.FileSystem.create(FileSystem.java:784) 在 org.apache.hadoop.mapreduce.lib.output。 TextOutputFormat.getRecordWriter(TextOutputFormat.java:132) 在 org.apache.hadoop.mapreduce.lib.output.MultipleOutputs.getRecordWriter(MultipleOutputs.java:475) 在

Answer 1

我找到了原因。因为在我的一个 reducer 中，它耗尽了内存。所以它隐式地抛出了一个内存不足的异常。 hadoop 停止当前的多重输出。也许另一个 reducer 线程想要输出，所以它创建了另一个多输出对象，所以发生了碰撞。