远程调用执行Hadoop Map/Reduce

港控/mmm° 2022-09-17 11:27 209阅读 0赞

### [远程调用执行Hadoop Map/Reduce][Hadoop Map_Reduce] ###

在Web项目中，由用户下发任务后，后台服务器远程调用JobTracker所在服务器，运行Map/Reduce更符合B/S架构的习惯。

由于网上没有相关资料，所以自己实现了一个，现在分享一下。

注：基于Hadoop1.1.2版本

**转发请注明地址：http://sgq0085.iteye.com/admin/blogs/1879442**

一个常见的WordCount如下：

Java代码   ![收藏代码][icon_star.png]

1.  **package** com.gqshao.hadoop.remote;  
2.    
3.  **import** java.io.IOException;  
4.  **import** java.util.\*;  
5.    
6.  **import** org.apache.hadoop.conf.\*;  
7.  **import** org.apache.hadoop.fs.Path;  
8.  **import** org.apache.hadoop.io.\*;  
9.  **import** org.apache.hadoop.mapreduce.\*;  
10. **import** org.apache.hadoop.mapreduce.lib.input.\*;  
11. **import** org.apache.hadoop.mapreduce.lib.output.\*;  
12. **import** org.apache.hadoop.util.\*;  
13.   
14. **public** **class** WordCount **extends** Configured **implements** Tool \{  
15.     **public** **static** **class** Map **extends** Mapper<LongWritable, Text, Text, IntWritable> \{  
16.         **private** **final** **static** IntWritable one = **new** IntWritable(1);  
17.         **private** Text word = **new** Text();  
18.   
19.         **public** **void** map(LongWritable key, Text value, Context context) **throws** IOException, InterruptedException \{  
20.             String line = value.toString();  
21.             StringTokenizer tokenizer = **new** StringTokenizer(line);  
22.             **while** (tokenizer.hasMoreTokens()) \{  
23.                 word.set(tokenizer.nextToken());  
24.                 context.write(word, one);  
25.             \}  
26.         \}  
27.     \}  
28.   
29.     **public** **static** **class** Reduce **extends** Reducer<Text, IntWritable, Text, IntWritable> \{  
30.         **public** **void** reduce(Text key, Iterable<IntWritable> values, Context context) **throws** IOException, InterruptedException \{  
31.             **int** sum = 0;  
32.             **for** (IntWritable val : values) \{  
33.                 sum += val.get();  
34.             \}  
35.             context.write(key, **new** IntWritable(sum));  
36.         \}  
37.     \}  
38.   
39.     **public** **int** run(String\[\] args) **throws** Exception \{  
40.         **this**.getClass().getResource("/hadoop/");  
41.         Configuration conf = getConf();  
42.         Job job = **new** Job(conf);  
43.         conf.set("mapred.job.tracker", "192.168.0.128:9001");  
44.         conf.set("fs.default.name", "hdfs://192.168.0.128:9000");  
45.         conf.set("hadoop.job.ugi", "hadoop");  
46.         conf.set("Hadoop.tmp.dir", "/user/gqshao/temp/");  
47.   
48.         job.setJarByClass(WordCount.**class**);  
49.         job.setJobName("wordcount");  
50.   
51.         job.setOutputKeyClass(Text.**class**);  
52.         job.setOutputValueClass(IntWritable.**class**);  
53.   
54.         job.setMapperClass(Map.**class**);  
55.         job.setReducerClass(Reduce.**class**);  
56.   
57.         job.setInputFormatClass(TextInputFormat.**class**);  
58.         job.setOutputFormatClass(TextOutputFormat.**class**);  
59.         String hdfs = "hdfs://192.168.0.128:9000";  
60.         args = **new** String\[\] \{ hdfs + "/user/gqshao/input/big", hdfs + "/user/gqshao/output/WordCount/" + **new** Date().getTime() \};  
61.         FileInputFormat.setInputPaths(job, **new** Path(args\[0\]));  
62.         FileOutputFormat.setOutputPath(job, **new** Path(args\[1\]));  
63.         **boolean** success = job.waitForCompletion(**true**);  
64.         **return** success ? 0 : 1;  
65.     \}  
66.   
67.     **public** **static** **void** main(String\[\] args) **throws** Exception \{  
68.         **int** ret = ToolRunner.run(**new** WordCount(), args);  
69.         System.exit(ret);  
70.     \}  
71. \}

在这里输入和输出目录都是指向HDFS上的，但实际运行的时候(一般 -Xms128m -Xmx512m -XX:MaxPermSize=128M)发现输出中有如下信息：

Java代码   ![收藏代码][icon_star.png]

1.  信息: Running job: job\_local\_0001

证明该Map/Reduce程序运行在Local中。也就是说，这种方式只能提前打好Jar包，放到Cluster服务器上，在通过Jar运行。

