MR案例：输出/输入SequenceFile

SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。在SequenceFile文件中，每一个key-value对被看做是一条记录(Record)，基于Record的压缩策略，SequenceFile文件支持三种压缩类型：

NONE: 对records不进行压缩; (组合1)

RECORD: 仅压缩每一个record中的value值(不包括key); (组合2)

BLOCK: 将一个block中的所有records(包括key)压缩在一起;(组合3)

package test0820;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.VLongWritable;
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.SequenceFileOutputFormat;public class Test0829 {public static void main(String[] args) throws Exception {        Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(Test0829.class);job.setMapperClass(WCMapper.class);job.setReducerClass(WCReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(VLongWritable.class);        // 设置输出类job.setOutputFormatClass(SequenceFileOutputFormat.class);/*** 设置sequecnfile的格式，对于sequencefile的输出格式，有多种组合方式,* 从下面的模式中选择一种，并将其余的注释掉*/
// 组合方式1：不压缩模式
        SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.NONE);//组合方式2：record压缩模式，并指定采用的压缩方式 ：默认、gzip压缩等//        SequenceFileOutputFormat.setOutputCompressionType(job,//                CompressionType.RECORD);//        SequenceFileOutputFormat.setOutputCompressorClass(job,//                DefaultCodec.class);//组合方式3：block压缩模式，并指定采用的压缩方式 ：默认、gzip压缩等//        SequenceFileOutputFormat.setOutputCompressionType(job,//                CompressionType.BLOCK);//        SequenceFileOutputFormat.setOutputCompressorClass(job,//                DefaultCodec.class);
FileInputFormat.addInputPaths(job, args[0]);SequenceFileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}//mappublic static class WCMapper extendsMapper<LongWritable, Text, Text, VLongWritable> {public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {String[] split = value.toString().split(":",2);if(split.length!=1){String[] splited = split[1].split(",");for(String s : splited){context.write(new Text(s), new VLongWritable(1L));}            }}}//reducepublic static class WCReducer extends Reducer<Text, VLongWritable, Text, VLongWritable>{@Overrideprotected void reduce(Text key, Iterable<VLongWritable> v2s, Context context)throws IOException, InterruptedException {long sum=0;for(VLongWritable vl : v2s){sum += vl.get();                 }context.write(key, new VLongWritable(sum));}}
}

MR输入SequenceFile

当输入文件格式是SequenceFile的时候，要使用SequenceFileInputformat类。由于SequenceFile都是以key和value的二进制形式存放的(注意hadoop类型的二进制的解释方式和原始二进制不一样，会多一些维护信息)，所以在读取SequenceFile文件时必须预先知道key和value对应的hadoop类型。

对于上面代码产生的SequenceFile结果文件，以SequenceFileInputformat类进行读取。其中key为Text类型，value为VLongWritable类型。

package test0820;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.VLongWritable;
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.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class SFInput02 {public static void main(String[] args) throws Exception {Job job = Job.getInstance(new Configuration());job.setJarByClass(SFinput.class);job.setMapperClass(SFMapper.class);job.setReducerClass(SFReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(VLongWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(VLongWritable.class);job.setInputFormatClass(SequenceFileInputFormat.class);SequenceFileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}public static class SFMapper extends Mapper<Text, VLongWritable,Text, VLongWritable> {public void map(Text key, VLongWritable value, Context context)throws IOException, InterruptedException {context.write(key, value);}}//reducepublic static class SFReducer extends Reducer<Text, VLongWritable,Text, VLongWritable>{@Overrideprotected void reduce(Text key, Iterable<VLongWritable> v2s,Context context)throws IOException, InterruptedException {for(VLongWritable vl : v2s){context.write(key, vl);}}}
}

如若不清楚SequenceFile文件中key和value的类型，可以使用SequenceFileAsTextInputFormat类。它将SequenceFile的key和value都转化成Text对象传入map中。

package test0820;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.SequenceFileAsTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class SFinput {public static void main(String[] args) throws Exception {Job job = Job.getInstance(new Configuration());job.setJarByClass(SFinput.class);job.setMapperClass(SFMapper.class);job.setReducerClass(SFReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);job.setInputFormatClass(SequenceFileAsTextInputFormat.class);SequenceFileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}public static class SFMapper extends Mapper<Text, Text,Text, Text> {public void map(Text key, Text value, Context context)throws IOException, InterruptedException {context.write(key, value);}}//reducepublic static class SFReducer extends Reducer<Text, Text,Text,Text>{@Overrideprotected void reduce(Text key, Iterable<Text> v2s,Context context)throws IOException, InterruptedException {for(Text text : v2s){context.write(key, text);}}}
}

最后还有一种sequencefileAsBinaryInputFormat 类，它将SequenceFile中的key和value都以原始二进制的形式封装在byteswritable对象中传给map，如何对二进制数据进行解释是map函数编写者的工作。

转载于:https://www.cnblogs.com/skyl/p/4769542.html

MR案例：输出/输入SequenceFile相关推荐

大数据学习笔记25：MR案例——自定义输入输出格式处理个人成绩
文章目录一.提出任务二.自定义输入格式 1.输入格式(InputFormat) 2.记录读取器(RecordReader) 三.自定义输出格式 1.输出格式(OutputFormat) 2.记录写 ...
大数据学习笔记27：MR案例——多输入源处理成绩
文章目录一.提出任务原始成绩数据任务1.整合两个文件的成绩数据任务2.统计每个学生各科平均分二.准备工作 1.启动hadoop服务 2.上传数据文件到HDFS 3.创建Maven项目Mult ...
MR案例：Reduce-Join
问题描述:两种类型输入文件:address(地址)和company(公司)进行一对多的关联查询,得到地址名(例如:Beijing)与公司名(例如:Beijing JD.Beijing Red Star ...
hadoop之mr案例
mr案例 (1)创建maven项目 (2)在po,.xml添加下面代码 <dependencies><dependency><groupId>junit</g ...
解读：MR多路径输入
对于在一个MR-Job中使用多路径作为输入文件,一般有三种方法: 1).多次调用,加载不同路径: import org.apache.hadoop.mapreduce.lib.input.FileIn ...
MR案例：CombineFileInputFormat
CombineFileInputFormat是一个抽象类.Hadoop提供了两个实现类CombineTextInputFormat和CombineSequenceFileInputFormat. 此案 ...
MAT之SVM：SVM之分类预测根据已有大量数据集案例，输入已有病例的特征向量实现乳腺癌诊断高准确率预测
MAT之SVM:SVM之分类预测根据已有大量数据集案例,输入已有病例的特征向量实现乳腺癌诊断高准确率预测目录输出结果代码设计输出结果代码设计 load BreastTissue_data.m ...
Java练习案例之输入一个字符串统计每个字母出现次数
Java小案例:输入一个字符串,要求统计字符串中每个字符串出现的次数需求: 结果预览: 思路: 原码: 需求: 键盘录入一个字符串,要求统计字符串中每个字符串出现的次数. 举例:键盘录入" ...
输入一组整数,0结束输入,之后输出输入的最大的和最小的整数.【思路】
package com.ykmimi.new1; /*** 输入一组整数,0结束输入,之后输出输入的最大的和最小的整数.*/ import java.util.Scanner;public class ...

MR案例：输出/输入SequenceFile

MR案例：输出/输入SequenceFile相关推荐

最新文章

热门文章