Hadoop之——命令列執行時指定引數
阿新 • • 發佈：2019-02-18
本文旨在提供一個Hadoop在執行的時候從命令列輸入要統計的檔案路徑和統計結果的輸出路徑，不多說直接上程式碼
1、Mapper類的實現
2、Reducer類的實現/** * KEYIN 即k1 表示行的偏移量 * VALUEIN 即v1 表示行文字內容 * KEYOUT 即k2 表示行中出現的單詞 * VALUEOUT 即v2 表示行中出現的單詞的次數，固定值1 */ static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{ protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException { final String[] splited = v1.toString().split("\t"); for (String word : splited) { context.write(new Text(word), new LongWritable(1)); } }; }
3、run方法的實現/** * KEYIN 即k2 表示行中出現的單詞 * VALUEIN 即v2 表示行中出現的單詞的次數 * KEYOUT 即k3 表示文字中出現的不同單詞 * VALUEOUT 即v3 表示文字中出現的不同單詞的總次數 * */ static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{ protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException { long times = 0L; for (LongWritable count : v2s) { times += count.get(); } ctx.write(k2, new LongWritable(times)); }; }
4、程式入口main@Override public int run(String[] args) throws Exception { //接收命令列引數 INPUT_PATH = args[0]; OUT_PATH = args[1]; Configuration conf = new Configuration(); final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf); final Path outPath = new Path(OUT_PATH); //如果已經存在輸出檔案，則先刪除已存在的輸出檔案 if(fileSystem.exists(outPath)){ fileSystem.delete(outPath, true); } final Job job = new Job(conf , WordCount.class.getSimpleName()); //*******打包執行必須執行的方法******* job.setJarByClass(WordCount.class); //1.1指定讀取的檔案位於哪裡 FileInputFormat.setInputPaths(job, INPUT_PATH); //指定如何對輸入檔案進行格式化，把輸入檔案每一行解析成鍵值對 job.setInputFormatClass(TextInputFormat.class); //1.2 指定自定義的map類 job.setMapperClass(MyMapper.class); //map輸出的<k,v>型別。如果<k3,v3>的型別與<k2,v2>型別一致，下面兩行程式碼可以省略 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); //1.3 分割槽 job.setPartitionerClass(HashPartitioner.class); //有一個reduce任務執行 job.setNumReduceTasks(1); //1.4 TODO 排序、分組 //1.5 TODO 規約 //2.2 指定自定義reduce類 job.setReducerClass(MyReducer.class); //指定reduce的輸出型別 job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); //2.3 指定寫出到哪裡 FileOutputFormat.setOutputPath(job, outPath); //指定輸出檔案的格式化類 job.setOutputFormatClass(TextOutputFormat.class); //把job提交給JobTracker執行 job.waitForCompletion(true); return 0; }
//程式入口Main方法
public static void main(String[] args) throws Exception {
ToolRunner.run(new WordCount(), args);
}package com.lyz.hadoop.count;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* 利用Hadoop MapReduce統計文字中每個單詞的數量
* @author liuyazhuang
*/
public class WordCount extends Configured implements Tool{
//要統計的檔案位置
static String INPUT_PATH = "";
//統計結果輸出的位置
static String OUT_PATH = "";
@Override
public int run(String[] args) throws Exception {
//接收命令列引數
INPUT_PATH = args[0];
OUT_PATH = args[1];
Configuration conf = new Configuration();
final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
final Path outPath = new Path(OUT_PATH);
//如果已經存在輸出檔案，則先刪除已存在的輸出檔案
if(fileSystem.exists(outPath)){
fileSystem.delete(outPath, true);
}
final Job job = new Job(conf , WordCount.class.getSimpleName());
//*******打包執行必須執行的方法*******
job.setJarByClass(WordCount.class);
//1.1指定讀取的檔案位於哪裡
FileInputFormat.setInputPaths(job, INPUT_PATH);
//指定如何對輸入檔案進行格式化，把輸入檔案每一行解析成鍵值對
job.setInputFormatClass(TextInputFormat.class);
//1.2 指定自定義的map類
job.setMapperClass(MyMapper.class);
//map輸出的<k,v>型別。如果<k3,v3>的型別與<k2,v2>型別一致，下面兩行程式碼可以省略
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
//1.3 分割槽
job.setPartitionerClass(HashPartitioner.class);
//有一個reduce任務執行
job.setNumReduceTasks(1);
//1.4 TODO 排序、分組
//1.5 TODO 規約
//2.2 指定自定義reduce類
job.setReducerClass(MyReducer.class);
//指定reduce的輸出型別
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
//2.3 指定寫出到哪裡
FileOutputFormat.setOutputPath(job, outPath);
//指定輸出檔案的格式化類
job.setOutputFormatClass(TextOutputFormat.class);
//把job提交給JobTracker執行
job.waitForCompletion(true);
return 0;
}
//程式入口Main方法
public static void main(String[] args) throws Exception {
ToolRunner.run(new WordCount(), args);
}
/**
* KEYIN 即k1 表示行的偏移量
* VALUEIN 即v1 表示行文字內容
* KEYOUT 即k2 表示行中出現的單詞
* VALUEOUT 即v2 表示行中出現的單詞的次數，固定值1
*/
static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException {
final String[] splited = v1.toString().split("\t");
for (String word : splited) {
context.write(new Text(word), new LongWritable(1));
}
};
}
/**
* KEYIN 即k2 表示行中出現的單詞
* VALUEIN 即v2 表示行中出現的單詞的次數
* KEYOUT 即k3 表示文字中出現的不同單詞
* VALUEOUT 即v3 表示文字中出現的不同單詞的總次數
*
*/
static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException {
long times = 0L;
for (LongWritable count : v2s) {
times += count.get();
}
ctx.write(k2, new LongWritable(times));
};
}
}