array(2) { ["docs"]=> array(10) { [0]=> array(10) { ["id"]=> string(3) "428" ["text"]=> string(77) "Visual Studio 2017 单独启动MSDN帮助（Microsoft Help Viewer）的方法" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(8) "DonetRen" ["tagsname"]=> string(55) "Visual Studio 2017|MSDN帮助|C#程序|.NET|Help Viewer" ["tagsid"]=> string(23) "[401,402,403,"300",404]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400964" ["_id"]=> string(3) "428" } [1]=> array(10) { ["id"]=> string(3) "427" ["text"]=> string(42) "npm -v；报错 cannot find module "wrapp"" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(4) "zzty" ["tagsname"]=> string(50) "node.js|npm|cannot find module "wrapp“|node" ["tagsid"]=> string(19) "[398,"239",399,400]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400760" ["_id"]=> string(3) "427" } [2]=> array(10) { ["id"]=> string(3) "426" ["text"]=> string(54) "说说css中pt、px、em、rem都扮演了什么角色" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(12) "zhengqiaoyin" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400640" ["_id"]=> string(3) "426" } [3]=> array(10) { ["id"]=> string(3) "425" ["text"]=> string(83) "深入学习JS执行--创建执行上下文（变量对象，作用域链，this）" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "Ry-yuan" ["tagsname"]=> string(33) "Javascript|Javascript执行过程" ["tagsid"]=> string(13) "["169","191"]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511399901" ["_id"]=> string(3) "425" } [4]=> array(10) { ["id"]=> string(3) "424" ["text"]=> string(30) "C# 排序技术研究与对比" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(9) "vveiliang" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(8) ".Net Dev" ["catesid"]=> string(5) "[199]" ["createtime"]=> string(10) "1511399150" ["_id"]=> string(3) "424" } [5]=> array(10) { ["id"]=> string(3) "423" ["text"]=> string(72) "【算法】小白的算法笔记：快速排序算法的编码和优化" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(9) "penghuwan" ["tagsname"]=> string(6) "算法" ["tagsid"]=> string(7) "["344"]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511398109" ["_id"]=> string(3) "423" } [6]=> array(10) { ["id"]=> string(3) "422" ["text"]=> string(64) "JavaScript数据可视化编程学习（二）Flotr2，雷达图" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "chengxs" ["tagsname"]=> string(28) "数据可视化|前端学习" ["tagsid"]=> string(9) "[396,397]" ["catesname"]=> string(18) "前端基本知识" ["catesid"]=> string(5) "[198]" ["createtime"]=> string(10) "1511397800" ["_id"]=> string(3) "422" } [7]=> array(10) { ["id"]=> string(3) "421" ["text"]=> string(36) "C#表达式目录树（Expression）" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(4) "wwym" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(4) ".NET" ["catesid"]=> string(7) "["119"]" ["createtime"]=> string(10) "1511397474" ["_id"]=> string(3) "421" } [8]=> array(10) { ["id"]=> string(3) "420" ["text"]=> string(47) "数据结构 队列_队列实例：事件处理" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "idreamo" ["tagsname"]=> string(40) "C语言|数据结构|队列|事件处理" ["tagsid"]=> string(23) "["246","247","248",395]" ["catesname"]=> string(12) "数据结构" ["catesid"]=> string(7) "["133"]" ["createtime"]=> string(10) "1511397279" ["_id"]=> string(3) "420" } [9]=> array(10) { ["id"]=> string(3) "419" ["text"]=> string(47) "久等了,博客园官方Android客户端发布" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(3) "cmt" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511396549" ["_id"]=> string(3) "419" } } ["count"]=> int(200) } 222 MapReduce的自定义结果文件名OutputFormat - 爱码网

MapReduce的自定义结果文件名OutputFormat

  1 package com.mengyao.hadoop.mapreduce;
  2 
  3 import java.io.IOException;
  4 import java.net.URI;
  5 import java.net.URISyntaxException;
  6 import java.text.SimpleDateFormat;
  7 import java.util.Date;
  8 
  9 import org.apache.hadoop.fs.FSDataOutputStream;
 10 import org.apache.hadoop.fs.FileSystem;
 11 import org.apache.hadoop.fs.Path;
 12 import org.apache.hadoop.io.LongWritable;
 13 import org.apache.hadoop.io.Text;
 14 import org.apache.hadoop.mapreduce.Job;
 15 import org.apache.hadoop.mapreduce.JobContext;
 16 import org.apache.hadoop.mapreduce.Mapper;
 17 import org.apache.hadoop.mapreduce.OutputCommitter;
 18 import org.apache.hadoop.mapreduce.OutputFormat;
 19 import org.apache.hadoop.mapreduce.Partitioner;
 20 import org.apache.hadoop.mapreduce.RecordWriter;
 21 import org.apache.hadoop.mapreduce.Reducer;
 22 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 23 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 24 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
 25 import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
 26 import org.apache.hadoop.util.Tool;
 27 import org.apache.hadoop.util.ToolRunner;
 28 import org.apache.hadoop.conf.Configuration;
 29 import org.apache.hadoop.conf.Configured;
 30 
 31 public class JobOutputRenameApp extends Configured implements Tool {
 32 
 33     
 34     static class JobOutputRenameRecordWriter extends RecordWriter<Text, LongWritable> {
 35 
 36         FSDataOutputStream out;
 37         
 38         public JobOutputRenameRecordWriter(){
 39             
 40         }
 41         
 42         public JobOutputRenameRecordWriter(FSDataOutputStream out){
 43             this.out = out;
 44         }
 45         
 46         @Override
 47         public void write(Text key, LongWritable value) throws IOException,
 48                 InterruptedException {
 49             out.writeUTF(key.toString());
 50             out.writeUTF("\t");
 51             out.writeLong(value.get());
 52             out.writeUTF("\n");
 53         }
 54 
 55         @Override
 56         public void close(TaskAttemptContext context) throws IOException,
 57                 InterruptedException {
 58             out.close();
 59         }
 60         
 61     }
 62     
 63     static class JobOutputRenameOutputFormat extends OutputFormat<Text, LongWritable> {
 64         @Override
 65         public RecordWriter<Text, LongWritable> getRecordWriter(
 66                 TaskAttemptContext context) throws IOException,
 67                 InterruptedException {
 68             final String output_dir = context.getConfiguration().get("output_dir");
 69             final String output_name = context.getConfiguration().get("output_name");
 70             FSDataOutputStream out = null;
 71             try {
 72                 FileSystem fs = FileSystem.get(new URI(output_dir), context.getConfiguration(), "root");
 73                 out = fs.create(new Path(output_dir+output_name), true);
 74             } catch (IllegalArgumentException e) {
 75                 e.printStackTrace();
 76             } catch (URISyntaxException e) {
 77                 e.printStackTrace();
 78             }
 79             
 80             return new JobOutputRenameRecordWriter(out);
 81         }
 82 
 83         @Override
 84         public void checkOutputSpecs(JobContext context) throws IOException,
 85                 InterruptedException {
 86         }
 87 
 88         @Override
 89         public OutputCommitter getOutputCommitter(TaskAttemptContext context)
 90                 throws IOException, InterruptedException {
 91             final String output_dir = context.getConfiguration().get("output_dir");
 92             return new FileOutputCommitter(new Path(output_dir), context);
 93         }
 94         
 95     }
 96     
 97     static class JobOutputRenameMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
 98         
 99         private Text k = null;
100         private LongWritable v = null;
101         
102         @Override
103         protected void setup(
104                 Mapper<LongWritable, Text, Text, LongWritable>.Context context)
105                 throws IOException, InterruptedException {
106             k = new Text();
107             v = new LongWritable(1L);
108         }
109 
110         @Override
111         protected void map(LongWritable key, Text value, Context context)
112                 throws IOException, InterruptedException {
113             String[] words = value.toString().split("\t");
114             for (String word : words) {
115                 k.set(word);
116                 context.write(k, v);
117             }
118         }
119         
120     }
121     
122     static class JobOutputRenamePartition extends Partitioner<Text, LongWritable> {
123 
124         @Override
125         public int getPartition(Text key, LongWritable value, int numPartitions) {
126             if (!key.toString().equals("hello")) {
127                 return 1;
128             }
129             return 0;
130         }
131         
132     }
133     
134     static class JobOutputRenameReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
135         @Override
136         protected void reduce(Text key, Iterable<LongWritable> value,
137                 Reducer<Text, LongWritable, Text, LongWritable>.Context context)
138                 throws IOException, InterruptedException {
139             long count = 0L;
140             for (LongWritable item : value) {
141                 count += item.get();
142             }
143             context.write(key, new LongWritable(count));
144         }
145     }
146     
147     @Override
148     public int run(String[] args) throws Exception {
149         Configuration conf = getConf();
150         conf.set("output_dir", args[1]);
151         conf.set("output_name", args[2]);
152         Job job = Job.getInstance(conf, JobOutputRenameApp.class.getSimpleName());
153         job.setJarByClass(JobOutputRenameApp.class);
154         
155         FileInputFormat.addInputPath(job, new Path(args[0]));
156         job.setInputFormatClass(TextInputFormat.class);
157         job.setMapperClass(JobOutputRenameMapper.class);
158         job.setMapOutputKeyClass(Text.class);
159         job.setMapOutputValueClass(LongWritable.class);
160         
161 //        job.setPartitionerClass(JobOutputRenamePartition.class);
162 //        job.setNumReduceTasks(2);
163         
164         job.setReducerClass(JobOutputRenameReducer.class);
165         job.setOutputKeyClass(Text.class);
166         job.setOutputValueClass(LongWritable.class);
167         job.setOutputFormatClass(JobOutputRenameOutputFormat.class);
168         
169         return job.waitForCompletion(true)?0:1;
170     }
171     
172     public static int createJob(String[] params){
173         Configuration conf = new Configuration();
174         int status = 1;
175         try {
176             status = ToolRunner.run(conf, new JobOutputRenameApp(), params);
177         } catch (Exception e) {
178             e.printStackTrace();
179             new RuntimeException(e);
180         }
181         
182         return status;
183     }
184 
185     public static void main(String[] args) throws Exception {
186         args = new String[3];
187         args[0] = "/testdata/words";
188         args[1] = "/job/mapreduce/"+JobOutputRenameApp.class.getSimpleName()+"_"+new SimpleDateFormat("yyyyMMddhhMMss").format(new Date())+"/";
189         args[2] = "wordcount.txt";
190         
191 //        String[] otherArgs = new GenericOptionsParser(args).getRemainingArgs();
192         if (args != null && args.length !=3) {
193             System.out.println("Usage: "+JobOutputRenameApp.class.getSimpleName()+" <input dir> <output dir> <output name> ");
194             System.exit(3);
195         } else {
196             int status = createJob(args);
197             System.exit(status);
198         }
199     }
200 }

 

相关文章：

猜你喜欢
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode