如何在HadoopMap/reduce中写入Map器中的多个文件?

mqkwyuun  于 2021-05-30  发布在  Hadoop
关注(0)|答案(3)|浏览(452)

我有一个map reduce作业,没有任何reducer,它解析输入文件并以parquet格式在mappers磁盘上写入一些输出。由于此作业可以将多个文件夹中的文件作为输入(每个日期一个文件夹),因此我还希望将输出拆分为多个文件夹,如:

01JAN15
    output-0000
    output-0001

02JAN15
    output-0000
    output-0001

我查看了文档中的multipleoutput format类,但它似乎只在reduce部分的几个文件夹中工作。
不知何故,写入同一目录中的多个文件是可行的,但是当我尝试写入多个目录时就会出现异常(可能是因为某些Map程序试图同时创建同一目录?)。
仅供参考,我的代码在Map器中如下所示:

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

我定义输出格式如下:

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

我得到的例外是:

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)

你知道我想做的事是否可行吗?我做错什么了?谢谢!

fcy6dtqo

fcy6dtqo1#

我知道你的意思,但不幸的是(据我所知)你需要自己去做
只需创建从配置和实现工具接口扩展的驱动程序类。然后,您可以简单地配置一个回调,在mapred执行完成后调用该回调,然后只编写代码来移动相应文件夹中的文件。
这是链接
Map完成后回调

yc0p9oo0

yc0p9oo02#

您可以使用分区输出到不同的文件中。

cygmwpex

cygmwpex3#

一个输出文件不能由多个进程(Map器或reducer)编写,因此为了生成多个输出文件,我要么定义自定义分区,要么在reducer中对数据进行分组,并在输出文件名中输入键。Map程序不可能将来自多个输入文件的数据写入同一个文件中。

相关问题