Mapreduce的简单实现和步骤

2023-05-16

package com.qfedu.bigdata.MR;

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;

/**

@ClassName wordCount
@Description TODO
@Author Chenfg
@Date 2018/9/20 0020 11:22
@Version 1.0
词频统计
输出数据：
word,n
Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

框架在调用咱们写的map方法时，会将数据作为参数（一个key，一个value）传递给map方法

KEYIN：是框架（maptask）要传递给map方法的输入参数中的key的数据类型

VALUEIN：是框架（maptask）要产地给map方法的输入参数中的value的数据类型

在默认情况下，框架传入的key是框架从待处理数据（文本文件）中读取到的‘某一行’数据的起始偏移量，所以类型Long

框架传入的value是框架从待处理数据中读到的‘某一行’的内容，所以类型是String

但是，Long或者String等java的原生态数据类型的序列化的效率较低，所以hadoop对其进行了封装改造，

```
有替代品：LongWriable/Text
```

map方法处理完数据后需要返回一个结果（一个key一个value的键值对数据）

KEYOUT：是咱们的map方法处理完成后返回结果中的key的数据类型

VALUEOUT：是咱们的map方法处理完成后返回结果中的value的数据类型

```
思路：
```
```
map阶段的处理逻辑：
```
```
1、用空格切分单词
```
```
2、循环遍历单词
```
```
3、输出结果，<word,1>
```
```
    map方法的调用规律：
```

    maptask没读取到一行数据就会调用此意map方法

*/
public class wordCount {
static class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//1、数据类型转换
String line = value.toString();

        //2、切分数据
        String[] words = line.split(" ");

        //3、循环遍历单词
        for (String word : words) {
            //4、输出结果
            context.write(new Text(word),new LongWritable(1));
        }
    }
}

/**
 * Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 *     reduce()方法要接收的输入参数是一个key一个迭代器<T>的values
 *     KEYIN：框架（reducetask）要传递给reduce方法的输入参数的key的数据类型
 *     VALUEIN：框架（reducetask）要传递给reduce方法的输入参数的value的数据类型
 *
 *
 *      KEYOUT：reduce方法处理后的数据的返回结果的key的数据类型
 *      VALUEOUT：reduce方法处理后的数据的返回结果的value的数据类型
 *
 *     reduce方法调用的规律：框架会从map阶段的输出结果中找出所有的key相同的<k,v>数据对组成一组数据，
 *     然后调用一次reduce()方法
 */
static class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        //为什么用迭代器
        /**
         * map端的输出数据：
         <a,1>
         <b,1>
         <c,1>
         <a,1>
         <a,1>
         <d,1>

         reduce端的输入数据：
         <a,<1,1,1>>
         */

        int count=0;
        //循环遍历迭代器
        for (LongWritable i :values) {
            count+=i.get();
        }

        //往下发送数据
        context.write(key,new LongWritable(count));
    }
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    //1、创建一个Configuration配置项
    Configuration conf = new Configuration();

    //2、配置连接参数
    conf.set("fs.defaultFS","hdfs://qianfeng");

    //高可用集群的设置项
    conf.set("fs.defaultFS","hdfs://qianfeng");
    conf.set("dfs.nameservices","qianfeng");
    conf.set("dfs.ha.namenodes.qianfeng","nn1,nn2");
    conf.set("dfs.namenode.rpc-address.qianfeng.nn1","hadoop01:9000");
    conf.set("dfs.namenode.rpc-address.qianfeng.nn2","hadoop02:9000");
    conf.set("dfs.client.failover.proxy.provider.qianfeng","org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

    //3、创建一个job对象
    Job job = Job.getInstance(conf,"wordCount");

    //4、描述对象
    //5、设置Job的执行路径
    job.setJarByClass(wordCount.class);

    //6、设置mapTask调用的业务逻辑类
    job.setMapperClass(WCMapper.class);

    //7、设置map端数据输出的类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(LongWritable.class);

    //8、设置mapTask调用的业务类
    job.setReducerClass(WCReducer.class);

    //9、设置reduce端的数据的输出类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);

    //10、设置Job的输入文件的路径
    FileInputFormat.setInputPaths(job,new Path(args[0]));

    //11、设置Job的输出文件的路径
    FileOutputFormat.setOutputPath(job,new Path(args[1]));

    //12、提交job

// job.submit();
boolean b = job.waitForCompletion(true);

    System.exit(b?0:1);
}package com.qfedu.bigdata.MR;

