自定義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 即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>轉換成形式

* 注意:hadoop不認java的基本類型

* JAVA HADOOP

* long LongWritable

* String Text

*/

public class MyMapper extends Mapper {

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 即k2 表示每一行中的每一個單詞(可出現單詞一樣的行,在此還未進行分組)

* VALUEIN 即v2 表示每一行中每個單詞的出現次數,在這里,固定值為1

* KEYOUT 即k3 表示整個文件中不同的單詞(在此已經進行了分組)

* VALUEOUT 即v3 表示整個文件中不同單詞的出現總次數

* 文件里的內容:

* HELLO JAVA

* HELLO HADOOP

* reduce方法接收的參數是map方法輸出的結果:

* 經過reduce業務邏輯處理,輸出后為:,,

*/

public class MyReducer extends Reducer {

protected void reduce(Text k2, Iterable 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文件夾下的JARfile選項

3.選擇要被打成jar包的java文件以及jar包的輸出目錄

4.點擊下一步

5.選擇程序的入口,點擊完成即可

6.將jar包拷貝到linux環境,在linux的命令行中輸入以下語句執行

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

第一個路徑是文件讀取地址(即要被統計單詞的文件) ? 第二個路徑是文件輸出路徑(即統計單詞后輸出的文件)

總結

以上是生活随笔為你收集整理的Linux运行hadoop命令,将hadoop程序打成jar包,在linux下以命令行方式运行(例如单词计算程序)...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。