如何在hadoop2.5.2使用命令行编译打包运行自己的mapreduce程序

网上的 MapReduce WordCount 教程对于如何编译 WordCount.Java 几乎是一笔带过… 而有写到的，大多又是 0.20 等旧版本版本的做法，即 javac -classpath /usr/local/Hadoop/hadoop-1.0.1/hadoop-core-1.0.1.jar WordCount.java，但较新的 2.X 版本中，已经没有 hadoop-core*.jar 这个文件，因此编辑和打包自己的 MapReduce 程序与旧版本有所不同。

本文以 Hadoop 2.7.2 环境下的 WordCount 实例来介绍 2.x 版本中如何编辑自己的 MapReduce 程序。

编译、打包 Hadoop MapReduce 程序

我们将 Hadoop 的 classhpath 信息添加到 CLASSPATH 变量中，在 ~/.bashrc 中增加如下几行：

[html] view plain copy

export HADOOP_HOME=/usr/local/hadoop

export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath):$CLASSPATH

别忘了执行 source ~/.bashrc 使变量生效，接着就可以通过 javac 命令编译 WordCount.java 了（使用的是 Hadoop 源码中的 WordCount.java，源码在文本最后面）：javac WordCount.java

编译时会有警告，可以忽略。编译后可以看到生成了几个 .class 文件。

接着把 .class 文件打包成 jar，才能在 Hadoop 中运行：

[html] view plain copy

jar -cvf WordCount.jar ./WordCount*.class

开始运行：

[html] view plain copy

hadoop jar WordCount.jar WordCount input output//hdfs上的input文件夹，命令执行所在位置为WordCount.jar同一目录

因为程序中声明了

package ，所以在命令中也要 org.apache.hadoop.examples 写完整：

[html] view plain copy

hadoop jar WordCount.jar org.apache.hadoop.examples.WordCount input output

查看：

[html] view plain copy

hadoop fs -cat /output/part-r-00000

WordCount.java 源码

package org.apache.hadoop.examples

import java.io.IOException

import java.util.StringTokenizer

import org.apache.hadoop.conf.Configuration

import org.apache.hadoop.fs.Path

import org.apache.hadoop.io.IntWritable

import org.apache.hadoop.io.Text

import org.apache.hadoop.mapreduce.Job

import org.apache.hadoop.mapreduce.Mapper

import org.apache.hadoop.mapreduce.Reducer

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat

import org.apache.hadoop.util.GenericOptionsParser

public class WordCount {

public static class TokenizerMapper

extends Mapper<Object, Text, Text, IntWritable>{

private final static IntWritable one = new IntWritable(1)

private Text word = new Text()

public void map(Object key, Text value, Context context

) throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString())

while (itr.hasMoreTokens()) {

word.set(itr.nextToken())

context.write(word, one)

}

}

}

public static class IntSumReducer

extends Reducer<Text,IntWritable,Text,IntWritable>{

private IntWritable result = new IntWritable()

public void reduce(Text key, Iterable<IntWritable>values,

Context context

) throws IOException, InterruptedException {

int sum = 0

for (IntWritable val : values) {

sum += val.get()

}

result.set(sum)

context.write(key, result)

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration()

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs()

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount <in><out>")

System.exit(2)

}

Job job = new Job(conf, "word count")

job.setJarByClass(WordCount.class)

job.setMapperClass(TokenizerMapper.class)

job.setCombinerClass(IntSumReducer.class)

job.setReducerClass(IntSumReducer.class)

job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

FileInputFormat.addInputPath(job, new Path(otherArgs[0]))

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]))

System.exit(job.waitForCompletion(true) ? 0 : 1)

}

}

大数据的时代， 到处张嘴闭嘴都是Hadoop, MapReduce, 不跟上时代怎么行？ 可是对一个hadoop的新手， 写一个属于自己的MapReduce程序还是小有点难度的， 需要建立一个maven项目， 还要搞清楚各种库的依赖， 再加上编译运行， 基本上头大两圈了吧。 这也使得很多只是想简单了解一下MapReduce的人望而却步。

本文会教你如何用最快最简单的方法编写和运行一个属于自己的MapReduce程序， let's go!

首先有两个前提：

1. 有一个已经可以运行的hadoop 集群（也可以是伪分布系统)， 上面的hdfs和mapreduce工作正常 （这个真的是最基本的了， 不再累述， 不会的请参考 http://hadoop.apache.org/docs/current/)