import java.io.IOException;

/**

@ClassName wordCount
@Description TODO
@Author Chenfg
@Date 2018/9/20 0020 11:22
@Version 1.0
词频统计
输出数据：
word,n
Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

框架在调用咱们写的map方法时，会将数据作为参数（一个key，一个value）传递给map方法

KEYIN：是框架（maptask）要传递给map方法的输入参数中的key的数据类型

VALUEIN：是框架（maptask）要产地给map方法的输入参数中的value的数据类型

在默认情况下，框架传入的key是框架从待处理数据（文本文件）中读取到的‘某一行’数据的起始偏移量，所以类型Long

框架传入的value是框架从待处理数据中读到的‘某一行’的内容，所以类型是String

但是，Long或者String等java的原生态数据类型的序列化的效率较低，所以hadoop对其进行了封装改造，

```
有替代品：LongWriable/Text
```

map方法处理完数据后需要返回一个结果（一个key一个value的键值对数据）

KEYOUT：是咱们的map方法处理完成后返回结果中的key的数据类型

VALUEOUT：是咱们的map方法处理完成后返回结果中的value的数据类型

```
思路：
```
```
map阶段的处理逻辑：
```
```
1、用空格切分单词
```
```
2、循环遍历单词
```
```
3、输出结果，<word,1>
```
```
    map方法的调用规律：
```

    maptask没读取到一行数据就会调用此意map方法

        //2、切分数据
        String[] words = line.split(" ");

        //3、循环遍历单词
        for (String word : words) {
            //4、输出结果
            context.write(new Text(word),new LongWritable(1));
        }
    }
}

/**
 * Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 *     reduce()方法要接收的输入参数是一个key一个迭代器<T>的values
 *     KEYIN：框架（reducetask）要传递给reduce方法的输入参数的key的数据类型
 *     VALUEIN：框架（reducetask）要传递给reduce方法的输入参数的value的数据类型
 *
 *
 *      KEYOUT：reduce方法处理后的数据的返回结果的key的数据类型
 *      VALUEOUT：reduce方法处理后的数据的返回结果的value的数据类型
 *
 *     reduce方法调用的规律：框架会从map阶段的输出结果中找出所有的key相同的<k,v>数据对组成一组数据，
 *     然后调用一次reduce()方法
 */
static class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        //为什么用迭代器
        /**
         * map端的输出数据：
         <a,1>
         <b,1>
         <c,1>
         <a,1>
         <a,1>
         <d,1>

         reduce端的输入数据：
         <a,<1,1,1>>
         */

        int count=0;
        //循环遍历迭代器
        for (LongWritable i :values) {
            count+=i.get();
        }

        //往下发送数据
        context.write(key,new LongWritable(count));
    }
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    //1、创建一个Configuration配置项
    Configuration conf = new Configuration();

    //2、配置连接参数
    conf.set("fs.defaultFS","hdfs://qianfeng");

    //高可用集群的设置项
    conf.set("fs.defaultFS","hdfs://qianfeng");
    conf.set("dfs.nameservices","qianfeng");
    conf.set("dfs.ha.namenodes.qianfeng","nn1,nn2");
    conf.set("dfs.namenode.rpc-address.qianfeng.nn1","hadoop01:9000");
    conf.set("dfs.namenode.rpc-address.qianfeng.nn2","hadoop02:9000");
    conf.set("dfs.client.failover.proxy.provider.qianfeng","org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

    //3、创建一个job对象
    Job job = Job.getInstance(conf,"wordCount");

    //4、描述对象
    //5、设置Job的执行路径
    job.setJarByClass(wordCount.class);

    //6、设置mapTask调用的业务逻辑类
    job.setMapperClass(WCMapper.class);

    //7、设置map端数据输出的类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(LongWritable.class);

    //8、设置mapTask调用的业务类
    job.setReducerClass(WCReducer.class);

    //9、设置reduce端的数据的输出类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);

    //10、设置Job的输入文件的路径
    FileInputFormat.setInputPaths(job,new Path(args[0]));

    //11、设置Job的输出文件的路径
    FileOutputFormat.setOutputPath(job,new Path(args[1]));

    //12、提交job

// job.submit();
boolean b = job.waitForCompletion(true);

    System.exit(b?0:1);
}

}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Mapreduce的简单实现和步骤的相关文章

