文档章节

hadoop计数器、分区、序列化等

Zero零_度
 Zero零_度
发布于 2015/01/07 09:57
字数 936
阅读 130
收藏 0

#程序员薪资揭榜#你做程序员几年了?月薪多少?发量还在么?>>>

package com.test;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/*
 * 手机号码  流量[类型1、类型2、类型3]
 * 13500001234 12,56,78
 * 18600001235 32,21,80
 * 15800001235 16,33,56
 * 13500001234 19,92,73
 * 18600001235 53,55,29
 * 18600001239 27,77,68
 * 
 * 计算得出
 * 手机号 类型1汇总 类型2汇总 类型3汇总
 */
public class WordCount extends Configured implements Tool {
 
 public static class Map extends Mapper<LongWritable, Text, Text, StreamWritable> {
  //避免每调用一次map就创建一次对象
  private final Text phoneNum = new Text();
  private final StreamWritable streamWritable = new StreamWritable();
  
  private String firstLine = "#_#";
  private String lastLine;
  
  public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
   String line = value.toString();
   
   //获得map输入的第一条记录
   if("#_#".equals(firstLine)) {
    firstLine = key.toString() + "\t" + line;
   }
   
   //获得map输入的最后一条记录
   lastLine = key.toString() + "\t" + line;
   
   //13500001234手机号码总共在多少行出现【自定义计数器】
   Counter helloCounter = (Counter) context.getCounter("Words", "13500001234");
   if(line.contains("13500001234")) {
    helloCounter.increment(1L);
   }
   
   String[] strs = line.split("\t");
   //手机号码
   phoneNum.set(strs[0]);
   
   //流量
   String[] stream = strs[1].split(",");
   streamWritable.set(Long.parseLong(stream[0]), Long.parseLong(stream[1]), Long.parseLong(stream[2]));
   
   context.write(phoneNum, streamWritable);
  }
  
  protected void cleanup(org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,StreamWritable>.Context context) throws IOException ,InterruptedException {
   //获得map输入的第一条记录
   System.out.println(firstLine);
   
   //获得map输出的最后一条记录
   System.out.println(lastLine);
  };
 }
 
 public static class Reduce extends Reducer<Text, StreamWritable, Text, StreamWritable> {
  //避免每调用一次reduce就创建一次对象
  private StreamWritable streamWritable = new StreamWritable();
  
  /*
   * map函数执行结束后,map输出的<k, v>一共有4个,分别是<hello, 1><you, 1>,<hello, 1>,<me, 1>
   * 分区,默认只有一个分区  job.setPartitionerClass
   * 排序 <hello, 1>,<hello, 1>,<me, 1><you, 1>
   * 分组 把相同key的value放到一个集合中 <hello, {1,1}><me, {1}><you, {1}>,每一组调用一次reduce函数
   * 归约(可选) job.setCombinerClass
   */
  public void reduce(Text key, Iterable<StreamWritable> values, Context context) throws IOException, InterruptedException {
   long stream1 = 0;
   long stream2 = 0;
   long stream3 = 0;
   
   Iterator<StreamWritable> it = values.iterator();
   while(it.hasNext()) {
    streamWritable = it.next();
    stream1 = stream1 + streamWritable.getStream1();
    stream2 = stream2 + streamWritable.getStream2();
    stream3 = stream3 + streamWritable.getStream3();
   }
   
   streamWritable.set(stream1, stream2, stream3);
   context.write(key, streamWritable);
  }
 }
 
 public int run(String[] args) throws Exception {
  Configuration conf = this.getConf();
  Job job = new Job(conf);
  job.setJarByClass(WordCount.class);
  job.setJobName(WordCount.class.getSimpleName());
  
  FileInputFormat.addInputPath(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
  //如果没有配置,默认值是1
  job.setNumReduceTasks(1);
  
  //指定map产生的数据按照什么规则分配到不同的reduce中,如果没有配置,默认是HashPartitioner.class
  job.setPartitionerClass(MyPartitioner.class);
  
  //FileInputFormat.getSplits决定map任务数量,XxxInputFormat.RecordReader处理每一个split,得到map输入的key、value
  //默认是TextInputFormat
  job.setInputFormatClass(TextInputFormat.class);
  job.setOutputFormatClass(TextOutputFormat.class);
  
  job.setMapperClass(Map.class);
  job.setCombinerClass(Reduce.class);
  job.setReducerClass(Reduce.class);
  
  //当reduce输出类型与map输出类型一致时,map的输出类型可以不设置
  job.setMapOutputKeyClass(Text.class);
  job.setMapOutputValueClass(StreamWritable.class);
  
  //reduce的输出类型一定要设置
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(StreamWritable.class);
  
  job.waitForCompletion(true);
  
  return job.isSuccessful()?0:1;
 }
 
 public static void main(String[] args) throws Exception {
  int exit = ToolRunner.run(new WordCount(), args);
  System.exit(exit);
 }
 
}
//自定义Partitioner
class MyPartitioner extends Partitioner<Text, StreamWritable> {
 @Override
 //返回值表示,分配到第几个reduce任务中
 public int getPartition(Text key, StreamWritable value, int numPartitions) {
  //13500001234手机号码分到第1个reduce,其余的分到第二个reduce
  if("13500001234".equals(key.toString())) {
   return 0;
  } else {
   return 1;
  }
 }
}
//自定义序列化类[处理手机流量]
//Serializable:Java序列化的信息非常臃肿,比如存在层层类继承的时候,继承关系序列化出去,还需要序列化回来。
//hadoop的Writable轻量很多
class StreamWritable implements Writable {
 private long stream1;
 
