I have a map reduce job without any reducer that parses input files and write some output on disk in the mappers in Parquet format.
Since this job can take files from several folders as input (one folder per date), I would like to also have the output split into folders, like :

```
01JAN15
    output-0000
    output-0001

02JAN15
    output-0000
    output-0001

```

I looked at the MultipleOutput format class in the [documentation](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.html), but it seems to be working only to write in several folders in the reduce part.

Somehow, writing into multiple files in the same directory works, but i get an exception as soon as i try to write in multiple directories (maybe because some mappers are trying to create the same directory at the same time ?).

FYI my code looks like that in the mapper:

```
mos.write("pb", null, message, date + "/output");

```

and I define the output format like that :

```
MultipleOutputs.addNamedOutput(job, "pb", ProtoParquetOutputFormat.class,
Void.class, com.google.protobuf.Message.class);

```

The exception I get is:

```
15/01/11 15:05:09 WARN ipc.Client: interrupted waiting to send rpc request to server
java.lang.InterruptedException
    at java.util.concurrent.FutureTask.awaitDone(FutureTask.java:400)
    at java.util.concurrent.FutureTask.get(FutureTask.java:187)
    at org.apache.hadoop.ipc.Client$Connection.sendRpcRequest(Client.java:1046)
    at org.apache.hadoop.ipc.Client.call(Client.java:1441)
    at org.apache.hadoop.ipc.Client.call(Client.java:1399)
    at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232)
    at com.sun.proxy.$Proxy9.getBlockLocations(Unknown Source)
    at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.getBlockLocations(ClientNamenodeProtocolTranslatorPB.java:254)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
    at com.sun.proxy.$Proxy10.getBlockLocations(Unknown Source)
    at org.apache.hadoop.hdfs.DFSClient.callGetBlockLocations(DFSClient.java:1220)
    at org.apache.hadoop.hdfs.DFSClient.getLocatedBlocks(DFSClient.java:1210)
    at org.apache.hadoop.hdfs.DFSClient.getLocatedBlocks(DFSClient.java:1200)
    at org.apache.hadoop.hdfs.DFSInputStream.fetchLocatedBlocksAndGetLastBlockLength(DFSInputStream.java:271)
    at org.apache.hadoop.hdfs.DFSInputStream.openInfo(DFSInputStream.java:238)
    at org.apache.hadoop.hdfs.DFSInputStream.<init>(DFSInputStream.java:231)
    at org.apache.hadoop.hdfs.DFSClient.open(DFSClient.java:1498)
    at org.apache.hadoop.hdfs.DistributedFileSystem$3.doCall(DistributedFileSystem.java:302)
    at org.apache.hadoop.hdfs.DistributedFileSystem$3.doCall(DistributedFileSystem.java:298)
    at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
    at org.apache.hadoop.hdfs.DistributedFileSystem.open(DistributedFileSystem.java:298)
    at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:766)
    at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:272)
    at parquet.hadoop.ParquetFileReader$2.call(ParquetFileReader.java:180)
    at parquet.hadoop.ParquetFileReader$2.call(ParquetFileReader.java:176)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

```

Do you know if what i want to do is possible? What am I doing wrong?
Thanks !