【无人机】【2019】无人机巡线路由算法研究

本文为美国密苏里理工大学 xff08 作者 xff1a YU LI xff09 的硕士论文 xff0c 共79页随着机器人技术的迅速发展 xff0c 越来越多的公用事业公司使用机器人来执行各种任务无人机是一种高效的高压输电线路巡检机器人
【电力电子】【2010.09】无人机系统的自适应控制

四旋翼直升无人机本文为美国麻省理工学院 xff08 作者 xff1a Zachary Thompson Dydek xff09 的博士学位论文 xff0c 共139页自适应控制被认为是未来高性能关键安全系统 xff08 如高超声速飞行
【新书推荐】【2018.11】无人机多传感器导航的非线性卡尔曼滤波

2018 11 无人机多传感器导航的非线性卡尔曼滤波Nonlinear Kalman Filter for Multi Sensor Navigation of Unmanned Aerial Vehicles xff0c 共254页如果
【源码】轻松实现四轴飞行器的模拟与控制

本模型取自题为 Quadcopter Simulation and Control Made Easy 的网络研讨会 xff0c 您可以在这里找到相关视频内容 xff1a https www mathworks com videos qua
【无人机】【2014.08】无人机循环路由

本文为以色列理工学院 xff08 作者 xff1a Nir Drucker xff09 的硕士论文 xff0c 共65页许多无人驾驶飞行器 xff08 UAV xff09 针对的国防和民用相关任务涉及在各种时间限制下监测预先确定的一组地面
学习【gRPC C++简单示例及代码】

文章目录 0 启动程序1 未作修改部分1 1 demo proto 2 添加注释部分2 1 server cc2 2 client cc 3 重点修改部分3 1 CMakeLists txt原始代码3 2 CMakeLists txt精简修
利用Gitee搭建私人仓库-https篇

利用Gitee搭建私人仓库 https篇第一步 xff1a 注册Gitee账号https gitee com 第二步 xff1a 创建私人仓库安装git 第一步 xff1a 下载git git下载地址第二步 xff1a 按照默认安装就好
嵌入式开发必看：PCB设计的这几个问题，你能明白了吗？

PCB设计布局 xff0c 是把电路器件放在印制电路板布线区内布局是否合理不仅影响后面的布线工作 xff0c 而且对整个电路板的性能也有重要影响在保证电路功能和性能指标后 xff0c 要满足工艺性检测和维修方面的要求 xff0c 元件
STM32实例-蜂鸣器实验

本章我们通过另外一个实验来讲述 STM32 IO 口的输出通过 STM32F1 芯片的一个 IO口控制板载有源无源蜂鸣器 xff0c 实现蜂鸣器控制蜂鸣器介绍蜂鸣器是一种一体化结构的电子讯响器 xff0c 采用直流电压供电 xff0
PCM分析及音量控制

PCM音量控制本文转自 xff1a http blog jianchihu net pcm volume control html 一声音的相关概念声音是介质振动在听觉系统中产生的反应声音总可以被分解为不同频率不同强度正弦波的叠加
总结大佬经验，如何学习STM32?(入门、进阶)

