1.資料源(虛構) ：手機號 ip 網址 上行流量 下行流量 狀態
2.要求 ：根據手機號磁區并計算其上下行流量之和，每個區以手機號 上流量 下流量 流量之和 格式輸出
3.java代碼
（1）Diver
package com.oracle.flowbean;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class FlowBeanDrivers {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//封裝輸出輸入路徑
args =new String[]{"C:/Users/input","C:/Users/output"};
System.setProperty("hadoop.home.dir","E:/hadoop-2.7.2/");
//1獲取job物件
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
//2設定jar加載路徑
job.setJarByClass(FlowBeanDrivers.class);
//3關聯mapper和reducer 及其他功能類
job.setMapperClass(FlowBeanMapper.class);
job.setReducerClass(FlowBeanReduce.class);
//3.1關聯自定義磁區類及磁區個數
job.setPartitionerClass(FlowBeanPartitioner.class);
//job.setNumReduceTasks(5);
//4設定map輸出的key和value
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);
//5設定最終輸出的key和value
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
//6設定輸入路徑
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));
//7提交job
boolean result=job.waitForCompletion(true);
System.exit(result? 0 : 1);
}
}
（2）重寫序列化類
package com.oracle.flowbean;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
* 1實作writable介面 重寫序列化和反序列化方法
* 2根據業務需求，設計類中的屬性
* 3生成setter和getter方法
* 4生成空參構造 ，給反序列化用
* 5生成gotring（0 自定義輸出格式
* 6實作序列和反序列方法
*/
public class FlowBean implements Writable {
private long upFlow;
private long downflow;
private long sumFlow;
//空參構造
public FlowBean() {
}
@Override
public String toString() {
return upFlow +"\t"+downflow +
"\t" + sumFlow ;
}
//序列化方法
public void write(DataOutput out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downflow);
out.writeLong(sumFlow);
}
//反序列化方法
public void readFields(DataInput in) throws IOException {
upFlow=in.readLong();
downflow=in.readLong();
sumFlow=in.readLong();
}
public long getUpFlow() {
return upFlow;
}
public void setUpFlow(long upFlow) {
this.upFlow = upFlow;
}
public long getDownflow() {
return downflow;
}
public void setDownflow(long downflow) {
this.downflow = downflow;
}
public long getSumFlow() {
return sumFlow;
}
public void setSumFlow(long sumFlow) {
this.sumFlow = sumFlow;
}
}
（3）mapper類
package com.oracle.flowbean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
/**
* 統計每個手機號上下行流量，及總流量
*/
public class FlowBeanMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
Text k = new Text();
FlowBean flowBean = new FlowBean();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//1獲取一行資料
String line = value.toString();
//2切割一行資料
String[] fields = line.split("\t");
//3封裝一個物件
//3.1封裝key到陣列
k.set(fields[1]);
//3.2封裝value
//封裝上行流量
long upFlow = Long.parseLong( fields[fields.length-3]);
//封裝下行流量
long downFlow = Long.parseLong( fields[fields.length-2]);
flowBean.setUpFlow(upFlow);
flowBean.setDownflow(downFlow);
//4寫出
context.write(k,flowBean);
}
}
（4）reduce類
package com.oracle.flowbean;
import com.oracle.flowbean.FlowBean;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class FlowBeanReduce extends Reducer<Text,FlowBean,Text,FlowBean> {
FlowBean v = new FlowBean();
@Override
protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
//累加求和
//總流量累加，上下行流量累加
//上行流量求和
long sum_upFlow = 0;
//下行流量求和
long sum_downFlow = 0;
for (FlowBean flowBean : values) {
sum_upFlow+=flowBean.getUpFlow();
sum_downFlow+=flowBean.getDownflow();
}
v.setUpFlow(sum_upFlow);
v.setDownflow(sum_downFlow);
v.setSumFlow(sum_downFlow+sum_upFlow);
//輸出
context.write(key,v);
}
}
（5）磁區類
package com.oracle.flowbean;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class FlowBeanPartitioner extends Partitioner<Text,FlowBean> {
public int getPartition(Text key, FlowBean value, int i) {
int partition=4;
//獲得手機號
String phonenumber = key.toString().substring(0, 3);
//根據phnenumber進行判斷磁區
if ("136".equals(phonenumber)){
partition=0;
}else if ("137".equals(phonenumber)){
partition=1;
}
else if ("138".equals(phonenumber)){
partition=2;
}else if ("139".equals(phonenumber)){
partition=3;
}
return partition;
}
}
4.小結
（1）默認情況不主動磁區
//job.setNumReduceTasks(*);
找到原始碼可見，默認get到一個區
public int getNumReduceTasks() {
return this.getInt("mapreduce.job.reduces", 1);
}
（2）設定磁區數大于磁區類設定的個數
大于設定的磁區數，超出的磁區為空，
（3）設定磁區數小于磁區類設定的個數
小于設定的磁區數，error，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/187926.html
標籤：Java
下一篇：vue 頁面跳轉的兩種方式