2. 集群上安装了JDK (编译运行时会用到）

正式开始

1. 首先登入hadoop 集群里面的一个节点， 创建一个java源文件， 偷懒起见， 基本盗用官方的word count (因为本文的目的是教会你如何快编写和运行一个MapReduce程序， 而不是如何写好一个功能齐全的MapReduce程序）

内容如下：

import java.io.IOException

import java.util.StringTokenizer

import org.apache.hadoop.conf.Configuration

import org.apache.hadoop.fs.Path

import org.apache.hadoop.io.IntWritable

import org.apache.hadoop.io.Text

import org.apache.hadoop.mapreduce.Job

import org.apache.hadoop.mapreduce.Mapper

import org.apache.hadoop.mapreduce.Reducer

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat

import org.apache.hadoop.util.GenericOptionsParser

public class myword {

public static class TokenizerMapper

extends Mapper<Object, Text, Text, IntWritable>{

private final static IntWritable one = new IntWritable(1)

private Text word = new Text()

public void map(Object key, Text value, Context context

) throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString())

while (itr.hasMoreTokens()) {

word.set(itr.nextToken())

context.write(word, one)

}

}

}

public static class IntSumReducer

extends Reducer<Text,IntWritable,Text,IntWritable>{

private IntWritable result = new IntWritable()

public void reduce(Text key, Iterable<IntWritable>values,

Context context

) throws IOException, InterruptedException {

int sum = 0

for (IntWritable val : values) {

sum += val.get()

}

result.set(sum)

context.write(key, result)

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration()

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs()

if (otherArgs.length != 2) {

System.err.println('Usage: wordcount <in><out>')

System.exit(2)

}

Job job = new Job(conf, 'word count')

job.setJarByClass(myword.class)

job.setMapperClass(TokenizerMapper.class)

job.setCombinerClass(IntSumReducer.class)

job.setReducerClass(IntSumReducer.class)

job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

FileInputFormat.addInputPath(job, new Path(otherArgs[0]))

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]))

System.exit(job.waitForCompletion(true) ? 0 : 1)

}

}

与官方版本相比， 主要做了两处修改

1） 为了简单起见，去掉了开头的 package org.apache.hadoop.examples

2） 将类名从 WordCount 改为 myword, 以体现是我们自己的工作成果 :)

2. 拿到hadoop 运行的class path, 主要为编译所用

运行命令

hadoop classpath

保存打出的结果，本文用的hadoop 版本是Pivotal 公司的Pivotal hadoop, 例子：

/etc/gphd/hadoop/conf:/usr/lib/gphd/hadoop/lib/*:/usr/lib/gphd/hadoop/.//*:/usr/lib/gphd/hadoop-hdfs/./:/usr/lib/gphd/hadoop-hdfs/lib/*:/usr/lib/gphd/hadoop-hdfs/.//*:/usr/lib/gphd/hadoop-yarn/lib/*:/usr/lib/gphd/hadoop-yarn/.//*:/usr/lib/gphd/hadoop-mapreduce/lib/*:/usr/lib/gphd/hadoop-mapreduce/.//*::/etc/gphd/pxf/conf::/usr/lib/gphd/pxf/pxf-core.jar:/usr/lib/gphd/pxf/pxf-api.jar:/usr/lib/gphd/publicstage:/usr/lib/gphd/gfxd/lib/gemfirexd.jar::/usr/lib/gphd/zookeeper/zookeeper.jar:/usr/lib/gphd/hbase/lib/hbase-common.jar:/usr/lib/gphd/hbase/lib/hbase-protocol.jar:/usr/lib/gphd/hbase/lib/hbase-client.jar:/usr/lib/gphd/hbase/lib/hbase-thrift.jar:/usr/lib/gphd/hbase/lib/htrace-core-2.01.jar:/etc/gphd/hbase/conf::/usr/lib/gphd/hive/lib/hive-service.jar:/usr/lib/gphd/hive/lib/libthrift-0.9.0.jar:/usr/lib/gphd/hive/lib/hive-metastore.jar:/usr/lib/gphd/hive/lib/libfb303-0.9.0.jar:/usr/lib/gphd/hive/lib/hive-common.jar:/usr/lib/gphd/hive/lib/hive-exec.jar:/usr/lib/gphd/hive/lib/postgresql-jdbc.jar:/etc/gphd/hive/conf::/usr/lib/gphd/sm-plugins/*:

3. 编译

运行命令

javac -classpath xxx ./myword.java

xxx部分就是上一步里面取到的class path

运行完此命令后， 当前目录下会生成一些.class 文件， 例如：

myword.class myword$IntSumReducer.class myword$TokenizerMapper.class

4. 将class文件打包成.jar文件

运行命令

jar -cvf myword.jar ./*.class

至此, 目标jar 文件成功生成

5. 准备一些文本文件， 上传到hdfs, 以做word count的input

例子：

随意创建一些文本文件， 保存到mapred_test 文件夹

运行命令

hadoop fs -put ./mapred_test/

确保此文件夹成功上传到hdfs 当前用户根目录下

6. 运行我们的程序

运行命令

hadoop jar ./myword.jar myword mapred_test output

顺利的话， 此命令会正常进行， 一个MapReduce job 会开始工作， 输出的结果会保存在 hdfs 当前用户根目录下的output 文件夹里面。

至此大功告成！

如果还需要更多的功能， 我们可以修改前面的源文件以达到一个真正有用的MapReduce job。

但是原理大同小异， 练手的话， 基本够了。

一个抛砖引玉的简单例子， 欢迎板砖。

转载

---