建站
咨詢
售后
建站咨詢
新聞中心
MapReduce概述：
在鄲城等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強(qiáng)發(fā)展的系統(tǒng)性、市場(chǎng)前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供網(wǎng)站設(shè)計(jì)制作、成都網(wǎng)站設(shè)計(jì) 網(wǎng)站設(shè)計(jì)制作定制網(wǎng)站,公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),成都品牌網(wǎng)站建設(shè),全網(wǎng)整合營(yíng)銷(xiāo)推廣,成都外貿(mào)網(wǎng)站制作,鄲城網(wǎng)站建設(shè)費(fèi)用合理。
MapReduce采用分而治之的思想，把對(duì)大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個(gè)主節(jié)點(diǎn)管理下的各個(gè)分節(jié)點(diǎn)共同完成，然后通過(guò)整合各個(gè)節(jié)點(diǎn)的中間結(jié)果，得到最終結(jié)果。簡(jiǎn)單來(lái)說(shuō)，MapReduce就是“任務(wù)的分解和結(jié)果的匯總”。
在Hadoop中，用于執(zhí)行MapReduce任務(wù)的機(jī)器角色有兩個(gè)：一個(gè)是JobTracker；另一個(gè)是TaskTracker。JobTracker用于調(diào)度工作的，TaskTracker是用于執(zhí)行工作的。一個(gè)Hadoop集群中只有一臺(tái)JobTracker。
在分布式計(jì)算中，MapReduce框架負(fù)責(zé)處理了并行編程中分布式存儲(chǔ)、工作調(diào)度、負(fù)載均衡、容錯(cuò)均衡、容錯(cuò)處理以及網(wǎng)絡(luò)通信等復(fù)雜問(wèn)題，把處理過(guò)程高度抽象為兩個(gè)函數(shù)：map和reduce，map負(fù)責(zé)把任務(wù)分解成多個(gè)任務(wù)，reduce負(fù)責(zé)把分解后多任務(wù)處理的結(jié)果匯總起來(lái)。
需要注意的是，用MapReduce來(lái)處理的數(shù)據(jù)集（或任務(wù)）必須具備這樣的特點(diǎn)：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個(gè)小數(shù)據(jù)集都可以完全并行地進(jìn)行處理。
程序使用的測(cè)試文本數(shù)據(jù)：
Dear River
Dear River Bear Spark
Car Dear Car Bear Car
Dear Car River Car
Spark Spark Dear Spark 1編寫(xiě)主要類
（1）Maper類
首先是自定義的Maper類代碼
public class WordCountMap extends Mapper {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//fields:代表著文本一行的的數(shù)據(jù): dear bear river
String[] words = value.toString().split("\t");
for (String word : words) {
// 每個(gè)單詞出現(xiàn)１次，作為中間結(jié)果輸出
context.write(new Text(word), new IntWritable(1));
}
}
} ?????這個(gè)Map類是一個(gè)泛型類型，它有四個(gè)形參類型，分別指定map()函數(shù)的輸入鍵、輸入值、輸出鍵和輸出值的類型。LongWritable：輸入鍵類型，Text：輸入值類型，Text:輸出鍵類型，IntWritable：輸出值類型.
?????String[] words = value.toString().split("\t");,words 的值為Dear River Bear River
?????輸入鍵key是一個(gè)長(zhǎng)整數(shù)偏移量，用來(lái)尋找第一行的數(shù)據(jù)和下一行的數(shù)據(jù)，輸入值是一行文本Dear River Bear River，輸出鍵是單詞Bear ，輸出值是整數(shù)1。
?????Hadoop本身提供了一套可優(yōu)化網(wǎng)絡(luò)序列化傳輸?shù)幕绢愋?，而不直接使用Java內(nèi)嵌的類型。這些類型都在org.apache.hadoop.io包中。這里使用LongWritable類型(相當(dāng)于Java的Long類型)、Text類型(相當(dāng)于Java中的String類型)和IntWritable類型(相當(dāng)于Java的Integer類型)。
?????map()方法的參數(shù)是輸入鍵和輸入值。以本程序?yàn)槔?，輸入鍵LongWritable key是一個(gè)偏移量，輸入值Text value是Dear Car Bear Car ，我們首先將包含有一行輸入的Text值轉(zhuǎn)換成Java的String類型，之后使用substring()方法提取我們感興趣的列。map()方法還提供了Context實(shí)例用于輸出內(nèi)容的寫(xiě)入。
（2）Reducer類
public class WordCountReduce extends Reducer {
/*
(River, 1)
(River, 1)
(River, 1)
(Spark , 1)
(Spark , 1)
(Spark , 1)
(Spark , 1)
key: River
value: List(1, 1, 1)
key: Spark
value: List(1, 1, 1,1)
*/
public void reduce(Text key, Iterable values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable count : values) {
sum += count.get();
}
context.write(key, new IntWritable(sum));// 輸出最終結(jié)果
};
} Reduce任務(wù)最初按照分區(qū)號(hào)從Map端抓取數(shù)據(jù)為：
(River, 1)
(River, 1)
(River, 1)
(spark, 1)
(Spark , 1)
(Spark , 1)
(Spark , 1)
經(jīng)過(guò)處理后得到的結(jié)果為：
key: hello value: List(1, 1, 1)
key: spark value: List(1, 1, 1,1)
所以reduce()函數(shù)的形參 Iterable<IntWritable> values 接收到的值為List(1, 1, 1)和List(1, 1, 1,1)
（3）Main函數(shù)
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WordCountMain {
//若在IDEA中本地執(zhí)行MR程序，需要將mapred-site.xml中的mapreduce.framework.name值修改成local
public static void main(String[] args) throws IOException,
ClassNotFoundException, InterruptedException {
if (args.length != 2 || args == null) {
System.out.println("please input Path!");
System.exit(0);
}
//System.setProperty("HADOOP_USER_NAME","hadoop2.7");
Configuration configuration = new Configuration();
//configuration.set("mapreduce.job.jar","/home/bruce/project/kkbhdp01/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");
//調(diào)用getInstance方法，生成job實(shí)例
Job job = Job.getInstance(configuration, WordCountMain.class.getSimpleName());
// 打jar包
job.setJarByClass(WordCountMain.class);
// 通過(guò)job設(shè)置輸入/輸出格式
// MR的默認(rèn)輸入格式是TextInputFormat，所以下兩行可以注釋掉
// job.setInputFormatClass(TextInputFormat.class);
// job.setOutputFormatClass(TextOutputFormat.class);
// 設(shè)置輸入/輸出路徑
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 設(shè)置處理Map/Reduce階段的類
job.setMapperClass(WordCountMap.class);
//map combine減少網(wǎng)路傳出量
job.setCombinerClass(WordCountReduce.class);
job.setReducerClass(WordCountReduce.class);
//如果map、reduce的輸出的kv對(duì)類型一致，直接設(shè)置reduce的輸出的kv對(duì)就行；如果不一樣，需要分別設(shè)置map, reduce的 輸出的kv類型
//job.setMapOutputKeyClass(.class)
// job.setMapOutputKeyClass(Text.class);
// job.setMapOutputValueClass(IntWritable.class);
// 設(shè)置reduce task最終輸出key/value的類型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 提交作業(yè)
job.waitForCompletion(true);
}
}2本地運(yùn)行
首先更改mapred-site.xml文件配置
將mapreduce.framework.name的值設(shè)置為local
然后本地運(yùn)行：
查看結(jié)果：
3集群運(yùn)行
方式一：
首先打包
更改配置文件，改成yarn模式
添加本地jar包位置：
Configuration configuration = new Configuration();
configuration.set("mapreduce.job.jar","C:\\Users\\tanglei1\\IdeaProjects\\Hadooptang\\target");
設(shè)置允許跨平臺(tái)遠(yuǎn)程調(diào)用：
configuration.set("mapreduce.app-submission.cross-platform","true");
修改輸入?yún)?shù)：
運(yùn)行結(jié)果：
方式二：
將maven項(xiàng)目打包，在服務(wù)器端用命令運(yùn)行mr程序
hadoop jar com.kaikeba.hadoop-1.0-SNAPSHOT.jar
com.kaikeba.hadoop.wordcount.WordCountMain /tttt.txt /wordcount11網(wǎng)站題目：如何用Mapreduce程序完成wordcount
文章轉(zhuǎn)載：http://www.dlmjj.cn/article/pdpigj.html