**转发请注明地址：http://sgq0085.iteye.com/admin/blogs/1879442**

如何远程运行Map/Reduce程序，经研究发现两点。

1.需要将Hadoop的配置文件加载到当前进程的ClassLoader中，或将配置文件放到/bin目录下。

通过跟踪 job.waitForCompletion(true);→submit();→info = jobClient.submitJobInternal(conf);→status = jobSubmitClient.submitJob(jobId, submitJobDir.toString(), jobCopy.getCredentials());

发现private JobSubmissionProtocol jobSubmitClient;分别有两个实现

在org.apache.hadoop.mapred.JobClient中init()方法中可以看到如果设置了conf中如果设置了mapred.job.tracker则在Hadoop Cluster中运行，否则是Local

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** init(JobConf conf) **throws** IOException \{  
2.    String tracker = conf.get("mapred.job.tracker", "local");  
3.    tasklogtimeout = conf.getInt(  
4.      TASKLOG\_PULL\_TIMEOUT\_KEY, DEFAULT\_TASKLOG\_TIMEOUT);  
5.    **this**.ugi = UserGroupInformation.getCurrentUser();  
6.    **if** ("local".equals(tracker)) \{  
7.      conf.setNumMapTasks(1);  
8.      **this**.jobSubmitClient = **new** LocalJobRunner(conf);  
9.    \} **else** \{  
10.     **this**.rpcJobSubmitClient =   
11.         createRPCProxy(JobTracker.getAddress(conf), conf);  
12.     **this**.jobSubmitClient = createProxy(**this**.rpcJobSubmitClient, conf);  
13.   \}          
14. \}

所以需要在运行时加载某目录下配置文件

方法如下：

Java代码   ![收藏代码][icon_star.png]

1.  /\*\* 
2.   \* 加载配置文件 
3.   \*/  
4.  **public** **static** **void** setConf(Class<?> clazz, Thread thread, String path) \{  
5.      URL url = clazz.getResource(path);  
6.      **try** \{  
7.          File confDir = **new** File(url.toURI());  
8.          **if** (!confDir.exists()) \{  
9.              **return**;  
10.         \}  
11.         URL key = confDir.getCanonicalFile().toURI().toURL();  
12.         ClassLoader classLoader = thread.getContextClassLoader();  
13.         classLoader = **new** URLClassLoader(**new** URL\[\] \{ key \}, classLoader);  
14.         thread.setContextClassLoader(classLoader);  
15.     \} **catch** (Exception e) \{  
16.         e.printStackTrace();  
17.     \}  
18. \}

2.设置运行时Jar包

继续看jobClient.submitJobInternal(conf);可以发现client在提交作业到Hadoop时需要把作业打包成jar，然后copy到fs的submitJarFile路径中。所以必须指定conf中的运行的Jar包。

方法如下：

Java代码   ![收藏代码][icon_star.png]

1.  /\*\* 
2.   \* 动态生成Jar包 
3.   \*/  
4.  **public** **static** File createJar(Class<?> clazz) **throws** Exception \{  
5.      String fqn = clazz.getName();  
6.      String base = fqn.substring(0, fqn.lastIndexOf("."));  
7.      base = "/" + base.replaceAll("\\\\.", Matcher.quoteReplacement("/"));  
8.      URL root = clazz.getResource("");  
9.    
10.     JarOutputStream out = **null**;  
11.     **final** File jar = File.createTempFile("HadoopRunningJar-", ".jar", **new** File(System.getProperty("java.io.tmpdir")));  
12.     System.out.println(jar.getAbsolutePath());  
13.     Runtime.getRuntime().addShutdownHook(**new** Thread() \{  
14.         **public** **void** run() \{  
15.             jar.delete();  
16.         \}  
17.     \});  
18.     **try** \{  
19.         File path = **new** File(root.toURI());  
20.         Manifest manifest = **new** Manifest();  
21.         manifest.getMainAttributes().putValue("Manifest-Version", "1.0");  
22.         manifest.getMainAttributes().putValue("Created-By", "RemoteHadoopUtil");  
23.         out = **new** JarOutputStream(**new** FileOutputStream(jar), manifest);  
24.         writeBaseFile(out, path, base);  
25.     \} **finally** \{  
26.         out.flush();  
27.         out.close();  
28.     \}  
29.     **return** jar;  
30. \}  
31.   
32. /\*\* 
33.  \* 递归添加.class文件 
34.  \*/  
35. **private** **static** **void** writeBaseFile(JarOutputStream out, File file, String base) **throws** IOException \{  
36.     **if** (file.isDirectory()) \{  
37.         File\[\] fl = file.listFiles();  
38.         **if** (base.length() > 0) \{  
39.             base = base + "/";  
40.         \}  
41.         **for** (**int** i = 0; i < fl.length; i++) \{  
42.             writeBaseFile(out, fl\[i\], base + fl\[i\].getName());  
43.         \}  
44.     \} **else** \{  
45.         out.putNextEntry(**new** JarEntry(base));  
46.         FileInputStream in = **null**;  
47.         **try** \{  
48.             in = **new** FileInputStream(file);  
49.             **byte**\[\] buffer = **new** **byte**\[1024\];  
50.             **int** n = in.read(buffer);  
51.             **while** (n != -1) \{  
52.                 out.write(buffer, 0, n);  
53.                 n = in.read(buffer);  
54.             \}  
55.         \} **finally** \{  
56.             in.close();  
57.         \}    
58.     \}  
59. \}