大佬经验一想快速入门进阶 xff0c 面向工作时先把工作中需要的东西学会 xff0c 一项一项的来 xff0c 比如先把GPIO弄清楚 xff0c 怎么初始化 xff0c 怎么输入输出 xff0c 然后定时器串口 ADC DAC D
VTK和ITK中的坐标系统

当我们在处理医学图像和应用时 xff0c 一个问题就是坐标系统之间的不同在图像应用中通常有三个坐标系 xff0c 分别是世界坐标系 xff0c 解剖坐标系和图像坐标系每种坐标系目的不同 xff0c 呈现数据的方式也不一样解剖坐标系由三
Keil MDK-ARM下载安装与和谐教程

Keil MDK ARM是编写调试ARM的不二工具 xff0c 普及度极高一下载方法一 xff0c Keil官网下载 xff1a https www keil com download product 点击MDK Arm后进入软件下载
C语言str族函数strpbrk()函数介绍及其算法实现

一 strpbrk 函数介绍在C语言中 xff0c 其函数原型格式一般为 xff1a span class token keyword char span span class token operator span span class
JAVA学习笔记037——接口和实现类的概念

Java中的接口 xff08 interface 使抽象类的概念更深入一层接口 interface 是抽象方法和常量值的定义的集合从本质上讲 xff0c 接口是一种特殊的抽象类 xff0c 这种抽象类中只包含常量和方法的定义 xff0c
springboot2.x The character [_] is never valid in a domain name

访问springcloud工程 xff0c springboot版本为2 1 15 xff0c 使用域名访问接口报400 但是直接用ip不报错发现是集成的tomcat版本不支持域名下划线大约是8 5 31以后的版本不支持带下划线的域名
DynamicDataSource 数据源连接混乱

DynamicDataSource 多数据源配置 xff0c 不多说了 xff0c 初始化几个数据源 xff0c 继承AbstractRoutingDataSource重写里面的determineCurrentLookupKey 方法但是
linux上tomcat设置各种jvm参数

bin catalina sh 最上面一行添加如下 JAVA OPTS 61 34 J A V A O P
BeanUtils.copyProperties()可以把两个拥有相同属性的实体类进行转化。减去了逐个属性赋值的麻烦。

一简介 BeanUtils提供对Java反射和自省API的包装其主要目的是利用反射机制对JavaBean的属性进行处理我们知道 xff0c 一个JavaBean通常包含了大量的属性 xff0c 很多情况下 xff0c 对JavaBea
启动spring boot 项目时报错：Could not resolve　type alias 'VARCHAR'

今天再项目里添加新功能时 xff0c 突然爆出 Cause org apache ibatis type TypeException Could not resolve type alias 39 VARCHAR 39 Cause java

随机推荐

TOP to Down设计简单例子 Creo3.0

1 打开Creo3 0 xff0c 新建装配 xff1a 2 点击模型创建 xff0c 创建骨架模型 3 点击创建子装配体 xff1b 用户定义默认 4 打开创建的子装配体 xff0c 创建零件 xff1b 约束默认 5 打开骨架零件
java.lang.ClassCastException: org.jetbrains.jps.builders.java.dependencyView.TypeRepr$ArrayType....

运行idea启动项目突然报错 xff0c 之前还可以启动 xff0c 没有任何更改 xff0c 突然报错 xff1a java lang ClassCastException org jetbrains jps builders java
feign使用熔断器，直接熔断

项目中配置熔断器 xff0c 开启之后直接熔断走fallback xff0c 明明下游服务可用 xff01 xff01 xff01 xff01 各种debug xff0c 看源码 xff0c 差资料 xff0c 无解 debug只是显示阻塞
查看docker容器占用内存

ps ef grep 容器Id root 64 wentao 2 order ps ef grep 3a61cb3fd4f6 root 7358 12956 0 09 14 00 00 00 containerd shim namespac
docker tag之后会有两不同名称的相同镜像并且镜像id也是一样的，如何删除

