一、MapReduce案例-流量統計
1： 需求一: 統計求和
統計每個手機號的上行數據包總和，下行數據包總和，上行總流量之和，下行總流量之和 分析：以手機號碼作為key值，上行流量，下行流量，上行總流量，下行總流量四個字段作為value值，然后以這個key，和value作為map階段的輸出，reduce階段的輸入
1.1: 自定義map的輸出value對象FlowBean
package flowcount; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * @author MoooJL * @data 2020/8/28-20:26 */
public class FlowBean implements Writable { private Integer upFlow; //上行數據包數
private Integer downFlow; //下行數據包數
private Integer upCountFlow; //上行流量總和
private Integer downCountFlow;//下行流量總和
public Integer getUpFlow() { return upFlow; } public void setUpFlow(Integer upFlow) { this.upFlow = upFlow; } public Integer getDownFlow() { return downFlow; } public void setDownFlow(Integer downFlow) { this.downFlow = downFlow; } public Integer getUpCountFlow() { return upCountFlow; } public void setUpCountFlow(Integer upCountFlow) { this.upCountFlow = upCountFlow; } public Integer getDownCountFlow() { return downCountFlow; } public void setDownCountFlow(Integer downCountFlow) { this.downCountFlow = downCountFlow; } @Override public String toString() { return upFlow +
"\t" + downFlow +
"\t" + upCountFlow +
"\t" + downCountFlow; } //序列化方法
@Override public void write(DataOutput out) throws IOException { out.writeInt(upFlow); out.writeInt(downFlow); out.writeInt(upCountFlow); out.writeInt(downCountFlow); } @Override public void readFields(DataInput in) throws IOException { this.upFlow = in.readInt(); this.downFlow = in.readInt(); this.upCountFlow = in.readInt(); this.downCountFlow = in.readInt(); } }
1.2：定義FlowMapper類
package flowcount; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.yarn.webapp.hamlet.Hamlet; import java.io.IOException; /** * @author MoooJL * @data 2020/8/28-20:31 */
public class FlowCountMapper extends Mapper<LongWritable, Text,Text,FlowBean> { /* 將K1和V1轉為K2和V2: K1 V1 0 1360021750219 128 1177 16852 200 ------------------------------ K2 V2 13600217502 FlowBean(19 128 1177 16852) */ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //1:拆分行文本數據,得到手機號--->K2
String[] split = value.toString().split("\t"); String phoneNum = split[1]; //2:創建FlowBean對象,並從行文本數據拆分出流量的四個四段,並將四個流量字段的值賦給FlowBean對象
FlowBean flowBean = new FlowBean(); flowBean.setUpFlow(Integer.parseInt(split[6])); flowBean.setDownFlow(Integer.parseInt(split[7])); flowBean.setUpCountFlow(Integer.parseInt(split[8])); flowBean.setDownCountFlow(Integer.parseInt(split[9])); //3:將K2和V2寫入上下文中
context.write(new Text(phoneNum), flowBean); } }
1.3：定義FlowReducer類
package flowcount; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; /** * @author MoooJL * @data 2020/8/28-21:54 */
public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> { @Override protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { //1:遍歷集合,並將集合中的對應的四個字段累計
Integer upFlow = 0; //上行數據包數
Integer downFlow = 0; //下行數據包數
Integer upCountFlow = 0; //上行流量總和
Integer downCountFlow = 0;//下行流量總和
for (FlowBean value : values) { upFlow += value.getUpFlow(); downFlow += value.getDownFlow(); upCountFlow += value.getUpCountFlow(); downCountFlow += value.getDownCountFlow(); } //2:創建FlowBean對象,並給對象賦值 V3
FlowBean flowBean = new FlowBean(); flowBean.setUpFlow(upFlow); flowBean.setDownFlow(downFlow); flowBean.setUpCountFlow(upCountFlow); flowBean.setDownCountFlow(downCountFlow); //3:將K3和V3下入上下文中
context.write(key, flowBean); } }
1.4：程序main函數入口FlowMain
package flowcount; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; 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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; /** * @author MoooJL * @data 2020/8/26-15:24 */
public class jobMain extends Configured implements Tool { //該方法用於指定一個job任務
@Override public int run(String[] args) throws Exception { //1、創建一個job對象
Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount"); //2、配置job任務的8個對象 //第一步 指定文件的讀取方式和路徑
job.setInputFormatClass(TextInputFormat.class); /* 如果打包運行出錯 加配置 job.setJarByClass(jobMain.class); */ TextInputFormat.addInputPath(job,new Path("file:///D:\\input\\flowcount")); //第二步 指定map階段的處理方式和數據類型
job.setMapperClass(FlowCountMapper.class); //設置map階段k2的類型
job.setMapOutputKeyClass(Text.class); //設置map階段v2的類型
job.setMapOutputValueClass(FlowBean.class); //第三(分區) 四(排序) 第五步(規約) 六(分組) 步 采用默認方式 //第七步 指定reduce階段的處理方式和數據類型
job.setReducerClass(FlowCountReducer.class); //設置k3 v3類型
job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); //第八步 設置輸出類型
job.setOutputFormatClass(TextOutputFormat.class); //本地運行模式
TextOutputFormat.setOutputPath(job,new Path("file:///D:\\output\\flowcount")); //等待任務結束
boolean b = job.waitForCompletion(true); return b ? 0:1; } public static void main(String[] args) throws Exception { Configuration configuration=new Configuration(); //啟動job任務
int run = ToolRunner.run(configuration, new jobMain(), args); System.exit(run); } }
1.5：運行截圖
2：需求二: 上行流量倒序排序（遞減排序）
分析，以需求一的輸出數據作為排序的輸入數據，自定義FlowBean,以FlowBean為map輸出的key，以手機號作為Map輸出的value，因為MapReduce程序會對Map階段輸出的key進行排序
2.1： 定義FlowBean實現WritableComparable實現比較排序
Java 的 compareTo 方法說明:
- compareTo 方法用於將當前對象與方法的參數進行比較。
- 如果指定的數與參數相等返回 0。
- 如果指定的數小於參數返回 -1。
- 如果指定的數大於參數返回 1。
例如：o1.compareTo(o2); 返回正數的話，當前對象（調用 compareTo 方法的對象 o1）要排在比較對象（compareTo 傳參對象 o2）后面，返回負數的話，放在前面
package flowcount.sort; import org.apache.hadoop.io.Writable; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * @author MoooJL * @data 2020/8/28-20:26 */
public class FlowBean implements WritableComparable<FlowBean> { private Integer upFlow; //上行數據包數
private Integer downFlow; //下行數據包數
private Integer upCountFlow; //上行流量總和
private Integer downCountFlow;//下行流量總和
public Integer getUpFlow() { return upFlow; } public void setUpFlow(Integer upFlow) { this.upFlow = upFlow; } public Integer getDownFlow() { return downFlow; } public void setDownFlow(Integer downFlow) { this.downFlow = downFlow; } public Integer getUpCountFlow() { return upCountFlow; } public void setUpCountFlow(Integer upCountFlow) { this.upCountFlow = upCountFlow; } public Integer getDownCountFlow() { return downCountFlow; } public void setDownCountFlow(Integer downCountFlow) { this.downCountFlow = downCountFlow; } @Override public String toString() { return upFlow +
"\t" + downFlow +
"\t" + upCountFlow +
"\t" + downCountFlow; } //序列化方法
@Override public void write(DataOutput out) throws IOException { out.writeInt(upFlow); out.writeInt(downFlow); out.writeInt(upCountFlow); out.writeInt(downCountFlow); } @Override public void readFields(DataInput in) throws IOException { this.upFlow = in.readInt(); this.downFlow = in.readInt(); this.upCountFlow = in.readInt(); this.downCountFlow = in.readInt(); } //指定排序規則
@Override public int compareTo(FlowBean flowBean) { // return this.upFlow.compareTo(flowBean.getUpFlow()) * -1;
return flowBean.upFlow - this.upFlow ; } }
2.2：定義FlowMapper
package flowcount.sort; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; /** * @author MoooJL * @data 2020/8/29-0:04 */
public class FlowSortMapper extends Mapper<LongWritable, Text,FlowBean,Text> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //1:拆分行文本數據(V1),得到四個流量字段,並封裝FlowBean對象---->K2
String[] split = value.toString().split("\t"); FlowBean flowBean = new FlowBean(); flowBean.setUpFlow(Integer.parseInt(split[1])); flowBean.setDownFlow(Integer.parseInt(split[2])); flowBean.setUpCountFlow(Integer.parseInt(split[3])); flowBean.setDownCountFlow(Integer.parseInt(split[4])); //2:通過行文本數據,得到手機號--->V2
String phoneNum = split[0]; //3:將K2和V2下入上下文中
context.write(flowBean, new Text(phoneNum)); } }
2.3:定義FlowReducer
package flowcount.sort; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; /** * @author MoooJL * @data 2020/8/29-0:10 */
public class FlowSortReducer extends Reducer<FlowBean, Text,Text,FlowBean> { @Override protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException { //1:遍歷集合,取出 K3,並將K3和V3寫入上下文中
for (Text value : values) { context.write(value, key); } } }
2.4:程序main函數入口
package flowcount.sort; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; 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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import sort.SortBean; /** * @author MoooJL * @data 2020/8/26-15:24 */
public class jobMain extends Configured implements Tool { //該方法用於指定一個job任務
@Override public int run(String[] args) throws Exception { //1、創建一個job對象
Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount_cort"); //2、配置job任務的8個對象 //第一步 指定文件的讀取方式和路徑
job.setInputFormatClass(TextInputFormat.class); /* 如果打包運行出錯 加配置 job.setJarByClass(jobMain.class); */ TextInputFormat.addInputPath(job,new Path("file:///D:\\output\\flowcount")); //第二步 指定map階段的處理方式和數據類型
job.setMapperClass(FlowSortMapper.class); //設置map階段k2的類型
job.setMapOutputKeyClass(FlowBean.class); //設置map階段v2的類型
job.setMapOutputValueClass(Text.class); //第三(分區) 四(排序) 第五步(規約) 六(分組) 步 采用默認方式 //第七步 指定reduce階段的處理方式和數據類型
job.setReducerClass(FlowSortReducer.class); //設置k3 v3類型
job.setOutputKeyClass(Text.class); job.setOutputValueClass(SortBean.class); //第八步 設置輸出類型
job.setOutputFormatClass(TextOutputFormat.class); //本地運行模式
TextOutputFormat.setOutputPath(job,new Path("file:///D:\\output\\flowcount_sort")); //等待任務結束
boolean b = job.waitForCompletion(true); return b ? 0:1; } public static void main(String[] args) throws Exception { Configuration configuration=new Configuration(); //啟動job任務
int run = ToolRunner.run(configuration, new jobMain(), args); System.exit(run); } }
2.5：運行截圖
3：需求三: 手機號碼分區
在需求一的基礎上，繼續完善，將不同的手機號分到不同的數據文件的當中去，需要自定義分區來實現，這里我們自定義來模擬分區，將以下數字開頭的手機號進行分開
135 開頭數據到一個分區文件
136 開頭數據到一個分區文件
137 開頭數據到一個分區文件
其他分區
3.1：自定義分區
package flowcount.partition; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; /** * @author MoooJL * @data 2020/8/29-0:28 */
public class FlowCountPartition extends Partitioner<Text,FlowBean> { /* 該方法用來指定分區的規則: 135 開頭數據到一個分區文件 136 開頭數據到一個分區文件 137 開頭數據到一個分區文件 其他分區 參數: text : K2 手機號 flowBean: V2 i : ReduceTask的個數 */ @Override public int getPartition(Text text, FlowBean flowBean, int i) { //1:獲取手機號
String phoneNum = text.toString(); //2:判斷手機號以什么開頭,返回對應的分區編號(0-3)
if(phoneNum.startsWith("135")){ return 0; }else if(phoneNum.startsWith("136")){ return 1; }else if(phoneNum.startsWith("137")){ return 2; }else{ return 3; } } }
3.2：作業運行設置
job.setPartitionerClass(FlowPartition.class); job.setNumReduceTasks(4);
3.3：運行結果
