mapreduce程序 ：
描述： 提取出二进制文件里的短链，并统计排序（由大到小排序）
思路：
HotlinkCount：
1 /*
2 * Copyright (c) EEFUNG 2014 All Rights Reserved
3 *
4 * 统计每个连接出现的频率
5 *
6 */
7 package com.eefung.hstore.mr.job.hotLink;
8
9 import cn.antvision.eagleattack.model.Status;
10 import com.eefung.hstore.help.HstoreConstants;
11 import com.eefung.hstore.help.AvroUtils;
12 import com.eefung.hstore.mr.input.TimeFramePathFilter;
13 import org.apache.hadoop.conf.Configuration;
14 import org.apache.hadoop.conf.Configured;
15 import org.apache.hadoop.fs.Path;
16 import org.apache.hadoop.io.BytesWritable;
17 import org.apache.hadoop.io.IntWritable;
18 import org.apache.hadoop.io.LongWritable;
19 import org.apache.hadoop.io.Text;
20 import org.apache.hadoop.mapreduce.Job;
21 import org.apache.hadoop.mapreduce.Mapper;
22 import org.apache.hadoop.mapreduce.Reducer;
23 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
24 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
25 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
26 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
27 import org.apache.hadoop.util.Tool;
28 import org.apache.hadoop.util.ToolRunner;
29 import org.slf4j.Logger;
30 import org.slf4j.LoggerFactory;
31 import java.io.IOException;
32 import java.util.regex.Matcher;
33 import java.util.regex.Pattern;
34
35 public class HotLinkCount extends Configured implements Tool {
36
37 private static final Logger logger = LoggerFactory
38 .getLogger(HotLinkCount.class);
39
40 public static void main(String[] args) throws Exception {
41 System.exit(ToolRunner.run(new Configuration(), new HotLinkCount(), args));
42 }
43
44 @Override
45 public int run(String[] args) throws Exception {
46 if (args.length != 3) {
47 logger.warn("wrong args!");
48 return 1;
49 }
50 String input = args[0].trim();
51 String output = args[1].trim();
52 int reducerNum = Integer.parseInt(args[2].trim());
53 Job job = Job.getInstance(getConf());
54 job.setJobName("HotLink Count");
55 job.setJarByClass(HotLinkCount.class);
56 job.setMapperClass(InnerMapper.class);
57 job.setReducerClass(InnerReducer.class);
58
59 job.setInputFormatClass(SequenceFileInputFormat.class);
60 job.setOutputFormatClass(TextOutputFormat.class);
61
62 job.setMapOutputKeyClass(Text.class);
63 job.setMapOutputValueClass(LongWritable.class);
64 job.setOutputKeyClass(LongWritable.class);
65 job.setOutputValueClass(Text.class);
66
67 job.setNumReduceTasks(reducerNum);
68 FileInputFormat.setInputPathFilter(job, TimeFramePathFilter.class);///
69
70
71 String[] sources = input.split(HstoreConstants.DELIMITER_COMMA);
72 for (String source : sources) {
73 FileInputFormat.addInputPath(job, new Path(source));
74 }
75
76 FileOutputFormat.setOutputPath(job, new Path(output));
77 return job.waitForCompletion(true) ? 0 : 1;
78 }
79
80 public static class InnerMapper
81 extends Mapper<LongWritable, BytesWritable, Text, LongWritable>
82 implements HstoreConstants {
83
84 private final static LongWritable one = new LongWritable(1);
85 private Text word = new Text();
86
87 Pattern pattern = Pattern
88 .compile("https?:\\/\\/[-A-Za-z0-9+&#%?=~_|!:,.;]+\\/[A-Za-z0-9+&#%=~_]{4,}");
89
90 protected void map(LongWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
91 Status status = null;
92 try {
93 status = AvroUtils.convertStatusAvroFromBytes(value.getBytes()); //
94
95
96 } catch (Exception e) {
97 e.printStackTrace();
98 }
99 if (status == null) {
100 return;
101 }
102 if (status.getContent() == null) {
103 return;
104 }
105 String sb = status.getContent().toString();
106 Matcher matcher = pattern.matcher(sb);
107 while (matcher.find()) {
108 word.set(matcher.group(0));
109 context.write(word, one);
110 }
111 }
112 }
113
114 public static class InnerReducer extends
115 Reducer<Text, LongWritable ,LongWritable,Text> implements HstoreConstants {
116
117 @Override
118 protected void reduce(Text key, Iterable<LongWritable> values,
119 Context context) throws IOException, InterruptedException {
120 Long sum = 0L;
121 for (LongWritable value : values) {
122 sum += value.get();
123 }
124 if (key.toString().length() == 0) {
125 return;
126 }
127 context.write( new LongWritable(sum),key);
128 }
129 //TimeFramePathFilter
130 }
131 }
hotlinkSort
1 package com.eefung.hstore.mr.job.hotLink;
2 import java.io.IOException;
3 import java.util.Iterator;
4
5 import com.eefung.hstore.help.HstoreConstants;
6 import org.apache.hadoop.conf.Configuration;
7 import org.apache.hadoop.conf.Configured;
8 import org.apache.hadoop.fs.Path;
9 import org.apache.hadoop.io.*;
10 import org.apache.hadoop.mapreduce.Job;
11 import org.apache.hadoop.mapreduce.Mapper;
12 import org.apache.hadoop.mapreduce.Reducer;
13 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
14 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
15 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
16 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
17 import org.apache.hadoop.util.Tool;
18 import org.apache.hadoop.util.ToolRunner;
19 import org.slf4j.Logger;
20 import org.slf4j.LoggerFactory;
21
22 public class HotLinkSort extends Configured implements Tool {
23
24 private static final Logger logger = LoggerFactory
25 .getLogger(HotLinkSort.class);
26
27
28
29 public static void main(String[] args) throws Exception {
30 System.exit(ToolRunner.run(new Configuration(), new HotLinkSort(), args));
31 }
32
33 @Override
34 public int run(String[] args) throws Exception {
35 if (args.length != 3) {
36 logger.warn("wrong args!");
37 return 1;
38 }
39 String input = args[0].trim();
40 String output = args[1].trim();
41 int reducerNum = Integer.parseInt(args[2].trim());
42 Job job = Job.getInstance(getConf());
43 job.setJobName("HotLinkSort");
44 job.setJarByClass(HotLinkSort.class);
45
46 job.setMapperClass(InnerMapper.class);
47 job.setReducerClass(InnerReducer.class);
48 job.setSortComparatorClass(InnerComparator.class);
49
50 job.setInputFormatClass(KeyValueTextInputFormat.class);
51 job.setOutputFormatClass(TextOutputFormat.class);
52
53 job.setMapOutputKeyClass(LongWritable.class);
54 job.setMapOutputValueClass(Text.class);
55 job.setOutputKeyClass(Text.class);
56 job.setOutputValueClass(LongWritable.class);
57
58 job.setNumReduceTasks(reducerNum);
59 String[] path = input.split(",");
60 for (int i = 0; i < path.length; i++) {
61 FileInputFormat.addInputPath(job, new Path(path[i]));
62 }
63 FileOutputFormat.setOutputPath(job, new Path(output));
64
65 return job.waitForCompletion(true) ? 0 : 1;
66 }
67
68 public static class InnerMapper
69 extends Mapper<Text,Text, LongWritable,Text>
70 implements HstoreConstants {
71
72
73 LongWritable key_out;
74 long key_long;
75 String key_str;
76
77 protected void map(Text key, Text value, org.apache.hadoop.mapreduce.Mapper<Text, Text, LongWritable,Text>.Context context) throws java.io.IOException, InterruptedException {
78
79 try {
80 key_str=key.toString();
81 logger.info("@@@@@@@@@@@@@@ value 是："+key_str);
82 key_long=Long.parseLong(key_str.trim());
83 key_out=new LongWritable(key_long);
84
85 context.write(key_out, value);
86 } catch (NumberFormatException e) {
87 e.printStackTrace();
88 } catch (IOException e) {
89 e.printStackTrace();
90 } catch (InterruptedException e) {
91 e.printStackTrace();
92 }
93
94 }
95 }
96
97 public static class InnerReducer
98 extends Reducer<LongWritable,Text,Text,LongWritable> implements
99 HstoreConstants {
100 @Override
101 protected void setup(Context context) throws IOException, InterruptedException {
102 super.setup(context);
103 // logger.info("@@@@@@@@@@@@@@@@@@@@@@@@@");
104 }
105
106 public void reduce(LongWritable key, Iterable<Text> values,
107 Context context
108 ) throws IOException, InterruptedException {
109
110 Iterator ite = values.iterator();
111 while(ite.hasNext())
112 {
113 String record = ite.next().toString();
114 Text temp=new Text();
115 temp.set(record);
116 context.write(temp, key);
117 }
118
119 }
120 }
121
122 public static class InnerComparator extends WritableComparator {
123
124 protected InnerComparator(){
125 super(LongWritable.class, true);
126 }
127
128 public int compare(WritableComparable a, WritableComparable b)
129 {
130 LongWritable aa=(LongWritable)a;
131 LongWritable bb=(LongWritable)b;
132 return -aa.compareTo(bb);
133 }
134 }
135 }
本来还有个短链转长链的过程：
1 package com.eefung.hstore.mr.job.hotLink;
2 import java.io.BufferedReader;
3 import java.io.FileReader;
4 import java.io.IOException;
5 import java.text.SimpleDateFormat;
6 import java.util.Date;
7 import java.util.Iterator;
8 import org.apache.http.HttpHost;
9 import org.apache.http.HttpResponse;
10 import org.apache.http.client.HttpClient;
11 import org.apache.http.client.methods.HttpGet;
12 import org.apache.http.client.methods.HttpUriRequest;
13 import org.apache.http.impl.client.DefaultHttpClient;
14 import org.apache.http.params.HttpConnectionParams;
15 import org.apache.http.protocol.BasicHttpContext;
16 import org.apache.http.protocol.ExecutionContext;
17 import org.apache.http.protocol.HttpContext;
18 import org.apache.http.util.EntityUtils;
19 import com.eefung.hstore.help.HstoreConstants;
20 import org.apache.hadoop.conf.Configuration;
21 import org.apache.hadoop.conf.Configured;
22 import org.apache.hadoop.fs.Path;
23 import org.apache.hadoop.io.*;
24 import org.apache.hadoop.mapreduce.Job;
25 import org.apache.hadoop.mapreduce.Mapper;
26 import org.apache.hadoop.mapreduce.Reducer;
27 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
28 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
29 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
30 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
31 import org.apache.hadoop.util.Tool;
32 import org.apache.hadoop.util.ToolRunner;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
35
36 public class HotLinkRedirector extends Configured implements Tool {
37
38 private static Redirector redirector = new Redirector();
39
40 private static final Logger logger = LoggerFactory
41 .getLogger(HotLinkRedirector.class);
42
43
44
45 public static void main(String[] args) throws Exception {
46 System.exit(ToolRunner.run(new Configuration(), new HotLinkRedirector(), args));
47
48 /*redirector = new Redirector();
49 BufferedReader reader = new BufferedReader(new FileReader("E:\\temp.txt"));
50 String shortLine = null;
51
52 SimpleDateFormat df1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
53 df1.format(new Date());// new Date()为获取当前系统时间
54
55
56 while (true) {
57 shortLine = reader.readLine();
58 if (shortLine == null)
59 break;
60
61 String url = redirector.getRedirectInfo(shortLine.trim());
62 // System.out.println("短链是： " + shortLine + " 长链是：" + url);
63
64 }
65 SimpleDateFormat df2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
66 df2.format(new Date());
67 System.out.println("开始 "+df1);
68 System.out.println("结束 "+df2);*/
69
70 }
71
72 @Override
73 public int run(String[] args) throws Exception {
74 if (args.length != 3) {
75 logger.warn("wrong args!");
76 return 1;
77 }
78 int reducerNum = Integer.parseInt(args[2].trim());
79 Job job = Job.getInstance(getConf());
80 job.setJobName("HotLink Redirector");
81
82 job.setJarByClass(HotLinkRedirector.class);
83 job.setMapperClass(InnerMapper.class);
84
85 job.setInputFormatClass(KeyValueTextInputFormat.class);
86 job.setOutputFormatClass(TextOutputFormat.class);
87
88 job.setMapOutputKeyClass(Text.class);
89 job.setMapOutputValueClass(LongWritable.class);
90 job.setOutputKeyClass(Text.class);
91 job.setOutputValueClass(LongWritable.class);
92
93 job.setNumReduceTasks(reducerNum);
94
95 FileInputFormat.addInputPath(job, new Path(args[0]));
96 FileOutputFormat.setOutputPath(job, new Path(args[1]));
97
98 return job.waitForCompletion(true) ? 0 : 1;
99 }
100
101 public static class InnerMapper
102 extends Mapper<Text,Text,Text,LongWritable>
103 implements HstoreConstants {
104 Text textOut = new Text();
105 LongWritable valueOut;
106
107
108 @Override
109 protected void cleanup(Context context) throws IOException, InterruptedException {
110 redirector.release();
111
112 }
113
114 protected void map(Text key, Text value, org.apache.hadoop.mapreduce.Mapper<Text, Text, Text, LongWritable>.Context context) throws java.io.IOException, InterruptedException {
115 String shortLink=key.toString();
116 String longLink=redirector.getRedirectInfo(shortLink);
117 if(longLink==null)
118 longLink=shortLink;
119 textOut.set(longLink);
120
121 String temp=value.toString();
122 long a=Long.parseLong(temp);
123 valueOut=new LongWritable(a);
124 context.write(textOut,valueOut);
125 }
126 }
127
128
129
130 private static class Redirector {
131 private HttpClient httpClient; // 发送HTTP请求和从由URI识别的资源接收HTTP响应。
132
133 public Redirector() {
134 httpClient = new DefaultHttpClient();
135 httpClient.getParams().setParameter(HttpConnectionParams.CONNECTION_TIMEOUT, 1000);
136 httpClient.getParams().setParameter(HttpConnectionParams.SO_TIMEOUT, 1000);
137
138 }
139
140 public void release() {
141 httpClient.getConnectionManager().shutdown();
142 }
143
144 public String getRedirectInfo(String shortLink) {
145 HttpResponse response=null;
146 try {
147 HttpGet httpGet = new HttpGet(shortLink); // 用于发送 HTTP GET 请求及接收 HTTP
148 HttpContext httpContext = new BasicHttpContext(); // 封装了一个独立的HTTP请求的特定的信息。
149 response = httpClient.execute(httpGet, httpContext);
150
151
152 HttpHost targetHost = (HttpHost) httpContext
153 .getAttribute(ExecutionContext.HTTP_TARGET_HOST);
154
155
156 HttpUriRequest realRequest = (HttpUriRequest) httpContext
157 .getAttribute(ExecutionContext.HTTP_REQUEST);
158
159 return targetHost.toString()
160 + realRequest.getURI().toString();
161
162 } catch (Throwable e) {
163 return null;
164 }finally {
165 try {
166 if (response != null) {
167 EntityUtils.consume(response.getEntity());
168 }
169 } catch (IOException e) {
170 e.printStackTrace();
171 }
172 }
173 }
174 }
175 }
但是这个转换的过程太慢了，数据量大的时候 放在哪个阶段在哪个阶段就kill掉了。
oozie 调度：
学习的时候主要是看的 http://shiyanjun.cn/archives/tag/oozie 这个博客
描述：分别跑sina，twitter，sohu，tencent，netease几个平台是数据，（先执行hotlinkCount 其结果用于hotlinkSort的输入）。
编写代码：在工程里的workflow目录下建hotlink目录，其下编写以下几个文件：
1. coordinator.xml:
1 <?xml version="1.0" encoding="UTF-8"?>
2 <coordinator-app name="hotlink" frequency="${coord:days(1)}"
3 start="${start}" end="${end}" timezone="UTC" xmlns="uri:oozie:coordinator:0.1">
4 <action>
5 <workflow>
6 <app-path>${root}/workflow.xml</app-path>
7 </workflow>
8 </action>
9 </coordinator-app>
2. job.properties:
1 nn=hdfs://hdt01:8020
2 rm=hdt01:8050
3
4 root=/eefung/yangzf/oozie/jobs/hotlink
5 start=2015-01-21T13:00Z
6 end=2020-01-01T00:00Z
7
8 dest=/hotlink
9
10 #oozie.wf.application.path=${nn}${root}
11
12 oozie.coord.application.path=${nn}${root}
13 oozie.libpath=${nn}${root}/lib
14
15
16 cbaseColumnFamily=content
17
18 beginDays=29
19 endDays=26
3. workflow.xml:
1 <?xml version="1.0" encoding="utf-8"?>
2 <workflow-app xmlns=‘uri:oozie:workflow:0.4‘ name=‘hotlink‘>
3
4 <start to="sina"/>
5
6 <action name="sina">
7 <sub-workflow>
8 <app-path>${root}/workflow-hotlink-mr.xml</app-path>
9 <propagate-configuration/>
10 <configuration>
11 <property>
12 <name>platform</name>
13 <value>sina</value>
14 </property>
15
16 </configuration>
17 </sub-workflow>
18 <ok to="twitter"/>
19 <error to="twitter"/>
20 </action>
21
22 <action name="twitter">
23 <sub-workflow>
24 <app-path>${root}/workflow-hotlink-mr.xml</app-path>
25 <propagate-configuration/>
26 <configuration>
27 <property>
28 <name>platform</name>
29 <value>twitter</value>
30 </property>
31
32 </configuration>
33 </sub-workflow>
34 <ok to="tencent"/>
35 <error to="tencent"/>
36 </action>
37
38 <action name="tencent">
39 <sub-workflow>
40 <app-path>${root}/workflow-hotlink-mr.xml</app-path>
41 <propagate-configuration/>
42 <configuration>
43 <property>
44 <name>platform</name>
45 <value>tencent</value>
46 </property>
47
48 </configuration>
49 </sub-workflow>
50 <ok to="netease"/>
51 <error to="netease"/>
52 </action>
53
54 <action name="netease">
55 <sub-workflow>
56 <app-path>${root}/workflow-hotlink-mr.xml</app-path>
57 <propagate-configuration/>
58 <configuration>
59 <property>
60 <name>platform</name>
61 <value>netease</value>
62 </property>
63
64 </configuration>
65 </sub-workflow>
66 <ok to="sohu"/>
67 <error to="sohu"/>
68 </action>
69
70 <action name="sohu">
71 <sub-workflow>
72 <app-path>${root}/workflow-hotlink-mr.xml</app-path>
73 <propagate-configuration/>
74 <configuration>
75 <property>
76 <name>platform</name>
77 <value>sohu</value>
78 </property>
79
80 </configuration>
81 </sub-workflow>
82 <ok to="end"/>
83 <error to="fail"/>
84 </action>
85
86 <kill name="fail">
87 <message>Failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
88 </kill>
89
90 <end name="end"/>
91
92 </workflow-app>
4. workflow-hotlink-mr:
1 <?xml version="1.0" encoding="utf-8"?>
2 <workflow-app xmlns=‘uri:oozie:workflow:0.4‘ name=‘hotlink‘>
3
4 <start to="count" />
5
6 <action name="count">
7 <map-reduce>
8 <job-tracker>${rm}</job-tracker>
9 <name-node>${nn}</name-node>
10 <prepare>
11 <delete path="${nn}${root}/output/${platform}/count"/>
12 </prepare>
13 <configuration>
14 <property>
15 <name>mapred.mapper.new-api</name>
16 <value>true</value>
17 </property>
18 <property>
19 <name>mapred.reducer.new-api</name>
20 <value>true</value>
21 </property>
22 <property>
23 <name>mapreduce.map.class</name>
24 <value>com.eefung.hstore.mr.job.hotLink.HotLinkCount$InnerMapper</value>
25 </property>
26 <property>
27 <name>mapreduce.reduce.class</name>
28 <value>com.eefung.hstore.mr.job.hotLink.HotLinkCount$InnerReducer</value>
29 </property>
30 <property>
31 <name>mapreduce.inputformat.class</name>
32 <value>org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat</value>
33 </property>
34 <property>
35 <name>mapreduce.outputformat.class</name>
36 <value>org.apache.hadoop.mapreduce.lib.output.TextOutputFormat</value>
37 </property>
38 <property>
39 <name>mapred.mapoutput.key.class</name>
40 <value>org.apache.hadoop.io.Text</value>
41 </property>
42 <property>
43 <name>mapred.mapoutput.value.class</name>
44 <value>org.apache.hadoop.io.LongWritable</value>
45 </property>
46 <property>
47 <name>mapred.output.key.class</name>
48 <value>org.apache.hadoop.io.LongWritable</value>
49 </property>
50 <property>
51 <name>mapred.output.value.class</name>
52 <value>org.apache.hadoop.io.Text</value>
53 </property>
54 <property>
55 <name>mapreduce.job.reduces</name>
56 <value>12</value>
57 </property>
58
59 <property>
60 <name>mapred.input.pathFilter.class</name>
61 <value>com.eefung.hstore.mr.input.TimeFramePathFilter</value>
62 </property>
63
64 <property>
65 <name>source.time.frame.begin</name>
66 <value>${beginDays}</value>
67 </property>
68 <property>
69 <name>source.time.frame.end</name>
70 <value>${endDays}</value>
71 </property>
72
73 <property>
74 <name>mapred.input.dir</name>
75 <value>${root}/input/${platform}/*</value>
76 </property>
77
78 <property>
79 <name>mapred.output.dir</name>
80 <value>${root}/output/${platform}/count</value>
81 </property>
82 </configuration>
83 </map-reduce>
84 <ok to="sort"/>
85 <error to="fail"/>
86 </action>
87
88 <action name="sort">
89 <map-reduce>
90 <job-tracker>${rm}</job-tracker>
91 <name-node>${nn}</name-node>
92 <prepare>
93 <delete path="${nn}${root}/output/${platform}/sort"/>
94 </prepare>
95 <configuration>
96 <property>
97 <name>mapred.mapper.new-api</name>
98 <value>true</value>
99 </property>
100 <property>
101 <name>mapred.reducer.new-api</name>
102 <value>true</value>
103 </property>
104
105 <property>
106 <name>mapred.input.dir</name>
107 <value>${root}/output/${platform}/count/*</value>
108 </property>
109 <property>
110 <name>mapred.output.dir</name>
111 <value>${root}/output/${platform}/sort</value>
112 </property>
113
114 <property>
115 <name>mapreduce.map.class</name>
116 <value>com.eefung.hstore.mr.job.hotLink.HotLinkSort$InnerMapper</value>
117 </property>
118 <property>
119 <name>mapreduce.reduce.class</name>
120 <value>com.eefung.hstore.mr.job.hotLink.HotLinkSort$InnerReducer</value>
121 </property>
122
123 <property>
124 <name>mapreduce.compare.class</name>
125 <value>com.eefung.hstore.mr.job.hotLink.HotLinkSort$InnerComparator</value>
126 </property>
127 <property>
128 <name>mapreduce.inputformat.class</name>
129 <value>org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat</value>
130 </property>
131 <property>
132 <name>mapreduce.outputformat.class</name>
133 <value>org.apache.hadoop.mapreduce.lib.output.TextOutputFormat</value>
134 </property>
135 <property>
136 <name>mapred.mapoutput.key.class</name>
137 <value>org.apache.hadoop.io.LongWritable</value>
138 </property>
139 <property>
140 <name>mapred.mapoutput.value.class</name>
141 <value>org.apache.hadoop.io.Text</value>
142 </property>
143 <property>
144 <name>mapred.output.key.class</name>
145 <value>org.apache.hadoop.io.Text</value>
146 </property>
147 <property>
148 <name>mapred.output.value.class</name>
149 <value>org.apache.hadoop.io.LongWritable</value>
150 </property>
151 <property>
152 <name>mapreduce.job.reduces</name>
153 <value>1</value>
154 </property>
155
156
157 </configuration>
158
159 </map-reduce>
160 <ok to="end"/>
161 <error to="fail"/>
162 </action>
163
164
165
166 <kill name="fail">
167 <message>Failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
168 </kill>
169 <end name="end"/>
170 </workflow-app>
运行：
1.
时间： 2024-08-28 10:54:00