如下 xff1a docker rmi 10 144 71 78 5000 centos7 jdk8 cn v1
springboot拦截获取controller返回值

64 ControllerAdvice public class ResponseBodyAnalysis implements ResponseBodyAdvice 64 Override public Object beforeBody
springsecurity或者oauth2中设置某个开头的路径拦截，并且放行某个子路径

64 Override public void configure HttpSecurity http throws Exception http csrf disable authorizeRequests antMatchers 34
工具类:把一个对象转换成hashmap类型

public static Map lt String Object gt objectToMap Object obj if obj 61 61 null return null Map lt String Object gt map 6
zookeeper 启动失败错误: 找不到或无法加载主类 org.apache.zookeeper.server.quorum.QuorumPeerMain

问题描述在服务器上安装完 zookeeper后 xff0c 启动zk服务报错原因分析 xff1a 于是乎 xff0c 去查看了 zk的日志文件 xff0c 发现了报错信息啥 xff1f 找不到启动类 xff01 xff01 xff0
Gazebo7 无法加载模型问题解决（黑屏）

Gazebo7 无法加载模型 Gazebo7 无法加载模型问题解决 xff08 黑屏 xff09 打开gazebo的时候会发现一直处于这种状态 xff0c 这是因为model库加载不正确导致的解决方法 xff1a 通过直接下载所有模型到用
Factor Graph, 因子图

引言因子图方法广泛应用于机器人姿态估计 xff0c 多种信号融合领域 xff0c 在机器视觉无人机导航无人驾驶领域应用广泛 xff0c 与传统卡尔曼滤波方式相比 xff0c 能提供更高精度 xff0c 更平滑的全局结果因子图本质上是
Hexo 最常用的几个命令

Hexo 约有二十个命令 xff0c 但普通用户经常使用的大概只有下列几个 hexo s hexo s 启动本地服务器 xff0c 用于预览主题默认地址 xff1a http localhost 4000 hexo s 是 hexo se
ESP32 CAM与服务器（python）TCP视频传输

ESP32 CAM 代码基于Arduino实现网络调试助手 https soft 3dmgame com down 213757 html include lt Arduino h gt include lt WiFi h gt inc
ESP32 CAM与服务器（python）UDP视频传输

ESP32 CAM Arduino代码 include 34 esp camera h 34 include lt WiFi h gt include 34 AsyncUDP h 34 include lt vector gt const
STM32F103C8T6开发板+MPU6050刷四轴飞控

下载betaflight NAZE固件 xff1a https github com betaflight betaflight releases tag v3 2 5 安装betaflight Configurator调参软件 STM32
js-Map和Set与Array互转

一 xff1a Map转Array span class token builtin class name let span map span class token operator 61 span new Map span class
STM32通用定时器之输出比较模式与PWM模式

STM32通用定时器之输出比较与PWM 通用定时器其实就两个基本的功能 xff1a 输入输出主要介绍一下输出英文手册是这么说的 xff1a Input capture Output compare PWM generation Edg
Linux bridge table(brctl)

目录一 brctl简介二常用命令 2 1 查看所有网桥信息 2 2 查看指定网桥信息 2 3 新建网桥 2 4 启用停用网桥 2 5 添加网桥端口 2 6 移除网桥端口 2 7 删除网桥需先移除相关端口三实验案例 3 1 测试
idea 2019- 3.3版本得控制台中文乱码问题（本人遇到的所有解决办法都在其中，如有缺漏希望大神查缺补漏）

一 File gt Settings gt File Encoding Global Encoding Project Encoding Default encodeing for properties files这三个位置都设置成UTF
Mapreduce的简单实现和步骤

package com qfedu bigdata MR import org apache hadoop conf Configuration import org apache hadoop fs Path import org apa

Mapreduce的简单实现和步骤

Mapreduce的简单实现和步骤 的相关文章

随机推荐

热门标签

Mapreduce的简单实现和步骤的相关文章