本文转自:http://www.randyho.com/?p=41
接触hadoop一年多了,但是自己一直没有用hadoop写过什么程序。最近,由于项目需要,将一些文件转换成hadoop的MapFile。网上的例子基本是直接处理文本输入,自定义输入格式的见到两个,但是都是用的旧的API,用新API写的还没有,可能高手不屑于写这些。但是处理自定义输入是每个用hadoop的人都要学会才行的,因为不是每个人的输入都是文本文件。
数据输入是hadoop的第一步,不能读自己的数据,后面的处理就无从谈起。文本格式处理起来容易些,对于二进制格式的文件,虽然hadoop有一个SequenceFileInputFormat,可以先把自己的数据转成SequenceFile,再处理,但是这样要多一倍的处理时间、存储空间。无奈之下,参考了hadoop的源代码,自己写了个ConverterInputFormat,在这里贴出来,供大家参考。
代码是基于hadoop 0.20的,其中的FetcherOutput是用Java的DataOutputStream写入到本地磁盘的,可以换成自己想要的格式。
ConvertertRecordReader好像必须有个默认的构造器。
package com.randyho.hadoop.converter;
import java.io.DataInputStream;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import com.randyho.FetcherOutput;
public class ConverterInputFormat extends FileInputFormat<Text, FetcherOutput> {
// Do not split files.
protected boolean isSplitable(JobContext context, Path file) {
return false;
}
public RecordReader<Text, FetcherOutput> createRecordReader(
InputSplit split, TaskAttemptContext context) throws IOException,
InterruptedException {
return new ConvertertRecordReader();
}
class ConvertertRecordReader extends RecordReader<Text, FetcherOutput> {
private DataInputStream dis;
private Text key = null;
private FetcherOutput value;
private boolean more = true;
private Configuration conf;
public ConvertertRecordReader(){
key = new Text();
value = new FetcherOutput();
more = true;
}
public void close() throws IOException {
if (dis != null) {
dis.close();
}
}
public Text getCurrentKey() throws IOException, InterruptedException {
return key;
}
public FetcherOutput getCurrentValue() throws IOException,
InterruptedException {
return value;
}
public float getProgress() throws IOException, InterruptedException {
return more ? 0f : 100f;
}
public void initialize(InputSplit gensplit, TaskAttemptContext context)
throws IOException, InterruptedException {
FileSplit split = (FileSplit) gensplit;
conf = context.getConfiguration();
Path file = split.getPath();
FileSystem fs = file.getFileSystem(conf);
System.out.println("reading: " + file);
// open the file
dis = fs.open(split.getPath());
}
public boolean nextKeyValue() throws IOException, InterruptedException {
if (dis.available() != 0) {
value.readFields(dis);
key.set(value.getUrl());
return true;
} else {
more = false;
return false;
}
}
}
}
分享到:
相关推荐
自定义inputFormat&&outputFormat1
Hadoop 自定义 Partitioner 实现
Hadoop 自定义 Partitioner 实现
hadoop自定义类型编程
Hadoop 自定义 Partitioner 源代码
自定义MapReduce的InputFormat,实现提取指定开始与结束限定符的内容。
Hadoop 代码使用方式 ...hadoop jar hadoop-mapreduce-custom-inputformat-1.0-SNAPSHOT.jar org.apache.hadoop.mapreduce.sample.SmallFileWordCount -Dmapreduce.input.fileinputformat.split.maxsize=10
Apache Hive 的 InputFormat,在查询 SequenceFiles 时将返回 (Text) 键和 (Text) 值。 我需要在不拆分内容的情况下完整解析大量文本文件。 HDFS 在处理大型连续文件时提供最佳吞吐量,因此我使用 Apache Mahout 将...
博客Hadoop机架感知配置自主实现的jar包 http://blog.csdn.net/lemonZhaoTao/article/details/70991813
在windows环境下开发hadoop时,需要配置HADOOP_HOME环境变量,变量值D:\hadoop-common-2.7.3-bin-master,并在Path追加%HADOOP_HOME%\bin,有可能出现如下错误: org.apache.hadoop.io.nativeio.NativeIO$Windows....
04-hadoop的自定义排序实现.avi 05-mr程序中自定义分组的实现.avi 06-shuffle机制.avi 07-mr程序的组件全貌.avi 08-textinputformat对切片规划的源码分析.avi 09-倒排索引的mr实现.avi 10-多个job在同一个...
可以自定义安装jdk和hadoop(格式为*tar.gz) 注意事项 1、安装完jdk和hadoop请手动source /etc/profile 刷新环境变量 2测试脚本环境为centOS6,其他操作系统会有些配置不一样,请手动调整 资源描述: 安装Linux教程...
《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf...
Hadoop 是一个处理、存储和分析海量的分布式、非结构化数据的开源框架。最初由 Yahoo 的工程师 Doug Cutting 和 Mike Cafarella Hadoop 是一个处理、存储和分析海量的分布式、非结构化数据的开源框架。最初由 Yahoo...
45_hadoop2.x_温度排序,分区,分组,自定义封装类02 46_hadoop2.x_温度排序,分区,分组,自定义封装类03 47_hadoop2.x_温度排序,分区,分组,自定义封装类04 48_hadoop2.x_温度排序,分区,分组,自定义封装类05 ...
Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下,开发分布式程序。充分利用集群的威力进 Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不...
本书从hadoop的缘起开始,由浅入深,结合理论和实践,全方位地介绍hado叩这一高性能处理海量数据集的理想工具。全书共14章,3个附录,涉及的主题包括:haddoop简介:mapreduce简介:hadoop分布式文件系统;hadoop的i...
hadoop-annotations-3.1.1.jar hadoop-common-3.1.1.jar hadoop-mapreduce-client-core-3.1.1.jar hadoop-yarn-api-3.1.1.jar hadoop-auth-3.1.1.jar hadoop-hdfs-3.1.1.jar hadoop-mapreduce-client-hs-3.1.1.jar ...
映射文件输入格式MapFiles 的 Hadoop InputFormat,它在将任何内容传递给映射器之前过滤不相关的 FileSplits。目的假设您的文件系统中有一些带有排序键的非常大的文件,并且键已排序。 在编写 MapReduce 作业时,您...