大数据大数据，机器学习，人工智能hadoop

《十小时入门大数据》学习笔记之Hadoop核心组件MapRedu

2018-12-19  本文已影响8人  腊月的梅花
目录

1. MapReduce概述

2. MapReduce编程模型

2.1 案例分析：WordCount => 词频统计

(统计文件中每个单词出现的次数，工作场景中的很多开发都是在WordCount的基础上进行改造的,因此深入理解该案例很重要.)

2.2 执行步骤

3. MapReduce架构

3.1 MapReduce1.x的架构
图片.png 1) JobTracker：JT (管理者)
3.2 MapReduce2.x架构

MapReduce2.x架构在 《十小时入门大数据》学习笔记之Hadoop核心组件YARN有详细的介绍，这里就不再赘述。

MapReduce2.x架构

4. MapReduce编程

4.1 核心概念
图片.png
4.2 编程实现WordCount
4.2.1 WC开发的流程

一共分为三步：
1）Map：读取输入的文件
2）Reduce：进行归并
3）Main：定义Driver，封装了MapReduce作业的所有信息

4.2.2 在Hadoop上运行自己编写的WC代码

1）利用maven进行编译打包：进入到项目的路径下，在终端输入下面代码

mvn clean package -DskipTests
出现HUILD SUCCES表示编译成功 图片.png

2）上传到服务器：同样在终端输入下面命令

scp target/hadoop-train-1.0.jar  hadoop@dell:~/data

3）运行：在服务器终端输入下面命令，其中hdfs://localhost:8020/hello.txt是文件的输入路径，hdfs://localhost:8020/output/wc是输出路径。

hadoop jar /home/hadoop/data/hadoop-train-1.0.jar com.lyh.hadoop.mapreduce.WordCountApp hdfs://localhost:8020/hello.txt hdfs://localhost:8020/output/wc

打开浏览器：http://localhost:8088 ，能够看到有一个我们的wordcount任务正在运行

图片.png 运行结束后，查看hdfs的/output/wc/目录，我们能够看到多出两个文件 图片.png 查看part-r-00000文件，可以看到我们的计算结果 图片.png
4.2.3 存在的问题
4.2.4 解决办法
hadoop fs -rm -r /output/wc
hadoop jar /home/hadoop/data/hadoop-train-1.0.jar com.lyh.hadoop.mapreduce.WordCountApp hdfs://localhost:8020/hello.txt hdfs://localhost:8020/output/wc
图片.png

保存，给wc_mr.sh添加执行权限：chmod u+x wc_mr.sh

执行脚本： ./wc_mr.sh 可以看到输出的日志中有删除文件夹的操作 图片.png 
// 准备清理已存在的输出目录
        Path outputPath = new Path(args[1]);
        FileSystem fileSystem = FileSystem.get(configuration);
        if(fileSystem.exists(outputPath)){
            fileSystem.delete(outputPath, true);
            System.out.println("output file exists, but is has deleted");
        }
图片.png
4.3 MapReduce编程之Combiner
 //通过job设置combiner处理类，其实逻辑上和我们的reduce是一模一样的
        job.setCombinerClass(MyReducer.class);
图片.png

适用场景：求和 次数等
不适用的场景：求平均数

4.4 MapReduce编程之Partitioner
xiaomi 100
huawei 600
iphone 200
xiaomi 700
xiaomi 200
huawei 800

要求统计各类手机的总销售量，分别存储在不同的文件中

5 JobHistory配置与启动停止

5.1 文件配置

1）在yarn-site.xml中添加

<!-- 开启日志聚合 -->
<property>
<name>yarn.log-aggregation-enable</name>
<value>true</value>
</property>

2）在mapred-site.xml中添加

<!-- 设置jobhistoryserver 没有配置的话 history入口不可用 --> 
<property> 
  <name>mapreduce.jobhistory.address</name> 
  <value>localhost:10020</value> 
</property> 
<!-- 配置web端口 --> 
<property>                                                          <name>mapreduce.jobhistory.webapp.address</name> 
  <value>localhost:19888</value> 
</property> 
<!-- 配置正在运行中的日志在hdfs上的存放路径 --> 
<property> 
    <name>mapreduce.jobhistory.intermediate-done-dir</name>
   <value>/history/done_intermediate</value> 
</property> 
<!-- 配置运行过的日志存放在hdfs上的存放路径 --> 
<property> 
   <name>mapreduce.jobhistory.done-dir</name> 
   <value>/history/done</value> 
</property>
5.2 启动与停止

1）启动： 在hadoop/sbin/目录下执行

./mr-jobhistory-daemon.sh start historyserver

2）停止：在hadoop/sbin/目录下执行

./mr-jobhistory-daemon.sh stop historyserver
6.基础代码

完整的项目见：https://download.csdn.net/download/qq_29557137/10862010

import org.apache.hadoop.conf.Configuration;
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.output.FileOutputFormat;

import java.io.IOException;

/**
 * 使用MapReduce开发WordCount应用程序
 */
public class WordCountApp {
    /**
     * Map：读取输入的文件,注意这里的Mapper<LongWritable, Text, Text, LongWritable>类型需要自己定义，因为我们的第一个<k1,v1>是文件的偏移量和字符串，因此k1的类型设置为LongWritable，Text，输出的是单词和单词出现的次数，因此<k2,v2>的类型设置为Text 和 LongWritable
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{

        LongWritable one = new LongWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            // 接收到的每一行数据
            String line = value.toString();

            //按照指定分隔符进行拆分
            String[] words = line.split(" ");

            for(String word :  words) {
                // 通过上下文把map的处理结果输出
                context.write(new Text(word), one);
            }

        }
    }

    /**
     * Reduce：归并操作
     */
    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

            long sum = 0;
            for(LongWritable value : values) {
                // 求key出现的次数总和
                sum += value.get();
            }
            // 最终统计结果的输出
            context.write(key, new LongWritable(sum));
        }
    }

    /**
     * 定义Driver：封装了MapReduce作业的所有信息
     */
    public static void main(String[] args) throws Exception{

        //创建Configuration
        Configuration configuration = new Configuration();

        //创建Job
        Job job = Job.getInstance(configuration, "wordcount");

        //设置job的处理类
        job.setJarByClass(WordCountApp.class);

        //设置作业处理的输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));

        //设置map相关参数
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置reduce相关参数
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //设置作业处理的输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
上一篇 下一篇

猜你喜欢

热点阅读