public
class
WordCountApp {
//
可以指定目錄,目錄下如果有二級目錄的話,是不會執行的,只會執行一級目錄.
private
static
final
String INPUT_PATH = "hdfs://hadoop1:9000/abd";
//
輸入路徑
private
static
final
String OUT_PATH = "hdfs://hadoop1:9000/out";
//
輸出路徑,reduce作業輸出的結果是一個目錄
//
_SUCCESS:在linux中,帶下劃線的這些文件一般都是被忽略不去處理的.表示作業執行成功.
//
_logs:產生的日志文件.
//
part-r-00000:產生的是我們的輸出的文件.開始以part開始.r:reduce輸出的結果,map輸出的結果是m,00000是序號
public
static
void
main(String[] args) {
Configuration conf
=
new
Configuration();
//
配置對象
try
{
FileSystem fileSystem
= FileSystem.get(
new
URI(OUT_PATH), conf);
fileSystem.delete(
new
Path(OUT_PATH),
true
);
Job job
=
new
Job(conf, WordCountApp.
class
.getSimpleName());
//
jobName:作業名稱
job.setJarByClass(WordCountApp.
class
);
FileInputFormat.setInputPaths(job, INPUT_PATH);
//
指定數據的輸入
job.setMapperClass(MyMapper.
class
);
//
指定自定義map類
job.setMapOutputKeyClass(Text.
class
);
//
指定map輸出key的類型
job.setMapOutputValueClass(LongWritable.
class
);
//
指定map輸出value的類型
job.setReducerClass(MyReducer.
class
);
//
指定自定義Reduce類
job.setOutputKeyClass(Text.
class
);
//
設置Reduce輸出key的類型
job.setOutputValueClass(LongWritable.
class
);
//
設置Reduce輸出的value類型
FileOutputFormat.setOutputPath(job,
new
Path(OUT_PATH));
//
Reduce輸出完之后,就會產生一個最終的輸出,指定最終輸出的位置
job.waitForCompletion(
true
);
//
提交給jobTracker并等待結束
}
catch
(Exception e) {
e.printStackTrace();
}
}
/**
* 輸入的key標示偏移量:這一行開始的字節. 輸入的value:當前的行文本的內容. MapReduce執行過程:
* 在這里邊,我們的數據輸入來自于原始文件,數據輸出寫出到hdfs, 中間的一堆都是map輸出產生的臨時結果.存放在map運行的linux磁盤上的,
* 當經過shuffle時,reduce就會通過http把map端的對應數據給取過來.
* mapred-default.xml中mapredcue.jobtracker
* .root.dir,mapred.tmp.dir存儲map產生的結果. 作業運行時產生這個目錄,作業運行完之后它會刪除目錄.
*/
public
static
class
MyMapper
extends
Mapper
<LongWritable, Text, Text, LongWritable>
{
//
源文件有兩行記錄,解析源文件會產生兩個鍵值對.分別是<0,hello you>,<10,hello me>,所以map函數會被調用兩次.
//
在計算機存儲的時候,是一維的結構.
@Override
protected
void
map(LongWritable key, Text value, Context context)
throws
IOException, InterruptedException {
//
為什么要把hadoop類型轉換為java類型?
String line =
value.toString();
String[] splited
= line.split("\t"
);
//
使用hashMap寫出去的優勢:減少鍵值對出現的個數.
Map<String, Integer> hashMap =
new
HashMap<String, Integer>
();
for
(String word : splited) {
//
在for循環體內,臨時變量word出現的此時是常量1
context.write(
new
Text(word),
new
LongWritable(1));
//
把每個單詞出現的次數1寫出去.
}
}
}
//
map函數執行結束后,map輸出的<k,v>一共有4個.<hello,1>,<you,1>,<hello,1>,<me,1>
//
map把數據處理完之后,就會進入reduce.
//
在進入shuffle之前,數據需要先進行分區.默認只有一個區.
//
對每個不同分區中的數據進行排序,分組.
//
排序后的結果:<hello,1>,<hello,1>,<me,1>,<you,1>
//
分組后的結果(相同key的value放在一個集合中):<hello,{1,1}>,<me,{1}>,<you,{1}>
//
規約(可選)
//
map中的數據分發到reduce的過程稱作shuffle
public
static
class
MyReducer
extends
Reducer
<Text, LongWritable, Text, LongWritable>
{
//
每一組調用一次reduce函數,一共調用了三次
@Override
protected
void
reduce(Text key, Iterable<LongWritable>
values,
Context context)
throws
IOException, InterruptedException {
//
count標示單詞key在整個文件出現的次數
//
分組的數量與reduce函數調用次數是相等的.
//
reduce函數調用次數與產生的<k,v>的數量拋開業務,沒有任何關系!
long
count = 0L
;
for
(LongWritable times : values) {
count
+=
times.get();
}
context.write(key,
new
LongWritable(count));
}
}
}
?
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
