package org.apache.hadoop.examples;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.join.*;
import org.apache.hadoop.mapred.lib.IdentityMapper;
import org.apache.hadoop.mapred.lib.IdentityReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
hadoop的map/reduce例子,演示了用CompositeInputFormat类来对两个数据集进行join操作,可以是inner join;outerjoin;也可以是override join。
* This is the trivial map/reduce program that does absolutely nothing
* other than use the framework to fragment and sort the input values.
*
* To run: bin/hadoop jar build/hadoop-examples.jar join
* [-m <i>maps</i>] [-r <i>reduces</i>]
* [-inFormat <i>input format class</i>]
* [-outFormat <i>output format class</i>]
* [-outKey <i>output key class</i>]
* [-outValue <i>output value class</i>]
* [-joinOp <inner|outer|override>]
* [<i>in-dir</i>]* <i>in-dir</i> <i>out-dir</i>
*/
public class Join extends Configured implements Tool {
static int printUsage() {
System.out.println("join [-m <maps>] [-r <reduces>] " +
"[-inFormat <input format class>] " +
"[-outFormat <output format class>] " +
"[-outKey <output key class>] " +
"[-outValue <output value class>] " +
"[-joinOp <inner|outer|override>] " +
"[input]* <input> <output>");
ToolRunner.printGenericCommandUsage(System.out);
return -1;
}
/**driver代码
* The main driver for sort program.
* Invoke this method to submit the map/reduce job.
* @throws IOException When there is communication problems with the
* job tracker.
*/
public int run(String[] args) throws Exception {
JobConf jobConf = new JobConf(getConf(), Sort.class);
jobConf.setJobName("join");
jobConf.setMapperClass(IdentityMapper.class); //设置map类是IdentityMapper
jobConf.setReducerClass(IdentityReducer.class);//设置reduce类是IdentityReducer
JobClient client = new JobClient(jobConf);
ClusterStatus cluster = client.getClusterStatus();
int num_maps = cluster.getTaskTrackers() *
jobConf.getInt("test.sort.maps_per_host", 10);
int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
String sort_reduces = jobConf.get("test.sort.reduces_per_host");
if (sort_reduces != null) {
num_reduces = cluster.getTaskTrackers() *
Integer.parseInt(sort_reduces);
}
Class<? extends InputFormat> inputFormatClass =
SequenceFileInputFormat.class;
Class<? extends OutputFormat> outputFormatClass =
SequenceFileOutputFormat.class;
Class<? extends WritableComparable> outputKeyClass = BytesWritable.class;
Class<? extends Writable> outputValueClass = TupleWritable.class;
String op = "inner"; //join是inner join
List<String> otherArgs = new ArrayList<String>();
for(int i=0; i < args.length; ++i) {
try {
if ("-m".equals(args[i])) {
num_maps = Integer.parseInt(args[++i]);
} else if ("-r".equals(args[i])) {
num_reduces = Integer.parseInt(args[++i]);
} else if ("-inFormat".equals(args[i])) {
inputFormatClass =
Class.forName(args[++i]).asSubclass(InputFormat.class);
} else if ("-outFormat".equals(args[i])) {
outputFormatClass =
Class.forName(args[++i]).asSubclass(OutputFormat.class);
} else if ("-outKey".equals(args[i])) {
outputKeyClass =
Class.forName(args[++i]).asSubclass(WritableComparable.class);
} else if ("-outValue".equals(args[i])) {
outputValueClass =
Class.forName(args[++i]).asSubclass(Writable.class);
} else if ("-joinOp".equals(args[i])) {
op = args[++i];
} else {
otherArgs.add(args[i]);
}
} catch (NumberFormatException except) {
System.out.println("ERROR: Integer expected instead of " + args[i]);
return printUsage();
} catch (ArrayIndexOutOfBoundsException except) {
System.out.println("ERROR: Required parameter missing from " +
args[i-1]);
return printUsage(); // exits
}
}
// Set user-supplied (possibly default) job configs
jobConf.setNumMapTasks(num_maps);
jobConf.setNumReduceTasks(num_reduces);
if (otherArgs.size() < 2) {
System.out.println("ERROR: Wrong number of parameters: ");
return printUsage();
}
FileOutputFormat.setOutputPath(jobConf,
new Path(otherArgs.remove(otherArgs.size() - 1)));
List<Path> plist = new ArrayList<Path>(otherArgs.size());
for (String s : otherArgs) {
plist.add(new Path(s));
}
jobConf.setInputFormat(CompositeInputFormat.class);//设置输入格式类是CompositeInputFormat
jobConf.set("mapred.join.expr", CompositeInputFormat.compose(
op, inputFormatClass, plist.toArray(new Path[0])));
//设置CompositeInputFormat的参数,包括join类型,输入格式类型。
jobConf.setOutputFormat(outputFormatClass);
jobConf.setOutputKeyClass(outputKeyClass);
jobConf.setOutputValueClass(outputValueClass);
Date startTime = new Date();
System.out.println("Job started: " + startTime);
JobClient.runJob(jobConf);
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took " +
(end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new Join(), args);
System.exit(res);
}
}
分享到:
相关推荐
代码范例列表 第1章 示例描述:本章演示如何开始使用JDK进行程序的开发。 HelloWorldApp.java 第一个用Java开发的应用程序。 firstApplet.java 第一个用Java开发的Applet小程序。 firstApplet.htm 用来装载...
赠送源代码:lucene-join-7.2.1-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.2.1.pom; 包含翻译后的API文档:lucene-join-7.2.1-javadoc-API文档-中文(简体)-英语-对照版.zip; Maven坐标:org.apache....
赠送源代码:lucene-join-6.6.0-sources.jar; 赠送Maven依赖信息文件:lucene-join-6.6.0.pom; 包含翻译后的API文档:lucene-join-6.6.0-javadoc-API文档-中文(简体)版.zip; Maven坐标:org.apache.lucene:lucene...
赠送源代码:parent-join-client-6.2.3-sources.jar 包含翻译后的API文档:parent-join-client-6.2.3-javadoc-API文档-中文(简体)版.zip 对应Maven信息:groupId:org.elasticsearch.plugin,artifactId:parent-...
赠送源代码:eclipse-collections-forkjoin-7.1.2-sources.jar; 赠送Maven依赖信息文件:eclipse-collections-forkjoin-7.1.2.pom; 包含翻译后的API文档:eclipse-collections-forkjoin-7.1.2-javadoc-API文档-...
赠送源代码:parent-join-client-6.2.3-sources.jar; 赠送Maven依赖信息文件:parent-join-client-6.2.3.pom; 包含翻译后的API文档:parent-join-client-6.2.3-javadoc-API文档-中文(简体)-英语-对照版.zip; ...
赠送源代码:parent-join-client-5.5.1-sources.jar; 赠送Maven依赖信息文件:parent-join-client-5.5.1.pom; 包含翻译后的API文档:parent-join-client-5.5.1-javadoc-API文档-中文(简体)-英语-对照版.zip; ...
赠送源代码:parent-join-client-6.3.0-sources.jar; 赠送Maven依赖信息文件:parent-join-client-6.3.0.pom; 包含翻译后的API文档:parent-join-client-6.3.0-javadoc-API文档-中文(简体)-英语-对照版.zip; ...
赠送源代码:parent-join-client-5.5.1-sources.jar; 赠送Maven依赖信息文件:parent-join-client-5.5.1.pom; 包含翻译后的API文档:parent-join-client-5.5.1-javadoc-API文档-中文(简体)版.zip; Maven坐标:org...
赠送源代码:lucene-join-7.7.0-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.7.0.pom; 包含翻译后的API文档:lucene-join-7.7.0-javadoc-API文档-中文(简体)版.zip; Maven坐标:org.apache.lucene:lucene...
赠送源代码:lucene-join-7.2.1-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.2.1.pom; 包含翻译后的API文档:lucene-join-7.2.1-javadoc-API文档-中文(简体)版.zip; Maven坐标:org.apache.lucene:lucene...
赠送源代码:lucene-join-7.3.1-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.3.1.pom; 包含翻译后的API文档:lucene-join-7.3.1-javadoc-API文档-中文(简体)版.zip; Maven坐标:org.apache.lucene:lucene...
赠送源代码:lucene-join-7.7.0-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.7.0.pom; 包含翻译后的API文档:lucene-join-7.7.0-javadoc-API文档-中文(简体)-英语-对照版.zip; Maven坐标:org.apache....
赠送源代码:lucene-join-7.3.1-sources.jar; 赠送Maven依赖信息文件:lucene-join-7.3.1.pom; 包含翻译后的API文档:lucene-join-7.3.1-javadoc-API文档-中文(简体)-英语-对照版.zip; Maven坐标:org.apache....
赠送源代码:lucene-join-6.6.0-sources.jar; 赠送Maven依赖信息文件:lucene-join-6.6.0.pom; 包含翻译后的API文档:lucene-join-6.6.0-javadoc-API文档-中文(简体)-英语-对照版.zip; Maven坐标:org.apache....
赠送源代码:parent-join-client-6.8.3-sources.jar; 赠送Maven依赖信息文件:parent-join-client-6.8.3.pom; 包含翻译后的API文档:parent-join-client-6.8.3-javadoc-API文档-中文(简体)版.zip; Maven坐标:org...
赠送源代码:parent-join-client-6.3.0-sources.jar; 赠送Maven依赖信息文件:parent-join-client-6.3.0.pom; 包含翻译后的API文档:parent-join-client-6.3.0-javadoc-API文档-中文(简体)版.zip; Maven坐标:org...
赠送源代码:parent-join-client-6.8.3-sources.jar; 赠送Maven依赖信息文件:parent-join-client-6.8.3.pom; 包含翻译后的API文档:parent-join-client-6.8.3-javadoc-API文档-中文(简体)-英语-对照版.zip; ...
赠送源代码:eclipse-collections-forkjoin-7.1.2-sources.jar; 赠送Maven依赖信息文件:eclipse-collections-forkjoin-7.1.2.pom; 包含翻译后的API文档:eclipse-collections-forkjoin-7.1.2-javadoc-API文档-...
它提供完全的Unicode支持,并且还可以处理大型CSS源代码。 ## Installation这是一个作曲家软件包。 有关基本的安装信息,请参见。 将以下行添加到您的composer.json文件: { "require" : { "crossjoin/css" : "1.0.*...