MapReduce类型与格式(输入与输出)

一、输入格式

(1)输入分片记录

①JobClient通过指定的输入文件的格式来生成数据分片InputSplit;

②一个分片不是数据本身,而是可分片数据的引用;

③InputFormat接口负责生成分片;

源码位置:org.apache.hadoop.mapreduce.lib.input包(新)

org.apache.hadoop.mapred.lib 包(旧)

查看其中FileInputFormat类中的getSplits()方法;

computeSplitSize()函数决定分片大小;

各种输入类的结构关系图:

(2)文件输入

抽象类:FileInputFormat

①FileInputFormat是所有使用文件作为数据源的InputFormat实现的基类;

②FileInputFormat输入数据格式的分配大小由数据块大小决定;

抽象类:CombineFileInputFormat

①可以使用CombineFileInputFormat来合并小文件;

②因为CombineFileInputFormat是一个抽象类,使用的时候需要创建一个CombineFileInputFormat的实体类,并且实现getRecordReader()的方法;

③避免文件分割的方法:

A.数据块大小尽可能大,这样使文件的大小小于数据块的大小,就不用进行分片;

B.继承FileInputFormat,并且重载isSplitable()方法;

(3)文本输入

类名:TextInputFormat

①TextInputFormat是默认的InputFormat,每一行数据就是一条记录;

②TextInputFormat的key是LongWritable类型的,存储该行在整个文件的偏移量,value是每行的数据内容,Text类型;

③输入分片与HDFS数据块关系:TextInputFormat每一条记录就是一行,很有可能某一行跨数据块存放;

类名:KeyValueInputFormat类

可以通过key为行号的方式来知道记录的行号,并且可以通过key.value.separator.in.input设置key与value的分割符;

类名:NLineInputFormat类

可以设置每个mapper处理的行数,可以通过mapred.line.input.format.lienspermap属性设置;

(4)二进制输入

类名:SequenceFileInputFormat

SequenceFileAsTextInputFormat

SequenceFileAsBinaryInputFormat

由于SequenceFile能够支持Splittable,所以能够作为mapreduce输入文件的格式,能够很方便的得到已经含有,value>的分片;

(5)多文件输入

类名:MultipleInputs

①MultipleInputs能够提供多个输入数据类型;

②通过addInputPath()方法来设置多路径;

(6)数据库格式输入

类名:DBInputFormat

①DBInputFormat是一个使用JDBC并且从关系型数据库中读取数据的一种输入格式;

②避免过多的数据库连接;

③HBase中的TableInputFormat可以让MapReduce程序访问HBase表里的数据;

实验部分:

新建项目TestMRInputFormat,新建包com.mr,导入相关依赖包

实验①,以SequenceFile作为输入,故预先运行SequenceFileWriter.java产生一个b.seq文件;

新建类:TestInputFormat1.java(基于WordCount.java修改):

package com.mr;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

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.output.FileOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;

public class TestInputFormat {

public static class TokenizerMapper

extends Mapper< IntWritable, Text, Text, IntWritable>{

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(IntWritable key, Text value, Context context

) throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

}

public static class IntSumReducer

extends Reducer {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable values,

Context context

) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable val : values) {

sum += val.get();

}

result.set(sum);

context.write(key, result);

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount ");

System.exit(2);

}

Job job = new Job(conf, "word count");

job.setJarByClass(TestInputFormat.class);

job.setMapperClass(TokenizerMapper.class);

job.setCombinerClass(IntSumReducer.class);

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

job.setInputFormatClass(SequenceFileInputFormat.class);//输入格式的设定

FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}

Eclipse中运行,参数配置如下图:

输出统计结果如下:

实验②,多种来源输入:

TestInputFormat2.java:

package com.mr;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

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.MultipleInputs;

import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;

public class TestInputFormat2 {

public static class Mapper1  //第一个mapper类

extends Mapper<<font color="#ed1c24">LongWritable, Text, Text, IntWritable>{

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(LongWritable key, Text value, Context context

) throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

}

public static class Mapper2 extends  //第二个mapper类

Mapper {

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(IntWritable key, Text value, Context context)

throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

}

public static class IntSumReducer

extends Reducer {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable values,

Context context

) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable val : values) {

sum += val.get();

}

result.set(sum);

context.write(key, result);

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = new Job(conf, "word count");

job.setJarByClass(TestInputFormat2.class);

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

Path path1 = new Path("/a.txt");

Path path2 = new Path("/b.seq");

//多输入

MultipleInputs.addInputPath(job, path1,TextInputFormat.class, Mapper1.class);

MultipleInputs.addInputPath(job, path2,SequenceFileInputFormat.class, Mapper2.class);

FileOutputFormat.setOutputPath(job, new Path("/output2"));

System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}

创建输入文本文件a.txt:

aaa bbb

ccc aaa

ddd eee

将项目打包为jar(不知道为什么eclipse中不能运行,还没找到原因,用jar命令可以运行):

File->Export->Runnable JAR file,命名jar文件为testMR.jar。

命令行中运行:

$hadoop jar testMR.jar com.mr.TestInputFormat2

输出统计结果如下:

二、输出格式

各种类关系结构图:

(1)文本输出

类名:TextOutputFormat

①默认的输出方式,key是LongWritable类型的,value是Text类型的;

