当前位置:  NOSQL>hadoop/hbase

Hadoop中MultipleOutputFormat示例代码

 
    发布时间:2013-7-13  


    本文导语:  Hadoop的MapReduce中多文件输出默认是TextOutFormat,输出为part-r- 00000和part-r-00001依次递增的文件名。hadoop提供了MultipleOutputFormat类,重写该类可实现定制自定义的文件名。 具体参考代码如下: (1)首先设置: job....

   HadoopMapReduce中多文件输出默认是TextOutFormat,输出为part-r- 00000和part-r-00001依次递增的文件名。hadoop提供了MultipleOutputFormat,重写该类可实现定制自定义的文件名。

  具体参考代码如下:

  (1)首先设置:  

job.setOutputFormatClass(m_customOutputFormatClass);

  (2)具体代码:

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
/**
 * TextOutputFormat extension which enables writing the mapper/reducer's output in multiple files.<br>
 * <p>
 * <b>WARNING</b>: The number of different folder shuoldn't be large for one mapper since we keep an
 * {@link RecordWriter} instance per folder name.
 * </p>
 * <p>
 * In this class the folder name is defined by the written entry's key.<br>
 * To change this behavior simply extend this class and override the
 * {@link HdMultipleFileOutputFormat#getFolderNameExtractor()} method and create your own
 * {@link FolderNameExtractor} implementation.
 * </p>
 *
 *
 *
 * @param <K> - Keys type
 * @param <V> - Values type
 */
public class HdMultipleFileOutputFormat<K, V> extends TextOutputFormat<K, V> {
    private String folderName;
    private class MultipleFilesRecordWriter extends RecordWriter<K, V> {
        private Map<String, RecordWriter<K, V>> fileNameToWriter;
        private FolderNameExtractor<K, V> fileNameExtractor;
        private TaskAttemptContext job;
        public MultipleFilesRecordWriter(FolderNameExtractor<K, V> fileNameExtractor, TaskAttemptContext job) {
            fileNameToWriter = new HashMap<String, RecordWriter<K, V>>();
            this.fileNameExtractor = fileNameExtractor;
            this.job = job;
        }
        @Override
        public void write(K key, V value) throws IOException, InterruptedException {
            String fileName = fileNameExtractor.extractFolderName(key, value);
            RecordWriter<K, V> writer = fileNameToWriter.get(fileName);
            if (writer == null) {
                writer = createNewWriter(fileName, fileNameToWriter, job);
                if (writer == null) {
                    throw new IOException("Unable to create writer for path: " + fileName);
                }
            }
            writer.write(key, value);
        }
        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            for (Entry<String, RecordWriter<K, V>> entry : fileNameToWriter.entrySet()) {
                entry.getValue().close(context);
            }
        }
    }
    private synchronized RecordWriter<K, V> createNewWriter(String folderName,
            Map<String, RecordWriter<K, V>> fileNameToWriter, TaskAttemptContext job) {
        try {
            this.folderName = folderName;
            RecordWriter<K, V> writer = super.getRecordWriter(job);
            this.folderName = null;
            fileNameToWriter.put(folderName, writer);
            return writer;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
    @Override
    public Path getDefaultWorkFile(TaskAttemptContext context, String extension) throws IOException {
        Path path = super.getDefaultWorkFile(context, extension);
        if (folderName != null) {
            String newPath = path.getParent().toString() + "/" + folderName + "/" + path.getName();
            path = new Path(newPath);
        }
        return path;
    }
    @Override
    public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        return new MultipleFilesRecordWriter(getFolderNameExtractor(), job);
    }
    public FolderNameExtractor<K, V> getFolderNameExtractor() {
        return new KeyFolderNameExtractor<K, V>();
    }
    public interface FolderNameExtractor<K, V> {
        public String extractFolderName(K key, V value);
    }
    private static class KeyFolderNameExtractor<K, V> implements FolderNameExtractor<K, V> {
        public String extractFolderName(K key, V value) {
            return key.toString();
        }
    }
}


  • 本站(WWW.)旨在分享和传播互联网科技相关的资讯和技术,将尽最大努力为读者提供更好的信息聚合和浏览方式。
    本站(WWW.)站内文章除注明原创外,均为转载,整理或搜集自网络.欢迎任何形式的转载,转载请注明出处.
    转载请注明:文章转载自:[169IT-IT技术资讯]
    本文标题:Hadoop中MultipleOutputFormat示例代码
相关文章推荐:


站内导航:


特别声明:169IT网站部分信息来自互联网,如果侵犯您的权利,请及时告知,本站将立即删除!

©2012-2021,,E-mail:www_#163.com(请将#改为@)

浙ICP备11055608号-3