定義FlowReducer類

package cn.learn.mapreduce_flowcount;import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> {@Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {//封裝新的FlowBeanFlowBean flowBean = new FlowBean();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();}flowBean.setUpFlow(upFlow);flowBean.setDownFlow(downFlow);flowBean.setUpCountFlow(upCountFlow);flowBean.setDownCountFlow(downCountFlow);//將K3和V3寫入上下文中context.write(key, flowBean);} }

程序main函數入口FlowMain

package cn.learn.mapreduce_flowcount;import cn.learn.mapreduce_sort.FlowCountSortReducer; 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;public class JobMain extends Configured implements Tool {@Overridepublic int run(String[] strings) throws Exception {//創建一個任務對象Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount");//打包放在集群運行時，需要做一個配置job.setJarByClass(JobMain.class);//第一步:設置讀取文件的類: K1 和V1job.setInputFormatClass(TextInputFormat.class);TextInputFormat.addInputPath(job, new Path("hdfs://node01:8020/input/flowcount"));//第二步：設置Mapper類job.setMapperClass(FlowCountMapper.class);//設置Map階段的輸出類型: k2 和V2的類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//第三,四，五，六步采用默認方式(分區，排序，規約，分組)//第七步 ：設置文的Reducer類job.setReducerClass(FlowCountReducer.class);//設置Reduce階段的輸出類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//設置Reduce的個數//第八步:設置輸出類job.setOutputFormatClass(TextOutputFormat.class);//設置輸出的路徑TextOutputFormat.setOutputPath(job, new Path("hdfs://node01:8020/out/flowcount_out"));boolean b = job.waitForCompletion(true);return b?0:1;}public static void main(String[] args) throws Exception {Configuration configuration = new Configuration();//啟動一個任務int run = ToolRunner.run(configuration, new JobMain(), args);System.exit(run);}}

?

總結

以上是生活随笔為你收集整理的MapReduce-流量统计求和-Reducer和JobMain代码编写的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。