 private long stream2;
 
 private long stream3;
 
 public long getStream1() {
  return stream1;
 }
 public void setStream1(long stream1) {
  this.stream1 = stream1;
 }
 public long getStream2() {
  return stream2;
 }
 public void setStream2(long stream2) {
  this.stream2 = stream2;
 }
 public long getStream3() {
  return stream3;
 }
 public void setStream3(long stream3) {
  this.stream3 = stream3;
 }
 public StreamWritable() {
  
 }
 
 public StreamWritable(long stream1, long stream2, long stream3) {
  this.set(stream1, stream2, stream3);
 }
 
 public void set(long stream1, long stream2, long stream3) {
  this.stream1 = stream1;
  this.stream2 = stream2;
  this.stream3 = stream3;
 }
 
 @Override
 public void write(DataOutput out) throws IOException {
  out.writeLong(stream1);//写出顺序和读入顺序一一对应
  out.writeLong(stream2);
  out.writeLong(stream3);
 }
 @Override
 public void readFields(DataInput in) throws IOException {
  this.stream1 = in.readLong();//写出顺序和读入顺序一一对应
  this.stream2 = in.readLong();
  this.stream3 = in.readLong();
 }
 
 //输出的时候会调用toString方法
 @Override
 public String toString() {
  return this.stream1+"\t"+this.stream2+"\t"+this.stream3;
 }
}

 

© 著作权归作者所有

上一篇: 插入排序
Zero零_度
粉丝 70
博文 1325
码字总数 283013
作品 0
程序员
私信 提问
加载中

评论(1)

猪兜兜
猪兜兜
13好东西
Hadoop大数据开发基础系列:五、MapReduce进阶编程

五、MapReduce进阶编程 目录: 1.筛选日志文件并生成序列化文件 2.Hadoop Java API读取序列化日志文件 3.优化日志文件统计程序 4.Eclipse提交日志文件统计程序 5.小结 6.实训 7.小练习 任务背...

osc_aglv22cb
2019/10/17
2
0
Apache Spark 官方文档 翻译 - 编程指南

最近用Apache Spark 处理一些大数据,学了spark官方英文文档,顺便翻译了方便学习。 spark 版本 2.2.0. 翻译官方文档原地址: http://spark.apache.org/docs/latest/rdd-programming-guide....

___k先生
2017/12/07
0
0
Hadoop(六)MapReduce的入门与运行原理

一 MapReduce入门 1.1 MapReduce定义 Mapreduce是一个分布式运算程序的编程框架,是用户开发“基于hadoop的数据分析应用”的核心框架; Mapreduce核心功能是将用户编写的业务逻辑代码和自带默...

osc_0zs17uxd
2018/05/26
2
0
Hadoop——Hive的序列化,文件存储,分桶和分区

一、序列化   1.正则表达式 创建表 //在加载文件时可以通过正则表达式来区分字段,字段名必须和文件中的字段名一致create table reg_table( ) row format serde 'org.apache.hadoop.hive.se...

osc_38e42cd3
2019/09/06
6
0
Spark编程指南V1.4.0(翻译)

Spark编程指南V1.4.0 · 简单介绍 · 接入Spark · Spark初始化 · 使用Shell · 在集群上部署代码 · 弹性分布式数据集 · 并行集合(Parallelized Collections) · 其它数据集 · RDD的操作...

技术mix呢
2017/10/12
0
0

没有更多内容

加载失败,请刷新页面

加载更多

文献速递20200524

一 文献题目:Characterizing the Causal Pathway for Genetic Variants Associated with Neurological Phenotypes Using Human Brain-Derived Proteome Data 不想看英文题目:使用人脑蛋白质......

osc_mbhfa1vl
29分钟前
9
0
【Java入门】JDK安装和环境变量配置(Win7版)

系统环境:Windows7 x64 安装JDK和JRE版本:1.8.0_191 1、下载JDK安装包 Oracle官网下载网址:https://www.oracle.com/technetwork/java/javase/downloads/index.html 选择需要下载的JDK版本...

osc_s7aj86hu
30分钟前
11
0
Android Q 指纹解锁流程

Android Q 指纹解锁流程 // Authentation Finger Schedule: (close Screen, Authentation with finger(Success))        // PhoneWindowManager        mKeyguardDelegate.onStar......

osc_0qnrwmy3
32分钟前
12
0
List的扩容机制,你真的明白吗?

一:背景 1. 讲故事 在前一篇大内存排查中,我们看到了Dictionary正在做扩容操作,当时这个字典的count=251w,你把字典玩的66飞起,其实都是底层为你负重前行,比如其中的扩容机制,当你遇到...

osc_34b9n45c
34分钟前
14
0
2020.04.12软件更新公告

原创软件区升级ComicsViewer、DjVuToy、Pdg2Pic、PdgRenamer、PdgThumbViewer

osc_6tgtqi6v
34分钟前
23
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部