②以“key \t value”的方式输出行;

(2)二进制输出

类名:SequenceFileOutputFormat

SequenceFileAsTextOutputFormat

SequenceFileAsBinaryOutputFormat

MapFileOutputFormat

(3)多文件输出

类名:MultipleOutputFormat

MultipleOutputs

区别:MultipleOutputs可以产生不同类型的输出;

(4)数据库输出

类名:DBOutputFormat

http://blog.sina.com.cn/s/blog_4438ac090101qfuh.html

时间: 2024-12-06 05:10:50

MapReduce类型与格式(输入与输出)的相关文章

CString中Format函数与格式输入与输出

CString中Format函数与格式输入与输出 Format是一个很常用,却又似乎很烦的方法,以下是它的完整概貌,以供大家查询之用: 格式化字符串forma("%d",12)意思是将一个整形的格式化的字符(我认为是保持其形状不变) 1).格式说明总是以%字符开始,以下是不同类型数据的格式方式%号后的说明: d输出带符号十进制数 o输出无符号八进制数 x输出无符号十六进制数 u输出无符号数 c输出单个字符 s输出一串字符 f输出实数(6位小数) e以指数形式输出实数 g选用f与e格式中

1.java.io包中定义了多个流类型来实现输入和输出功能,

1.java.io包中定义了多个流类型来实现输入和输出功能,可以从不同的角度对其进行分 类,按功能分为:(C),如果为读取的内容进行处理后再输出,需要使用下列哪种流?(G)   A.输入流和输出流 B.字节流和字符流 C.节点流和处理流   D.File stream E.Pipe stream F.Random stream G.Filter stream

MapReduce输入输出类型、格式及实例

输入格式 1.输入分片与记录 2.文件输入 3.文本输入 4.二进制输入 5.多文件输入 6.数据库格式输入 1.输入分片与记录 1.JobClient通过指定的输入文件的格式来生成数据分片InputSplit. 2.一个分片不是数据本身,而是可分片数据的引用. 3.InputFormat接口负责生成分片. InputFormat 负责处理MR的输入部分,有三个作用: 验证作业的输入是否规范. 把输入文件切分成InputSplit. 提供RecordReader 的实现类,把InputSplit

Hadoop 学习笔记一 ---MapReduce 的输入和输出

Hadoop 中的MapReduce库支持几种不同格式的输入数据.例如,文本模式的输入数据的每一行被视为一个key/value pair,其中key为文件的偏移量,value为那一行的内容.每一种输入类型的实现都必须能够把输入数据分割成数据片段,并能够由单独的Map任务来对数据片段进行后续处理. 一.  输入格式InputFormat 当运行一个M-R 作业的时候,我们需要为作业制定它的输入格式.InputFormat为Hadoop作业的所有输入格式的抽象基类,它描述了作业输入需要满足的规范细节

MapReduce的类型与格式

输入格式 输入分片与记录 之前讨论过,输入数据的每个分片对应一个map任务来处理 在MapReduce中输入分片被表示为InputSplit类,原型如下: public abstract class InputSplit{ //该分片的长度,用于排序分片,有限处理大分片 public abstract long getLength() throw IOException,InterruptedException; //该分片所在的存储位置(主机),不直接存储数据,而是指向数据的引用 public

[转]Spring MVC 学习笔记 json格式的输入和输出

Spring mvc处理json需要使用jackson的类库,因此为支持json格式的输入输出需要先修改pom.xml增加jackson包的引用 <!-- json --> <dependency> <groupId>org.codehaus.jackson</groupId> <artifactId>jackson-core-lgpl</artifactId> <version>1.8.1</version>

c#代码01--控制台的简单输入与输出及日期的格式输出

/* 使用ReadLine()完成控制台的输入输出内容 */ using System; namespace Test { class Test1 { static void Main(string[] args){ Console.Write("请输入:");//输出 请输入 且不换行 string read = Console.ReadLine();//将输入内容赋值到变量read Console.Write("您输入的内容为:");//控制台输出 您输入的内容

C++ Primer Plus 第17章 输入,输出和文件

第17章 输入.输出和文件 1.当到达输入语句时,他将刷新输出缓冲区中当前所有的输出 2.streambuf类 提供了对缓冲区的各种操作 ios_base类表示流的一般特征 ios类基于ios_base,包含了一个指向streambuf对象的指针 ostream类从ios派生,提供可输出方法 istream类从ios派生,提供可输入方法 iostream类基于ostream和istream,继承了输入和输出方法 3.包含iostream则自动创建了8个流对象 cin 标准输入 wcin 宽字符流

IO-04. 混合类型数据格式化输入(5)

IO-04. 混合类型数据格式化输入(5) 时间限制 400 ms 内存限制 65536 kB 代码长度限制 8000 B 判题程序 Standard 作者 乔林(清华大学) 本题要求编写程序,顺序读入浮点数1.整数.字符.浮点数2,再按照字符.整数.浮点数1.浮点数2的顺序输出. 输入格式: 输入在一行中顺序给出浮点数1.整数.字符.浮点数2,其间以1个空格分隔. 输出格式: 在一行中按照字符.整数.浮点数1.浮点数2的顺序输出,其中浮点数保留小数点后2位. 输入样例: 2.12 88 c 4