修改后的WordCount如下：

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** WordCount **extends** Configured **implements** Tool \{  
2.      **public** **static** **class** Map **extends** Mapper<LongWritable, Text, Text, IntWritable> \{  
3.          **private** **final** **static** IntWritable one = **new** IntWritable(1);  
4.          **private** Text word = **new** Text();  
5.    
6.          **public** **void** map(LongWritable key, Text value, Context context) **throws** IOException, InterruptedException \{  
7.              String line = value.toString();  
8.              System.out.println("line===>" + line);  
9.              StringTokenizer tokenizer = **new** StringTokenizer(line);  
10.             **while** (tokenizer.hasMoreTokens()) \{  
11.                 word.set(tokenizer.nextToken());  
12.                 context.write(word, one);  
13.             \}  
14.         \}  
15.     \}  
16.   
17.     **public** **static** **class** Reduce **extends** Reducer<Text, IntWritable, Text, IntWritable> \{  
18.         **public** **void** reduce(Text key, Iterable<IntWritable> values, Context context) **throws** IOException, InterruptedException \{  
19.             **int** sum = 0;  
20.             **for** (IntWritable val : values) \{  
21.                 sum += val.get();  
22.             \}  
23.             context.write(key, **new** IntWritable(sum));  
24.         \}  
25.     \}  
26.   
27.     **public** **int** run(String\[\] args) **throws** Exception \{  
28.         Configuration conf = getConf();  
29.         Job job = **new** Job(conf);  
30.         System.out.println(conf.get("mapred.job.tracker"));  
31.         System.out.println(conf.get("fs.default.name"));  
32.         /\*\* 
33.          \* TODO:调用二 
34.          \*/  
35.         File jarFile = RemoteHadoopUtil.createJar(WordCount.**class**);  
36.         ((JobConf) job.getConfiguration()).setJar(jarFile.toString());  
37.         job.setJarByClass(WordCount.**class**);  
38.         job.setJobName("wordcount");  
39.   
40.         job.setOutputKeyClass(Text.**class**);  
41.         job.setOutputValueClass(IntWritable.**class**);  
42.   
43.         job.setMapperClass(Map.**class**);  
44.         job.setReducerClass(Reduce.**class**);  
45.   
46.         job.setInputFormatClass(TextInputFormat.**class**);  
47.         job.setOutputFormatClass(TextOutputFormat.**class**);  
48.         String hdfs = "hdfs://192.168.0.128:9000";  
49.         args = **new** String\[\] \{ hdfs + "/user/gqshao/input/WordCount/", hdfs + "/user/gqshao/output/WordCount/" + **new** Date().getTime() \};  
50.         FileInputFormat.setInputPaths(job, **new** Path(args\[0\]));  
51.         FileOutputFormat.setOutputPath(job, **new** Path(args\[1\]));  
52.         **boolean** success = job.waitForCompletion(**true**);  
53.         System.out.println(job.isComplete());  
54.         System.out.println("JobID: " + job.getJobID());  
55.         **return** success ? 0 : 1;  
56.     \}  
57.   
58.     **public** **static** **void** main(String\[\] args) **throws** Exception \{  
59.         /\*\* 
60.          \* TODO:调用一 
61.          \*/  
62.         RemoteHadoopUtil.setConf(WordCount.**class**, Thread.currentThread(), "/hadoop");  
63.         **int** ret = ToolRunner.run(**new** WordCount(), args);  
64.         System.exit(ret);  
65.     \}  
66. \}

**转发请注明地址：http://sgq0085.iteye.com/admin/blogs/1879442**

附件中有完整代码和测试用例，欢迎讨论。解压后在文件目录中运行mvn eclipse:clean eclipse:eclipse即可(前提是需要有Maven)

*  [hadoop-remote.rar][] (15.4 KB)
 *  下载次数: 12
 *

[Hadoop Map_Reduce]: http://sgq0085.iteye.com/blog/1879442
[icon_star.png]: /images/20220828/f60469680f58432eadd8646aaf67606c.png
[hadoop-remote.rar]: http://dl.iteye.com/topics/download/c495a1f5-889a-3862-bd89-27c5ffd3966a