Apache Hadoop 运行分布式程序方法总结（Streaming方式与原生JAVA接口）-服务器

1. Hadoop Streaming方式运行程序Hadoop Streaming可以运行除JAVA语言以外，其它的语言编写的程序。其启动脚本示例如下： 1 #!/bin/sh 2 3# 参数合法性判断4 5if [ $# != 7 ]; then 6echo./bin/avp_platform_startu...

1. Hadoop Streaming方式运行程序

Hadoop Streaming可以运行除JAVA语言以外，其它的语言编写的程序。其启动脚本示例如下：

             1 #!/bin/sh 2 3# 参数合法性判断
 4 5if [ $# != 7 ]; then 6echo"./bin/avp_platform_startup.sh [USER_NAME] [INPUT_PAT] [OUTPUT_PAT] [MAP_TASKS] [REDUCE_TASKS] [CLASS_ID] [CODE_TYPE]" 7    exit
 8fi 910# GLOBAL VARS
11 USER_NAME=$112 INPUT_PAT=$213 OUTPUT_PAT=$314 MAP_TASKS=$415 REDUCE_TASKS=$516 CLASS_ID=$617 CODE_TYPE=$71819# Hadoop Start Area
20 /home/work/software/hadoop/bin/hadoop jar /home/work/software/hadoop/contrib/streaming/hadoop-streaming.jar 21     -input /home/$USER/$USER_NAME/output/webpage/$INPUT_PAT/22     -output /home/$USER/$USER_NAME/output/avp/avp-extract-$USER_NAME\_nlp_$OUTPUT_PAT-`date +%F-%H-%M-%S` 23     -mapper "avp_extract_mapper.sh . . $CODE_TYPE"24     -reducer "avp_extract_reducer.sh"25     -file ./script/avp_extract_mapper.sh26     -file ./script/avp_extract_reducer.sh27     -file ./script/extract_tools/tidy_page.py 28     -file ./script/decode.pl 29     -file ./script/extract_tools/format.py 30     -file ./script/extract_tools/extract_tool.py 31     -file ./class/$CLASS_ID/site.list 32     -file ./class/$CLASS_ID/*.conf 33    -jobconf mapred.job.name=$USER_NAME\_avp_platform_extract_tools_$OUTPUT_PAT 34    -jobconf mapred.map.tasks=$MAP_TASKS 35    -jobconf mapred.reduce.tasks=$REDUCE_TASKS

2. JAVA原生接口编写HADOOP程序

第一步，需要将用JAVA编写的代码打包成JAR包。

<SPAN style=‘font-family: "courier new", courier;‘>MaxTemperatureMapper.java

         1
        package
         oldapi;

         2
         3
        import
         java.io.IOException;

         4
         5
        import
         org.apache.hadoop.io.IntWritable;

         6
        import
         org.apache.hadoop.io.LongWritable;

         7
        import
         org.apache.hadoop.io.Text;

         8
        import
         org.apache.hadoop.mapred.MapReduceBase;

         9
        import
         org.apache.hadoop.mapred.Mapper;

        10
        import
         org.apache.hadoop.mapred.OutputCollector;

        11
        import
         org.apache.hadoop.mapred.Reporter;

        12
        13
        public
        class MaxTemperatureMapper extends MapReduceBase
14implements Mapper<LongWritable, Text, Text, IntWritable> {
1516privatestaticfinalint MISSING = 9999;
1718publicvoid map(LongWritable key, Text value,
19       OutputCollector<Text, IntWritable> output, Reporter reporter)
20throws IOException {
2122     String line = value.toString();
23     String year = line.substring(15, 19);
24int airTemperature;
25if (line.charAt(87) == ‘+‘) { // parseInt doesn‘t like leading plus signs26       airTemperature = Integer.parseInt(line.substring(88, 92));
27     } else {
28       airTemperature = Integer.parseInt(line.substring(87, 92));
29    }
30     String quality = line.substring(92, 93);
31if (airTemperature != MISSING && quality.matches("[01459]")) {
32       output.collect(new Text(year), new IntWritable(airTemperature));
33    }
34  }
35 }

<SPAN style=‘font-family: "courier new", courier; font-size: 16px;‘>MaxTemperatureReducer.java

         1
        package
         oldapi;

         2
         3
        import
         java.io.IOException;

         4
        import
         java.util.Iterator;

         5
         6
        import
         org.apache.hadoop.io.IntWritable;

         7
        import
         org.apache.hadoop.io.Text;

         8
        import
         org.apache.hadoop.mapred.MapReduceBase;

         9
        import
         org.apache.hadoop.mapred.OutputCollector;

        10
        import
         org.apache.hadoop.mapred.Reducer;

        11
        import
         org.apache.hadoop.mapred.Reporter;

        12
        13
        public
        class MaxTemperatureReducer extends MapReduceBase
14implements Reducer<Text, IntWritable, Text, IntWritable> {
1516publicvoid reduce(Text key, Iterator<IntWritable> values,
17       OutputCollector<Text, IntWritable> output, Reporter reporter)
18throws IOException {
1920int maxValue = Integer.MIN_VALUE;
21while (values.hasNext()) {
22       maxValue = Math.max(maxValue, values.next().get());
23    }
24     output.collect(key, new IntWritable(maxValue));
25  }
26 }

<SPAN style=‘font-family: "courier new", courier; font-size: 16px;‘>MaxTemperature.java

         1
        package
         oldapi;

         2
         3
        import
         java.io.IOException;

         4
         5
        import
         org.apache.hadoop.fs.Path;

         6
        import
         org.apache.hadoop.io.IntWritable;

         7
        import
         org.apache.hadoop.io.Text;

         8
        import
         org.apache.hadoop.mapred.FileInputFormat;

         9
        import
         org.apache.hadoop.mapred.FileOutputFormat;

        10
        import
         org.apache.hadoop.mapred.JobClient;

        11
        import
         org.apache.hadoop.mapred.JobConf;

        12
        import
         org.apache.hadoop.util.Tool;

        13
        import
         org.apache.hadoop.util.ToolRunner;

        14
        import
         org.apache.hadoop.conf.Configured;

        15
        import
         oldapi.MaxTemperatureMapper;

        16
        import
         oldapi.MaxTemperatureReducer;

        17
        18
        public
        class MaxTemperature extends Configured implements Tool{
1920publicint run(String[] args) throws IOException {
21if (args.length != 2) {
22       System.err.println("Usage: MaxTemperature <input path> <output path>");
23       System.exit(-1);
24    }
2526     JobConf conf = new JobConf(MaxTemperature.class);
27     conf.setJobName("Max temperature");
2829     FileInputFormat.addInputPath(conf, new Path(args[0]));
30     FileOutputFormat.setOutputPath(conf, new Path(args[1]));
3132     conf.setMapperClass(MaxTemperatureMapper.class);
33     conf.setCombinerClass(MaxTemperatureReducer.class);
34     conf.setReducerClass(MaxTemperatureReducer.class);
3536     conf.setOutputKeyClass(Text.class);
37     conf.setOutputValueClass(IntWritable.class);
3839    JobClient.runJob(conf);
40return 0;
41  }
4243publicstaticvoid main(String[] args) throws Exception {
44int exitCode = ToolRunner.run(new MaxTemperature(),args);
45      System.exit(exitCode);
46  }
4748 }

打包命令：（即会生成MaxTemperature.jar文件，-C参数指定的文件夹目录结构：MaxTemperature/oldapi/*.java）

        1 jar cvf MaxTemperature.jar -C MaxTemperature/ .

第二步，启动运行。

        1
        sudo -u sniper hadoop jar MaxTemperature.jar oldapi.MaxTemperature /home/sniper/zhuliang/sample.txt /home/sniper/zhuliang/max-temp-output

附：输入文件样例

        1
        0067011990999991950051507004+68750+023550FM-12+038299999V0203301N00671220001CN9999999N9+00001+99999999999

        2
        0043011990999991950051512004+68750+023550FM-12+038299999V0203201N00671220001CN9999999N9+00221+99999999999

        3
        0043011990999991950051518004+68750+023550FM-12+038299999V0203201N00261220001CN9999999N9-00111+99999999999

        4
        0043012650999991949032412004+62300+010750FM-12+048599999V0202701N00461220001CN0500001N9+01111+99999999999

        5 0043012650999991949032418004+62300+010750FM-12+048599999V0202701N00461220001CN0500001N9+00781+99999999999

原文：http://www.cnblogs.com/bjzhuliang/p/3618724.html