Join.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:6k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.examples;
  19. import java.io.IOException;
  20. import java.util.*;
  21. import org.apache.hadoop.conf.Configuration;
  22. import org.apache.hadoop.conf.Configured;
  23. import org.apache.hadoop.fs.Path;
  24. import org.apache.hadoop.io.BytesWritable;
  25. import org.apache.hadoop.io.Writable;
  26. import org.apache.hadoop.io.WritableComparable;
  27. import org.apache.hadoop.mapred.*;
  28. import org.apache.hadoop.mapred.join.*;
  29. import org.apache.hadoop.mapred.lib.IdentityMapper;
  30. import org.apache.hadoop.mapred.lib.IdentityReducer;
  31. import org.apache.hadoop.util.Tool;
  32. import org.apache.hadoop.util.ToolRunner;
  33. /**
  34.  * This is the trivial map/reduce program that does absolutely nothing
  35.  * other than use the framework to fragment and sort the input values.
  36.  *
  37.  * To run: bin/hadoop jar build/hadoop-examples.jar join
  38.  *            [-m <i>maps</i>] [-r <i>reduces</i>]
  39.  *            [-inFormat <i>input format class</i>] 
  40.  *            [-outFormat <i>output format class</i>] 
  41.  *            [-outKey <i>output key class</i>] 
  42.  *            [-outValue <i>output value class</i>] 
  43.  *            [-joinOp &lt;inner|outer|override&gt;]
  44.  *            [<i>in-dir</i>]* <i>in-dir</i> <i>out-dir</i> 
  45.  */
  46. public class Join extends Configured implements Tool {
  47.   static int printUsage() {
  48.     System.out.println("join [-m <maps>] [-r <reduces>] " +
  49.                        "[-inFormat <input format class>] " +
  50.                        "[-outFormat <output format class>] " + 
  51.                        "[-outKey <output key class>] " +
  52.                        "[-outValue <output value class>] " +
  53.                        "[-joinOp <inner|outer|override>] " +
  54.                        "[input]* <input> <output>");
  55.     ToolRunner.printGenericCommandUsage(System.out);
  56.     return -1;
  57.   }
  58.   /**
  59.    * The main driver for sort program.
  60.    * Invoke this method to submit the map/reduce job.
  61.    * @throws IOException When there is communication problems with the 
  62.    *                     job tracker.
  63.    */
  64.   public int run(String[] args) throws Exception {
  65.     JobConf jobConf = new JobConf(getConf(), Sort.class);
  66.     jobConf.setJobName("join");
  67.     jobConf.setMapperClass(IdentityMapper.class);        
  68.     jobConf.setReducerClass(IdentityReducer.class);
  69.     JobClient client = new JobClient(jobConf);
  70.     ClusterStatus cluster = client.getClusterStatus();
  71.     int num_maps = cluster.getTaskTrackers() * 
  72.                    jobConf.getInt("test.sort.maps_per_host", 10);
  73.     int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
  74.     String sort_reduces = jobConf.get("test.sort.reduces_per_host");
  75.     if (sort_reduces != null) {
  76.        num_reduces = cluster.getTaskTrackers() * 
  77.                        Integer.parseInt(sort_reduces);
  78.     }
  79.     Class<? extends InputFormat> inputFormatClass = 
  80.       SequenceFileInputFormat.class;
  81.     Class<? extends OutputFormat> outputFormatClass = 
  82.       SequenceFileOutputFormat.class;
  83.     Class<? extends WritableComparable> outputKeyClass = BytesWritable.class;
  84.     Class<? extends Writable> outputValueClass = TupleWritable.class;
  85.     String op = "inner";
  86.     List<String> otherArgs = new ArrayList<String>();
  87.     for(int i=0; i < args.length; ++i) {
  88.       try {
  89.         if ("-m".equals(args[i])) {
  90.           num_maps = Integer.parseInt(args[++i]);
  91.         } else if ("-r".equals(args[i])) {
  92.           num_reduces = Integer.parseInt(args[++i]);
  93.         } else if ("-inFormat".equals(args[i])) {
  94.           inputFormatClass = 
  95.             Class.forName(args[++i]).asSubclass(InputFormat.class);
  96.         } else if ("-outFormat".equals(args[i])) {
  97.           outputFormatClass = 
  98.             Class.forName(args[++i]).asSubclass(OutputFormat.class);
  99.         } else if ("-outKey".equals(args[i])) {
  100.           outputKeyClass = 
  101.             Class.forName(args[++i]).asSubclass(WritableComparable.class);
  102.         } else if ("-outValue".equals(args[i])) {
  103.           outputValueClass = 
  104.             Class.forName(args[++i]).asSubclass(Writable.class);
  105.         } else if ("-joinOp".equals(args[i])) {
  106.           op = args[++i];
  107.         } else {
  108.           otherArgs.add(args[i]);
  109.         }
  110.       } catch (NumberFormatException except) {
  111.         System.out.println("ERROR: Integer expected instead of " + args[i]);
  112.         return printUsage();
  113.       } catch (ArrayIndexOutOfBoundsException except) {
  114.         System.out.println("ERROR: Required parameter missing from " +
  115.             args[i-1]);
  116.         return printUsage(); // exits
  117.       }
  118.     }
  119.     // Set user-supplied (possibly default) job configs
  120.     jobConf.setNumMapTasks(num_maps);
  121.     jobConf.setNumReduceTasks(num_reduces);
  122.     if (otherArgs.size() < 2) {
  123.       System.out.println("ERROR: Wrong number of parameters: ");
  124.       return printUsage();
  125.     }
  126.     FileOutputFormat.setOutputPath(jobConf, 
  127.       new Path(otherArgs.remove(otherArgs.size() - 1)));
  128.     List<Path> plist = new ArrayList<Path>(otherArgs.size());
  129.     for (String s : otherArgs) {
  130.       plist.add(new Path(s));
  131.     }
  132.     jobConf.setInputFormat(CompositeInputFormat.class);
  133.     jobConf.set("mapred.join.expr", CompositeInputFormat.compose(
  134.           op, inputFormatClass, plist.toArray(new Path[0])));
  135.     jobConf.setOutputFormat(outputFormatClass);
  136.     jobConf.setOutputKeyClass(outputKeyClass);
  137.     jobConf.setOutputValueClass(outputValueClass);
  138.     Date startTime = new Date();
  139.     System.out.println("Job started: " + startTime);
  140.     JobClient.runJob(jobConf);
  141.     Date end_time = new Date();
  142.     System.out.println("Job ended: " + end_time);
  143.     System.out.println("The job took " + 
  144.         (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
  145.     return 0;
  146.   }
  147.   public static void main(String[] args) throws Exception {
  148.     int res = ToolRunner.run(new Configuration(), new Join(), args);
  149.     System.exit(res);
  150.   }
  151. }