InputFomrat各种子类实例
0 TextInputFormat extends FileInputFomrat<LongWritable,Text> 是默认读取文件的切分器
其内的LineRecordReader:用来读取每一行的内容,
LineRecordReader:内的 nextKeyValue(){}中,key的赋值在:
initialize()方法内, key=start=split.getStart(); split假如对应文件 hello.txt 期内为hello you hello me
那么起始位置就是0
end = start + split.getLength(),
而行文本在方法 读取到的行字节长度=readLine(...)中读取,对应到LineReader.readLine(...) 170行
string key = getCurrentKey() string value = getCurrentValue() 中得到
然后在Mapper类中:
while(LineRecordReader.nextKeyValue()){
key = linerecordreader.getCurrentKey()'
value = linerecordreader.getCurrentValue()
map.(key,value,context); 不停的将键值对写出去
}
1 DBInputFormat:
DBInputFormat 在读取数据时,产生的键值对是 <LongWritable,DBWritable的实例>
LongWritable仍旧是偏移量,
可以参看 org.apache.hadoop.mapreduce.lib.db.DBRecordReader.nextKeyValue()/232行,如下
key.set(pos + split.getStart()); 来确认 表示的仍旧是偏移量
package inputformat; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import java.net.URI; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import mapreduce.MyWordCount; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.db.DBConfiguration; import org.apache.hadoop.mapreduce.lib.db.DBInputFormat; import org.apache.hadoop.mapreduce.lib.db.DBWritable; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * 目的: 将mysql/test库/myuser表中将字段id,name对应的属性通过 mapreduce(下面例子仅是通过map 没有reduce操作)将记录写出到hdfs中 * mysql--->map--->hdfs * 要运行本示例 * 1.把mysql的jdbc驱动放到各TaskTracker节点的hadoop/mapreduce/lib目录下 * 2.重启集群 * */ public class MyDBInputFormatApp { private static final String OUT_PATH = "hdfs://hadoop0:9000/out"; public static void main(String[] args) throws Exception{ Configuration conf = new Configuration(); // 连接数据库 代码尽量考前写 写在后面执行会报错 DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://hadoop0:3306/test", "root", "admin"); final FileSystem filesystem = FileSystem.get(new URI(OUT_PATH), conf); if(filesystem.exists(new Path(OUT_PATH))){ filesystem.delete(new Path(OUT_PATH), true); } final Job job = new Job(conf , MyDBInputFormatApp.class.getSimpleName()); // 创建job job.setJarByClass(MyDBInputFormatApp.class); job.setInputFormatClass(DBInputFormat.class);// 指定inputsplit具体实现类 // 下面方法参数属性为: 操作javabean, 对应表名, 查询条件,排序要求,需要查询的表字段 DBInputFormat.setInput(job, MyUser.class, "myuser", null, null, "id", "name");// // 设置map类和map处理的 key value 对应的数据类型 job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setNumReduceTasks(0); //指定不需要使用reduce,直接把map输出写入到HDFS中 job.setOutputKeyClass(Text.class); // 设置job output key 输出类型 job.setOutputValueClass(NullWritable.class);// 设置job output value 输出类型 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); job.waitForCompletion(true); } //<k1,v1>对应的是数据库对应表下记录位置,和这行对应的JavaBean, <k2,v2>表示经过map处理好输出结果 public static class MyMapper extends Mapper<LongWritable, MyUser, Text, NullWritable>{ protected void map(LongWritable key, MyUser value, Context context) throws java.io.IOException ,InterruptedException { context.write(new Text(value.toString()), NullWritable.get()); }; } /** * Writable是为了在Hadoop各节点之间传输使用的,因此需要实例化 * DBWritable表示和数据库传输时使用的 * @author zm * */ public static class MyUser implements Writable, DBWritable{ int id; String name; // 针对Writable 需要重写的方法 @Override public void write(DataOutput out) throws IOException { out.writeInt(id); Text.writeString(out, name); } @Override public void readFields(DataInput in) throws IOException { this.id = in.readInt(); this.name = Text.readString(in); } // 针对DBWritable需要重写的方法 @Override public void write(PreparedStatement statement) throws SQLException { statement.setInt(1, id); statement.setString(2, name); } @Override public void readFields(ResultSet resultSet) throws SQLException { this.id = resultSet.getInt(1); this.name = resultSet.getString(2); } @Override public String toString() { return id + "\t" + name; } } }
2 NLineInputFormat:
这种格式下,split的数量就不是由文件对应block块个数决定的, 而是由设置处理多少行决定,
比如一个文件 100行, 设置NlineInputFormat 处理2行,那么会产生50个map任务, 每个map任务
仍旧一行行的处理 会调用2次map函数、
package inputformat; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * TextInputFormat处理的数据来自于一个InputSplit。InputSplit是根据大小划分的。 * NLineInputFormat决定每个Mapper处理的记录数是相同的。 * 设置map处理行数多,则需要产生的map个数就会减少 */ public class MyNLineInputFormatApp { private static final String INPUT_PATH = "hdfs://hadoop0:9000/hello"; private static final String OUT_PATH = "hdfs://hadoop0:9000/out"; public static void main(String[] args) throws Exception{ // 定义conf Configuration conf = new Configuration(); //设置每个map可以处理多少条记录,默认是1行,这里设置为每个map处理的记录数都是2个 conf.setInt("mapreduce.input.lineinputformat.linespermap", 2); final FileSystem filesystem = FileSystem.get(new URI(OUT_PATH), conf); if(filesystem.exists(new Path(OUT_PATH))){ filesystem.delete(new Path(OUT_PATH), true); } // 定义job final Job job = new Job(conf , MyNLineInputFormatApp.class.getSimpleName()); job.setJarByClass(MyNLineInputFormatApp.class); // 定义 inputformat要处理的文件位置和具体处理实现类 FileInputFormat.setInputPaths(job, INPUT_PATH); job.setInputFormatClass(NLineInputFormat.class); // 设置map job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); // 设置reduce job.setReducerClass(MyReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 设置处理最终结果输出路径 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); job.waitForCompletion(true); } public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{ //解析源文件会产生2个键值对,分别是<0,hello you><10,hello me>;所以map函数会被调用2次 protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { //为什么要把hadoop类型转换为java类型? final String line = value.toString(); final String[] splited = line.split("\t"); //产生的<k,v>对少了 for (String word : splited) { //在for循环体内,临时变量word的出现次数是常量1 context.write(new Text(word), new LongWritable(1)); } }; } //map函数执行结束后,map输出的<k,v>一共有4个,分别是<hello,1><you,1><hello,1><me,1> //分区,默认只有一个区 //排序后的结果:<hello,1><hello,1><me,1><you,1> //分组后的结果:<hello,{1,1}> <me,{1}> <you,{1}> //归约(可选) //map产生的<k,v>分发到reduce的过程称作shuffle public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{ //每一组调用一次reduce函数,一共调用了3次 //分组的数量与reduce函数的调用次数有什么关系? //reduce函数的调用次数与输出的<k,v>的数量有什么关系? protected void reduce(Text key, java.lang.Iterable<LongWritable> values, org.apache.hadoop.mapreduce.Reducer<Text,LongWritable,Text,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { //count表示单词key在整个文件中的出现次数 long count = 0L; for (LongWritable times : values) { count += times.get(); } context.write(key, new LongWritable(count)); }; } }
3 KeyValueInputFormat:
如果行中有分隔符,那么分隔符前面的作为key,后面的作为value
如果行中没有分隔符,那么整行作为key,value为空
默认分隔符为 \t
package inputformat; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader; import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * 以hello文件内容为如下为例: * hello you * hello me * * 特点是: * Each line is divided into key and value parts by a separator byte. If no such a byte exists, the key will be the entire line and value will be empty 通过分隔符将每一行切分 切分后结果分别作为key value 如果没有分隔符,那么正一行就作为key 值为null 如果一行中有多个制表符的话,会取第一个作为key 剩余作为value,后面的也不会再分割了 KeyValueInputForamt他用特定分隔符分割来形成自己的key value,看源码(KeyValueLineRecordReader下为\t)默制默认分隔符为制表符 输出结果为: hello 1 you 1 helllo 1 me 1 */ public class MyKeyValueTextInputFormatApp { private static final String INPUT_PATH = "hdfs://hadoop0:9000/hello"; private static final String OUT_PATH = "hdfs://hadoop0:9000/out"; public static void main(String[] args) throws Exception{ Configuration conf = new Configuration(); conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t"); final FileSystem filesystem = FileSystem.get(new URI(OUT_PATH), conf); if(filesystem.exists(new Path(OUT_PATH))){ filesystem.delete(new Path(OUT_PATH), true); } // 创建job final Job job = new Job(conf , MyKeyValueTextInputFormatApp.class.getSimpleName()); job.setJarByClass(MyKeyValueTextInputFormatApp.class); // 设置InputFormat处理文件路径和具体操作实体类 FileInputFormat.setInputPaths(job, INPUT_PATH); job.setInputFormatClass(KeyValueTextInputFormat.class); // 设置map job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); // 设置reduce 这里reduce设置为0 job.setNumReduceTasks(0); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 设置最终结果输出路径 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); job.waitForCompletion(true); } public static class MyMapper extends Mapper<Text, Text, Text, LongWritable>{ protected void map(Text key, Text value, org.apache.hadoop.mapreduce.Mapper<Text,Text,Text,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { context.write(key, new LongWritable(1)); context.write(value, new LongWritable(1)); }; } }
4 GenericWritable
适用于 不同输入源下,多map输出类型不同
package inputformat; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.GenericWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat; import org.apache.hadoop.mapreduce.lib.input.MultipleInputs; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * MyMapper, MyMapper2的 v2输出类型一个是longWritable,一个是String, 两者需要统一成一个输出类型, * 以方便job在设置v2类型----> job.setMapOutputValueClass(MyGenericWritable.class) * * 文件hello 内容为: * hello you * hello me * * 文件hello2 内容为: * hello,you * hello,me * @author zm * * *结果: *[root@master hadoop]# hadoop fs -text /out/part-r-00000 Warning: $HADOOP_HOME is deprecated. hello 4 me 2 you 2 */ public class MyGenericWritableApp { private static final String OUT_PATH = "hdfs://master:9000/out"; public static void main(String[] args) throws Exception{ Configuration conf = new Configuration(); final FileSystem filesystem = FileSystem.get(new URI(OUT_PATH), conf); if(filesystem.exists(new Path(OUT_PATH))){ filesystem.delete(new Path(OUT_PATH), true); } final Job job = new Job(conf , MyGenericWritableApp.class.getSimpleName()); job.setJarByClass(MyGenericWritableApp.class); // 设置每种输入文件的位置 具体切分文件类 和对应的处理map类 MultipleInputs.addInputPath(job, new Path("hdfs://master:9000/hello"), KeyValueTextInputFormat.class, MyMapper.class); MultipleInputs.addInputPath(job, new Path("hdfs://master:9000/hello2"), TextInputFormat.class, MyMapper2.class); // 设置map //job.setMapperClass(MyMapper.class); //不应该有这一行 上面已经设置好了map类 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(MyGenericWritable.class); // 设置reduce job.setReducerClass(MyReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 设置输出结果存放路径 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); job.waitForCompletion(true); } public static class MyMapper extends Mapper<Text, Text, Text, MyGenericWritable>{ //解析源文件会产生2个键值对,分别是<hello,you> <hello,me>;所以map函数会被调用2次 // 处理后结果为: <hello,(MyGenericWritable(1),MyGenericWritable(1))> <you,(MyGenericWritable(1))> <me,(MyGenericWritable(1))> protected void map(Text key, Text value, org.apache.hadoop.mapreduce.Mapper<Text,Text,Text,MyGenericWritable>.Context context) throws java.io.IOException ,InterruptedException { context.write(key, new MyGenericWritable(new LongWritable(1))); context.write(value, new MyGenericWritable(new LongWritable(1))); }; } public static class MyMapper2 extends Mapper<LongWritable, Text, Text, MyGenericWritable>{ //解析源文件会产生2个键值对,分别是<0,(hello,you)><10,(hello,me)>;键值对内的()是我自己加上去的为了便于和前面偏移量的,区分开来 所以map函数会被调用2次 // 处理后结果为: <hello,(MyGenericWritable("1"),MyGenericWritable("1"))> <you,(MyGenericWritable("1"))> <me,(MyGenericWritable("1"))> protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,MyGenericWritable>.Context context) throws java.io.IOException ,InterruptedException { //为什么要把hadoop类型转换为java类型? final String line = value.toString(); final String[] splited = line.split(","); //产生的<k,v>对少了 for (String word : splited) { System.out.println("MyMapper2 word is:" + word); //在for循环体内,临时变量word的出现次数是常量1 final Text text = new Text("1"); context.write(new Text(word), new MyGenericWritable(text)); } }; } //map产生的<k,v>分发到reduce的过程称作shuffle public static class MyReducer extends Reducer<Text, MyGenericWritable, Text, LongWritable>{ //每一组调用一次reduce函数,一共调用了3次 //分组的数量与reduce函数的调用次数有什么关系? //reduce函数的调用次数与输出的<k,v>的数量有什么关系? protected void reduce(Text key, java.lang.Iterable<MyGenericWritable> values, org.apache.hadoop.mapreduce.Reducer<Text,MyGenericWritable,Text,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { //count表示单词key在整个文件中的出现次数 long count = 0L; for (MyGenericWritable times : values) { final Writable writable = times.get(); if(writable instanceof LongWritable) { count += ((LongWritable)writable).get(); } if(writable instanceof Text) { count += Long.parseLong(((Text)writable).toString()); } } context.write(key, new LongWritable(count)); }; } /** * * @author zm * */ public static class MyGenericWritable extends GenericWritable{ public MyGenericWritable() {} public MyGenericWritable(Text text) { super.set(text); } public MyGenericWritable(LongWritable longWritable) { super.set(longWritable); } // 数组里面存放要处理的类型 @Override protected Class<? extends Writable>[] getTypes() { return new Class[] {LongWritable.class, Text.class}; } } }
5 CombineTextInputFormat:
将输入源目录下多个小文件 合并成一个文件(split)来交给mapreduce处理 这样只会生成一个map任务
比如用户给的文件全都是10K那种的文件, 其内部也是用的TextInputFormat 当合并大小大于(64M)128M的时候,
也会产生对应个数的split
6 SequenceFile: 也是合并, 还没明白和CombineTextInputFormat的区别在哪里:
import java.io.File; import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; import java.util.Collection; import org.apache.commons.io.FileUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.SequenceFile.Writer; import org.apache.hadoop.io.Text; public class SequenceFileMore { public static void main(String[] args) throws IOException, URISyntaxException { final Configuration conf = new Configuration(); final FileSystem fs = FileSystem.get(new URI("hdfs://h2single:9000/"), conf); Path path = new Path("/sf_logs"); //写操作 final Writer writer = new SequenceFile.Writer(fs, conf, path, Text.class, BytesWritable.class); // false表示不迭代子目录 Collection<File> listFiles = FileUtils.listFiles(new File("/usr/local/logs"), new String[]{"log"}, false); for(File file : listFiles){ // 将/usr/local/logs下的所有.log文件 以对应文件文件名为key 对应文件内容字节数组为value 共同写入到/sf_logs内 String fileName = file.getName(); Text key = new Text(fileName); byte[] bytes = FileUtils.readFileToByteArray(file); BytesWritable value = new BytesWritable(bytes); writer.append(key, value); } IOUtils.closeStream(writer); //读操作 final SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf); final Text key = new Text(); final BytesWritable val = new BytesWritable(); while (reader.next(key, val)) { String fileName = "/usr/local/logs_bak/" + key.toString(); File file = new File(fileName); FileUtils.writeByteArrayToFile(file, val.getBytes()); } IOUtils.closeStream(reader); } }
7 MultipleInputs: 对应于 多个文件处理类型下 比如又要处理数据库的文件 同时又要处理小文件
这里仅将main函数拼接展示下,各自对应的mapper类自己去写: