将hadoop程序打成jar包,在linux下以命令行方式运行(例如单词计算程序)

自定义Mapper

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
* Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>(指的是泛型)
* KEYIN 即k1 表示每一行的起始位置(偏移量offset)
* VALUEIN 即v1 表示每一行的文本内容
* KEYOUT 即k2 表示每一行中的每一个单词(可出现单词一样的行,在此还未进行分组)
* VALUEOUT 即v2 表示每一行中每个单词的出现次数,在这里,固定值为1
*
* 1.1从文件里读取内容:
* HELLO JAVA
* HELLO HADOOP
* 转换成<0,HEELO JAVA>,<11,HELLO HADOOP>形式 0和10表示每一行的偏移量从1开始,一个单词和空格都算一个偏移量
* 1.2将<0,HEELO JAVA>,<11,HELLO HADOOP>转换成<HELLO JAVA,1><HELLO HADOOP,1>形式
* 注意:hadoop不认java的基本类型
* JAVA HADOOP
* long LongWritable
* String Text
*/
public class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper.Context context)throws IOException, InterruptedException {
String[] splits = value.toString().split("\t");
for (String word : splits) {
context.write(new Text(word), new LongWritable(1L));
}
}
}

自定义Reducer

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
* Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>(指的是泛型)
* KEYIN 即k2 表示每一行中的每一个单词(可出现单词一样的行,在此还未进行分组)
* VALUEIN 即v2 表示每一行中每个单词的出现次数,在这里,固定值为1
* KEYOUT 即k3 表示整个文件中不同的单词(在此已经进行了分组)
* VALUEOUT 即v3 表示整个文件中不同单词的出现总次数
* 文件里的内容:
* HELLO JAVA
* HELLO HADOOP
* reduce方法接收的参数是map方法输出的结果:<HELLO,{1,1}><JAVA,{1}><HADOOP,{1}>
* 经过reduce业务逻辑处理,输出后为:<HELLO,2>,<JAVA,1>,<HADOOP,1>
*/
public class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context) throws IOException, InterruptedException {
long sum = 0L;
for (LongWritable v2 : v2s) {
sum += v2.get();
}
context.write(k2, new LongWritable(sum));
}
}

程序驱动

import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountTest {
static String INPUT_PATH = "";
static String OUT_PATH = "";

public static void main(String[] args) throws Exception {
INPUT_PATH = args[0];
OUT_PATH = args[1];

Configuration conf = new Configuration();

//如果输出目录存在就删除
FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH),conf);
if(fileSystem.exists(new Path(OUT_PATH))){
fileSystem.delete(new Path(OUT_PATH),true);//第二个参数,true表示删除的是文件夹,false表示删除的是文件
}
//第二个参数是作业名称(随意)
Job job = new Job(conf, WordCountTest.class.getSimpleName());

/**
* ----------------------这句代码是用于打包运行的(必须要写)-------------------
*/
job.setJarByClass(WordCountTest.class);

//1.1 指定输入文件目录
FileInputFormat.setInputPaths(job, INPUT_PATH);

//1.2 指定自定义的Mapper类
job.setMapperClass(MyMapper.class);

//1.3分区

//1.4 排序、分组

//1.5 (可选)归约

//2.1 分配节点,不需要我们关心

//2.2 指定自定义的Reducer类
job.setReducerClass(MyReducer.class);
//指定Reducer输出的key和value类型 以下两句话不能省略,因为省略了mapper就没依据 了
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);

//2.3 指定输出的路径
FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));

//把作业提交给JobTracker运行
job.waitForCompletion(true);
}
}
1.在eclipse项目中选中要被打包的程序入口,点击右键选择Export
2.点击java文件夹下的JAR
file选项



3.选择要被打成jar包的java文件以及jar包的输出目录



4.点击下一步



5.选择程序的入口,点击完成即可



6.将jar包拷贝到linux环境,在linux的命令行中输入以下语句执行

hadoop jar jar.jar hdfs://hadoop:9000/hello hdfs://hadoop:9000/testOut     

第一个路径是文件读取地址(即要被统计单词的文件)   第二个路径是文件输出路径(即统计单词后输出的文件)