当前位置: 首页 > news >正文

Hadoop3教程(十八):MapReduce之MapJoin案例分析

文章目录

  • (118)MapJoin案例需求分析
    • ReduceJoin的问题
    • 如何解决ReduceJoin的问题
    • 如何将一个文件主动缓存到集群的内存里
  • (119)MapJoin案例代码实现
  • 参考文献

(118)MapJoin案例需求分析

ReduceJoin的问题

在ReduceJoin中,合并的操作是在Reduce阶段进行的,所以相比Map阶段,Reduce阶段的处理压力过大。另外,相同的产品ID的数据会进入同一个Reducer中,如果这个产品ID下数据过多,其他产品ID的数据很少,那么会导致前面那个Reducer压力过大,这就是数据倾斜问题。

如何解决ReduceJoin的问题

那如何解决这种问题呢?

比较好的方法是不使用ReduceJoin,使用MapJoin,即在Map阶段实现拼接。

思路简单来说,就是将产品码表放进内存,orders.txt正常切片进入mapper,然后mapper处理的时候,就逐行对orders.txt里的数据进行产品码值的替换。

基于这种方式,MapJoin的适用场景也就很明显了,MapJoin适用于一张或多张表特别小(不能把内存撑爆了),一张表特别大的场景

如何将一个文件主动缓存到集群的内存里

那问题来了,在Hadoop里怎么把一张表主动缓存到内存当中,且还能在map()里调用呢?

首先我们需要在驱动类里,指定将文件加载到缓存:

//缓存普通文件到Task运行节点。
job.addCacheFile(new URI("file:///e:/cache/pd.txt"));
//如果是集群运行,需要设置HDFS路径
job.addCacheFile(new URI("hdfs://hadoop102:8020/cache/pd.txt"));// MapJoin的话就不需要Reduce阶段了
job.setNumReduceTasks(0);

然后在自定义Mapper类的setup()里,按以下流程编写代码,以读取缓存的文件数据:

//1. 获取缓存的文件;
// 2.循环读取缓存文件中每一行;
// 3. 切割;
// 4. 缓存数据到集合;

setup()执行完成后,才会执行map()

所以我们最后在map()里,获取一行后,截取到pid,从内存中码表拿到产品中文名,拼接给出就可以。

(119)MapJoin案例代码实现

过了一遍教程,其实就是对上一小节的代码实现。

总的来说,就是只有一个Map阶段,在Map阶段中,在map()处理之前,先把码表读进内存中,然后map()在一行一行读取后,直接使用内存中的码表对指定字段进行替换即可。

对我来讲用处不大,所以这里直接跳过,但还是补充一下代码:

在MapJoinDriver驱动类中添加缓存文件:

package com.atguigu.mapreduce.mapjoin;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.net.URI;
import java.net.URISyntaxException;public class MapJoinDriver {public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {// 1 获取job信息Configuration conf = new Configuration();Job job = Job.getInstance(conf);// 2 设置加载jar包路径job.setJarByClass(MapJoinDriver.class);// 3 关联mapperjob.setMapperClass(MapJoinMapper.class);// 4 设置Map输出KV类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);// 5 设置最终输出KV类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);// 加载缓存数据job.addCacheFile(new URI("file:///D:/input/tablecache/pd.txt"));// Map端Join的逻辑不需要Reduce阶段,设置reduceTask数量为0job.setNumReduceTasks(0);// 6 设置输入输出路径FileInputFormat.setInputPaths(job, new Path("D:\\input"));FileOutputFormat.setOutputPath(job, new Path("D:\\output"));// 7 提交boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}
}

在MapJoinMapper类中的setup方法中读取缓存文件,并在map()里进行替换:

package com.atguigu.mapreduce.mapjoin;import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {private Map<String, String> pdMap = new HashMap<>();private Text text = new Text();//任务开始前将pd数据缓存进pdMap@Overrideprotected void setup(Context context) throws IOException, InterruptedException {//通过缓存文件得到小表数据pd.txtURI[] cacheFiles = context.getCacheFiles();Path path = new Path(cacheFiles[0]);//获取文件系统对象,并开流FileSystem fs = FileSystem.get(context.getConfiguration());FSDataInputStream fis = fs.open(path);//通过包装流转换为reader,方便按行读取BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "UTF-8"));//逐行读取,按行处理String line;while (StringUtils.isNotEmpty(line = reader.readLine())) {//切割一行    
//01	小米String[] split = line.split("\t");pdMap.put(split[0], split[1]);}//关流IOUtils.closeStream(reader);}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//读取大表数据    
//1001	01	1String[] fields = value.toString().split("\t");//通过大表每行数据的pid,去pdMap里面取出pnameString pname = pdMap.get(fields[1]);//将大表每行数据的pid替换为pnametext.set(fields[0] + "\t" + pname + "\t" + fields[2]);//写出context.write(text,NullWritable.get());}
}

参考文献

  1. 【尚硅谷大数据Hadoop教程,hadoop3.x搭建到集群调优,百万播放】

相关文章:

Hadoop3教程(十八):MapReduce之MapJoin案例分析

文章目录 &#xff08;118&#xff09;MapJoin案例需求分析ReduceJoin的问题如何解决ReduceJoin的问题如何将一个文件主动缓存到集群的内存里 &#xff08;119&#xff09;MapJoin案例代码实现参考文献 &#xff08;118&#xff09;MapJoin案例需求分析 ReduceJoin的问题 在R…...

SOAR安全事件编排自动化响应-安全运营实战

SOAR是最近几年安全市场上最火热的词汇之一。各个安全产商都先后推出了相应的产品&#xff0c;但大部分都用得不是很理想。SOAR不同与传统的安全设备&#xff0c;买来后实施部署就完事&#xff0c;SOAR是一个安全运营系统&#xff0c;是实现安全运营过程中人、工具、流程的有效…...

连锁药店的自有品牌之争:老百姓大药房能否突围?

文丨新熔财经 作者丨楷楷 近年来&#xff0c;随着医保谈判药品的“双通道”&#xff08;即消费者可在有资质的药店买到新进医保的创新药&#xff09;&#xff0c;以及“门诊统筹”将药店纳入医保报销等医改政策出台&#xff0c;药企开始重新重视起零售药店渠道&#xff0c;很…...

智能台灯语音控制丨解放双手

台灯是日常生活中一种常见的照明产品。以往的台灯大多都是采取手动控制&#xff0c;通过按键去对台灯的亮度进行调整。随着科技的发展&#xff0c;台灯也开始走向了智能化。人们已经能够对智能台灯进行语音控制&#xff0c;通过调节灯光开关、色温、灯光亮度等操作&#xff0c;…...

网络库OKHTTP(2)面试题

序、慢慢来才是最快的方法。 背景 OkHttp 是一套处理 HTTP 网络请求的依赖库&#xff0c;由 Square 公司设计研发并开源&#xff0c;目前可以在 Java 和 Kotlin 中使用。对于 Android App 来说&#xff0c;OkHttp 现在几乎已经占据了所有的网络请求操作。 OKHttp源码官网 问1…...

探索Java NIO:究竟在哪些领域能大显身手?揭秘原理、应用场景与官方示例代码

一、NIO简介 Java NIO&#xff08;New IO&#xff09;是Java SE 1.4引入的一个新的IO API&#xff0c;它提供了比传统IO更高效、更灵活的IO操作。与传统IO相比&#xff0c;Java NIO的优势在于它支持非阻塞IO和选择器&#xff08;Selector&#xff09;等特性&#xff0c;能够更…...

论文阅读 Memory Enhanced Global-Local Aggregation for Video Object Detection

Memory Enhanced Global-Local Aggregation for Video Object Detection Abstract 人类如何识别视频中的物体&#xff1f;由于单一帧的质量低下&#xff0c;仅仅利用一帧图像内的信息可能很难让人们在这一帧中识别被遮挡的物体。我们认为人们识别视频中的物体有两个重要线索&…...

Java 常用类(包装类)

目录 八大Wrapper类包装类的分类 装箱和拆箱包装类和基本数据类型之间的转换常见面试题 包装类方法包装类型和String类型的相互转换包装类常用方法&#xff08;以Integer类和Character类为例&#xff09;Integer类和Character类的常用方法 Integer创建机制&#xff08;面试题&a…...

ES|QL:Elasticsearch的 新一代查询语言

作者&#xff1a;李捷 “学会选择很难。学会正确选择更难。而在一个充满无限可能的世界里学会正确选择则更难&#xff0c;也许是太难了。” 巴里-施瓦茨&#xff08;Barry Schwartz&#xff09;在《选择的悖论--多就是少》&#xff08;The Paradox of Choice -More is Less&…...

C语言实现句子中的单词颠倒排序

一、运行结果 二、源代码 # define _CRT_SECURE_NO_WARNINGS # include <stdio.h> # include <assert.h>//实现逆转函数&#xff1b; void reverse(char* left, char* right) {//断言left和right都不能为空&#xff1b;assert(left);assert(right);//循环逆转字母…...

MySQL学习(八)——锁

文章目录 1. 锁概述2. 全局锁2.1 全局锁的必要性2.2 语法2.3 全局锁的特点 3. 表级锁3.1 表锁3.2 元数据锁3.3 意向锁3.4 自增锁 4. 行级锁4.1 介绍4.2 记录锁4.3 间隙锁4.4 临键锁 1. 锁概述 锁是计算机协调多个进程或线程并发访问某一资源的机制。在数据库中&#xff0c;除传…...

让iPhone用电脑的网络上网

让iPhone用电脑的网络上网&#xff0c;可以按照以下步骤操作&#xff1a; 在iPhone上找到并点击“设置”选项&#xff0c;进入“蜂窝移动网络”。打开“个人热点”选项。此时下方的弹出对话框会显示“仅USB”。用数据线将你的iPhone与电脑相连&#xff0c;并在电脑上打开“控制…...

ThreeJS-3D教学十-有宽度的line

webgl中线是没有宽度的&#xff0c;现实的应用中一般做法都是将线拓宽成面来绘制。默认threejs的线宽是无法调节的&#xff0c;需要用有厚度的线 THREE.Line2。 先看效果图&#xff1a; 看下代码&#xff1a; <!DOCTYPE html> <html lang"en"> <he…...

安装Elasticsearch步骤(包含遇到的问题及解决方案)

注&#xff1a;笔者是在centos云服务器环境下安装的Elasticsearch 目录 1.安装前准备 2.下载Elasticsearch 3.启动Elasticsearch 非常容易出问题 第一次运行时&#xff0c;可能出现如下错误&#xff1a; 一、内存不足原因启动失败 二、使用root用户启动问题 三、启动ES自…...

网络编程面试笔试真题

网络编程笔试面试真题 1、关于Linux系统中多线程的信号处理&#xff0c;说法中不正确的是&#xff1f; A&#xff1a;在线程环境霞&#xff0c;产生的信号是传递给整个进程的 B&#xff1a;一般情况下&#xff0c;信号会随机给进程的一个线程 C&#xff1a;对某个信号处理函数…...

MySQL官方文档如何查看,MySQL中文文档

这里写自定义目录标题 MySQL官方文档如何查看MySQL中文文档 MySQL官方文档如何查看 MySQL官网地址&#xff1a;https://dev.mysql.com/doc/ 比如这里我要找InnoDB架构 MySQL中文文档 MySQL 5.1中文文档地址&#xff1a;https://www.mysqlzh.com/...

第七章:最新版零基础学习 PYTHON 教程—Python 列表(第四节 -如何在 Python 中查找列表的长度)

列表是 Python 日常编程不可或缺的一部分,所有 Python 用户都必须学习,了解其实用程序和操作是必不可少的,而且总是有好处的。因此,本文讨论了找到第一个这样的实用程序。使用Python 的列表中的元素。 目录 在 Python 中查找列表的长度...

XPS虽没流行,但还在使用!在Windows 10中打开XPS文件的最佳方法

当Windows Vista发布时&#xff0c;微软推出了XPS格式&#xff0c;这是PDF的替代品。XPS文件格式并不是什么新鲜事&#xff0c;但从未获得过多大的吸引力。 因此&#xff0c;XPS&#xff08;XML Paper Specification&#xff09;文件是微软对Adobe PDF文件的竞争对手。尽管XPS…...

23 种设计模式详解(C#案例)

&#x1f680;设计模式简介 设计模式&#xff08;Design pattern&#xff09;代表了最佳的实践&#xff0c;通常被有经验的面向对象的软件开发人员所采用。设计模式是软件开发人员在软件开发过程中面临的一般问题的解决方案。这些解决方案是众多软件开发人员经过相当长的一段时…...

@SpringBootApplication配置了scanBasePackages导致请求一直404,分析下原因

出现RequestMapping注解的Controller类可能是因为SpringBootApplication注解中配置了scanBasePackages导致的请求一直返回404错误。 SpringBootApplication注解是Spring Boot的核心注解之一&#xff0c;它用于启动Spring Boot应用程序。这个注解实际上是一个组合注解&#xff…...

STM32F4基本定时器使用和原理详解

STM32F4基本定时器使用和原理详解 前言如何确定定时器挂载在哪条时钟线上配置及使用方法参数配置PrescalerCounter ModeCounter Periodauto-reload preloadTrigger Event Selection 中断配置生成的代码及使用方法初始化代码基本定时器触发DCA或者ADC的代码讲解中断代码定时启动…...

苍穹外卖--缓存菜品

1.问题说明 用户端小程序展示的菜品数据都是通过查询数据库获得&#xff0c;如果用户端访问量比较大&#xff0c;数据库访问压力随之增大 2.实现思路 通过Redis来缓存菜品数据&#xff0c;减少数据库查询操作。 缓存逻辑分析&#xff1a; ①每个分类下的菜品保持一份缓存数据…...

Cloudflare 从 Nginx 到 Pingora:性能、效率与安全的全面升级

在互联网的快速发展中&#xff0c;高性能、高效率和高安全性的网络服务成为了各大互联网基础设施提供商的核心追求。Cloudflare 作为全球领先的互联网安全和基础设施公司&#xff0c;近期做出了一个重大技术决策&#xff1a;弃用长期使用的 Nginx&#xff0c;转而采用其内部开发…...

k8s业务程序联调工具-KtConnect

概述 原理 工具作用是建立了一个从本地到集群的单向VPN&#xff0c;根据VPN原理&#xff0c;打通两个内网必然需要借助一个公共中继节点&#xff0c;ktconnect工具巧妙的利用k8s原生的portforward能力&#xff0c;简化了建立连接的过程&#xff0c;apiserver间接起到了中继节…...

QT: `long long` 类型转换为 `QString` 2025.6.5

在 Qt 中&#xff0c;将 long long 类型转换为 QString 可以通过以下两种常用方法实现&#xff1a; 方法 1&#xff1a;使用 QString::number() 直接调用 QString 的静态方法 number()&#xff0c;将数值转换为字符串&#xff1a; long long value 1234567890123456789LL; …...

代理篇12|深入理解 Vite中的Proxy接口代理配置

在前端开发中,常常会遇到 跨域请求接口 的情况。为了解决这个问题,Vite 和 Webpack 都提供了 proxy 代理功能,用于将本地开发请求转发到后端服务器。 什么是代理(proxy)? 代理是在开发过程中,前端项目通过开发服务器,将指定的请求“转发”到真实的后端服务器,从而绕…...

【数据分析】R版IntelliGenes用于生物标志物发现的可解释机器学习

禁止商业或二改转载&#xff0c;仅供自学使用&#xff0c;侵权必究&#xff0c;如需截取部分内容请后台联系作者! 文章目录 介绍流程步骤1. 输入数据2. 特征选择3. 模型训练4. I-Genes 评分计算5. 输出结果 IntelliGenesR 安装包1. 特征选择2. 模型训练和评估3. I-Genes 评分计…...

Mysql中select查询语句的执行过程

目录 1、介绍 1.1、组件介绍 1.2、Sql执行顺序 2、执行流程 2.1. 连接与认证 2.2. 查询缓存 2.3. 语法解析&#xff08;Parser&#xff09; 2.4、执行sql 1. 预处理&#xff08;Preprocessor&#xff09; 2. 查询优化器&#xff08;Optimizer&#xff09; 3. 执行器…...

JVM 内存结构 详解

内存结构 运行时数据区&#xff1a; Java虚拟机在运行Java程序过程中管理的内存区域。 程序计数器&#xff1a; ​ 线程私有&#xff0c;程序控制流的指示器&#xff0c;分支、循环、跳转、异常处理、线程恢复等基础功能都依赖这个计数器完成。 ​ 每个线程都有一个程序计数…...

若依登录用户名和密码加密

/*** 获取公钥&#xff1a;前端用来密码加密* return*/GetMapping("/getPublicKey")public RSAUtil.RSAKeyPair getPublicKey() {return RSAUtil.rsaKeyPair();}新建RSAUti.Java package com.ruoyi.common.utils;import org.apache.commons.codec.binary.Base64; im…...