MapReduce高级-读写数据库

news2024/11/16 0:24:28

MapReduce 读取数据库

为什么要读写数据库

本质上讲数据库是存储数据的介质，MapReduce是处理数据的计算引擎。通常企业会使用关系型数据库（RDBMS）来存储业务的相关数据，随着业务数据的规模越来越大，不可避免的存在性能下降的问题，这里存在两个说法：

  • 百度： MySQL单表数据量大于2000万行，性能会明显下降
  • 案例：单表行数超过500w行或者单表容量大于2G，推荐使用分库分表

此时我们可以通过使用MapReduce从MySQL中定期迁移使用频率比较低的历史数据到HDFS中：

  • 一方面可以降低MySQL的存储核计算负载
  • 通过分布式计算引擎可以更加高效的处理过去的历史数据

如何实现读写数据库

对于MapReduce框架来说，使用InputFormart进行读取数据，读取的数据首先由Mapper 进行处理，然后交给Reduce处理，最终使用OutputFormat进行数据的输出操作，默认情况下，输入输出的组件实现都是针对文本数据处理的，分别是TextInputFormat、TextOutputFormat。

为了方便MapReduce直接访问关系型数据库（MySQL、Oracle），Hadoop提供了DBInputFormat、DBOutputFormat两个类，其中DBInputForm负责从数据库读取数据，而DBOutputFormat负责把数据写入数据库中

image-20230419084624025

使用测试

需求

在MySQL中shop数据库下的produce中的数据导出存放在指定的文件系统目录下。

那么传统的读取方式肯定不行，那么采用什么方式来读取呢？

DBInputFormat

DBInputFormat类用于从SQL表中读取数据，底层一行一行的读取表中的数据，返回<K,V>键值对，

其中K是LongWritable类型，表示表中数据的记录行号，从0开始

V是DBWritable类型，表示该行数据对应的对象类型

DBInputFormat能够读取MySQL本质上还是在底层封装了JDBC，所以在后续项目中还要加上JDBC的驱动

读取MySQL数据

DBInputFormat在底层封装了MySQL，那么在使用的过程中，就需要加上JDBC的驱动，后续为了方便，这里也加上了lombok的依赖来简化开发

POM文件整体

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.wxk</groupId>
    <artifactId>HDFS-HDFS2Test</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.25</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.26</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <archive>
                        <manifest>
                            <addClasspath>true</addClasspath>
                            <classpathPrefix>lib/</classpathPrefix>
                            <mainClass>MapReduceTest.WordDriver</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

编写Bean文件

在编写Bean文件的时候需要实现Writable和DBWritable这两个接口

package MapReduceTest.DB.Reader;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @author wxk
 * @date 2023/04/19/17:39
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderBean implements Writable, DBWritable {
    private int id;
    private String order;
    private String time;

    @Override
    public String toString() {
        return id + "\t" + order + "\t" +time;
    }

    // 序列化方法，将数据写出去
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(id);
        out.writeUTF(order);
        out.writeUTF(time);
    }

    //序列化方法，将数据读取进来
    @Override
    public void readFields(DataInput in) throws IOException {
        this.id = in.readInt();
        this.order=in.readUTF();
        this.time= in.readUTF();
    }

    //序列化 写入数据库
    @Override
    public void write(PreparedStatement ps) throws SQLException {
        ps.setInt(1,id);
        ps.setString(2,order);
        ps.setString(3,time);
    }

    //将查询结果赋予给此对象
    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.id=resultSet.getInt(1);
        this.order=resultSet.getString(2);
        this.time=resultSet.getString(3);
    }
}

编写Mapper文件

在配置Mapper文件中，我们需要了解一下信息：

Mapper中的类型表示的输入输出的KV的格式：输入的K是Long类型，V是GoodsBean类型，输出的K是Long类型，V是字符串类型。这里输入的KEY是字符串类型是因为K是一个偏移量，表示当前读取的是哪一行，后续可以根据自己的想法进行设置

package MapReduceTest.DB.Reader;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author wxk
 * @date 2023/04/19/9:53
 */
public class ReaderMapper extends Mapper<LongWritable,OrderBean,LongWritable, Text> {
    Text out =new Text();
    @Override
    protected void map(LongWritable key, OrderBean value, Context context) throws IOException, InterruptedException {
        out.set(value.toString());
        context.write(key,out);
    }
}

配置运行的Driver驱动

package MapReduceTest.DB.Reader;

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.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author wxk
 * @date 2023/04/19/9:59
 */
public class ReaderDriver {
    public static void main(String[] args) throws Exception {
        //配置文件对象
        Configuration conf = new Configuration();
        //配置当前作业需要的JDBC密码
        DBConfiguration.configureDB(
                conf,
                "com.mysql.cj.jdbc.Driver",
                "jdbc:mysql://localhost:3306/shop",
                "root",
                "20020219"
        );
        //创建作业的job
        Job job = Job.getInstance(conf, ReaderDriver.class.getSimpleName());
        //设置MapReduce的输出格式
        job.setJarByClass(ReaderDriver.class);
        job.setMapperClass(ReaderMapper.class);
        //key的格式
        job.setOutputKeyClass(LongWritable.class);
        //value的格式
        job.setOutputValueClass(Text.class);
        //不需要Reduce阶段，就把ReduceTask设置为 表明不在执行MapReduce
        job.setNumReduceTasks(0);
        //设置输入组件
        job.setInputFormatClass(DBInputFormat.class);
        FileOutputFormat.setOutputPath(job,new Path("E://mysql_out"));
        DBInputFormat.setInput(
                job,
                OrderBean.class,
                "select * from `order`",
                "select count(*) from `order`");
        final boolean b = job.waitForCompletion(true);
        System.out.println(b ? 0:1);

    }
}

运行之后，查看文件夹：

image-20230419172517196

查看文件：

image-20230419175755642

对比数据库：

image-20230419175815938

可见任务基本完成

这里有一个小细节，就是输出文件名和之前的不一样，在这里中间是m，而之前是r如图：

image-20230419173132644

这里输出是m是因为Reduce环节根本就没有进行，所以是m而不是r，而之前的是因为走的是全流程，最后经过了Reduce的处理，结果是r

如果经过了Reduce操作，那么输出文件中是r，如果仅仅经过了Map的处理，那么就是m

写入MySQL数据

将数据库中的数据进行清空，然后进行一个配置

Map
package MapReduceTest.DB.Writer;

import MapReduceTest.DB.Reader.OrderBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author wxk
 * @date 2023/04/19/20:05
 */
public class WriteDBMapper extends Mapper<LongWritable, Text, NullWritable, OrderBean> {
    OrderBean outValue = new OrderBean();
    NullWritable outKey = NullWritable.get();
    private final static int INCR_NUMBER = 1;

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //计数器的模拟
        Counter sc = context.getCounter("wxk", "sc_counter");
        Counter fe = context.getCounter("wxk", "fe_counter");


        String[] split = value.toString().split("\t");
        if (split.length != 4) {
            //长度不为4表明数据不合法
            fe.increment(INCR_NUMBER);
        } else {
            outValue.setId(Integer.parseInt(split[1]));
            outValue.setOrder(split[2]);
            outValue.setTime(split[3]);
            context.write(outKey,outValue);
            //合法数据，就加一
            sc.increment(INCR_NUMBER);
        }
    }
}
Reduce
package MapReduceTest.DB.Writer;

import MapReduceTest.DB.Reader.OrderBean;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author wxk
 * @date 2023/04/19/20:27
 */
//在这里输出的时候Key必须为DBWritable类，V随意，因为最终是将K写入到数据库中
public class WriteDBReduce extends Reducer<NullWritable, OrderBean, OrderBean, NullWritable> {
    NullWritable outValue = NullWritable.get();

    @Override
    protected void reduce(NullWritable key, Iterable<OrderBean> values, Context context) throws IOException, InterruptedException {
        for (OrderBean item : values) {
            context.write(item, outValue);
        }
    }
}
Driver
package MapReduceTest.DB.Writer;

import MapReduceTest.DB.Reader.OrderBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;

/**
 * @author wxk
 * @date 2023/04/19/20:32
 */
public class WriteDBDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        DBConfiguration.configureDB(
                conf,
                "com.mysql.cj.jdbc.Driver",
                "jdbc:mysql://localhost:3306/shop?useSSL=false&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true",
                "root",
                "20020219"
        );
        Job job = Job.getInstance(conf,WriteDBDriver.class.getSimpleName());
        //设置Mapper驱动
        job.setMapperClass(WriteDBMapper.class);
        //设置驱动
        job.setJarByClass(WriteDBDriver.class);
        //设置Mapper输出Key的类型
        job.setMapOutputKeyClass(NullWritable.class);
        //设置Mapper输出Value的类型
        job.setMapOutputValueClass(OrderBean.class);
        //设置Reduce
        job.setReducerClass(WriteDBReduce.class);
        //设置Reduce输出的Key的类型
        job.setOutputKeyClass(OrderBean.class);
        //设置Reduce输出Value的类型
        job.setOutputValueClass(NullWritable.class);
        //设置输入路径
        FileInputFormat.setInputPaths(job,new Path("E://mysql_out"));
        //设置输出格式
        job.setOutputFormatClass(DBOutputFormat.class);
        //配置作业协入数据库的表/字段
        DBOutputFormat.setOutput(job,
                "`order`",
                "id","`order`","time");
        boolean b = job.waitForCompletion(true);
        System.out.println(b ? 0: 1);
    }
}

运行之后：

image-20230419213009532

在这里可以看到成功插入了20条，失败0条

查看MySQL数据库：

image-20230419213114588

插入成功

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/446832.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

C++ -3- 类和对象 (中) | 拷贝构造函数 赋值运算符重载（二）

C++ -3- 类和对象 (中) | 拷贝构造函数 赋值运算符重载（二）

文章目录 4.拷贝构造函数什么是拷贝构造函数&#xff1f;应用——示例&#xff1a;日期计算器什么情况下需要自己实现拷贝构造函数&#xff1f; 5.赋值运算符重载运算符重载&#xff08;重要&#xff09;赋值运算符重载 拷贝构造函数和赋值重载函数 4.拷贝构造函数 什么是拷贝…
阅读更多...
【进阶C语言】静态版通讯录的实现（详细讲解+全部源码）

【进阶C语言】静态版通讯录的实现（详细讲解+全部源码）

前言 &#x1f4d5;作者简介&#xff1a;热爱跑步的恒川&#xff0c;正在学习C/C、Java、Python等。 &#x1f4d7;本文收录于C语言进阶系列&#xff0c;本专栏主要内容为数据的存储、指针的进阶、字符串和内存函数的介绍、自定义类型结构、动态内存管理、文件操作等&#xff0…
阅读更多...
Javaee Spring JdbcTemplate基本使用查询数据库表的内容 基于xml配置方式

Javaee Spring JdbcTemplate基本使用查询数据库表的内容 基于xml配置方式

目录 哈哈哈哈&#xff0c;说好是要写一篇关于jdbcTemplate的基本使用&#xff0c;貌似说跑题了&#xff0c;但是主体还是用jdbctemplate实现的&#xff0c;有耐心看完的话相信能有点点收获的哦&#xff01; 项目结构&#xff1a; 用到的数据库: 小结&#xff1a; 遇到了个小…
阅读更多...
【Python_Opencv图像处理框架】图像基本操作

【Python_Opencv图像处理框架】图像基本操作

写在前面 很幸运能选择Python语言进行学习&#xff0c;这是有关Opencv的图像处理的第一篇文章&#xff0c;讲解了有关图像处理的一些基础操作&#xff0c;作为初学者&#xff0c;我尽己所能&#xff0c;但仍会存在疏漏的地方&#xff0c;希望各位看官不吝指正❤️ 写在中间 1…
阅读更多...
Docker容器---介绍、安装

Docker容器---介绍、安装

Docker基本管理 一、Docker概述1、IT架构2、什么是docker3、Docker特点4、Docker与KVM区别 二、Docker核心概念1、镜像 容器 仓库2、虚拟架构有哪些 三、Docker使用场景1、Docker在内核中支持的2种重要技术2、应用场景 四、Docker安装1、YUM安装docker2、设置阿里云镜像源3、查…
阅读更多...
Java多线程初阶（一）（图片＋源码＋超详细）

Java多线程初阶（一）（图片＋源码＋超详细）

线程的概念参照以往的这篇文章&#x1f43b; 目录 1.创建线程 1.1 继承Thread类 1.2 实现Runnable接口 eg&#xff1a;常用的简写方式 2.Thread类中的常用API 3. start方法和run方法 4. 继承Thread类启动新线程的逻辑 5. 实现Runnable接口启动新线程的逻辑 6. 线程相关…
阅读更多...
Linux基础—日志分析

Linux基础—日志分析

Linux基础—日志分析 一、日志的功能1.日志消息的级别2.设备字段说明 二、日志文件的分类1.内核及系统日志2.用户日志3.程序日志 三、日志文件1.日志文件查看2.主要日志文件介绍3.日志管理策略 一、日志的功能 用于记录系统、程序运行中发生的各种事件 通过阅读日志&#xff0c…
阅读更多...
图像描述算法排位赛：SceneXplain与MiniGPT-4谁将夺得桂冠？

图像描述算法排位赛：SceneXplain与MiniGPT-4谁将夺得桂冠？

如果你对 AI 前沿感兴趣&#xff0c;本场「图像描述算法排位赛」绝对是你不能错过的&#xff01;在这场较量中&#xff0c;SceneXplain 和 MiniGPT-4 将会比试&#xff0c;谁将摘得这场比赛的桂冠&#xff1f; &#x1f4ce; 直接上手体验&#xff1a;scenex.jina.ai 背景介绍 …
阅读更多...
手机存储数据恢复软件哪个好用？试了10款，我只认准这一款！

手机存储数据恢复软件哪个好用？试了10款，我只认准这一款！

案例&#xff1a;手机存储数据恢复软件哪个好用&#xff1f; 【有没有好介绍的苹果数据恢复软件&#xff1f;可以恢复好几年数据的那种软件&#xff1f;求推荐&#xff01;】 手机中的数据是用户平时使用手机时不可避免的内容&#xff0c;这些数据包括照片、视频、音乐、文件等…
阅读更多...
Baumer工业相机堡盟工业相机如何通过BGAPISDK的软触发实现两相机同步采集（C++）

Baumer工业相机堡盟工业相机如何通过BGAPISDK的软触发实现两相机同步采集（C++）

Baumer工业相机堡盟工业相机如何通过BGAPISDK的软触发实现两相机的同步采集&#xff08;C&#xff09; Baumer工业相机Baumer工业相机的高速同步采集的技术背景Baumer工业相机通过BGAPI SDK在回调函数里同步保存图像工业相机在回调函数BufferEvent保存工业相机使用软触发进行同…
阅读更多...
【观察】华为：新一代楼宇网络，使能绿建智慧化

【观察】华为：新一代楼宇网络，使能绿建智慧化

“碳达峰”、“碳中和”目标是我国生态文明建设和高质量可持续发展的重要战略安排&#xff0c;将推动全社会加速向绿色低碳转型。作为全球既有建筑和每年新建建筑量最大的国家&#xff0c;大力发展绿色建筑对中国全方位迈向低碳社会、实现高质量发展具有重要意义。 《“十四五”…
阅读更多...
使用FirmAE 对zyxel路由器固件仿真实践 | 信息安全

使用FirmAE 对zyxel路由器固件仿真实践 | 信息安全

一、FirmAE简介 FirmAE 是一个执行仿真和漏洞分析的全自动框架。FirmAE 使用五种仲裁技术显著提高仿真成功率&#xff08;从Firmadyne的 16.28% 提高到 79.36%&#xff09;。 FirmAE的整体架构为如上图所示。与Firmadyne类似&#xff0c;FirmAE在预先构建的自定义Linux内核和库…
阅读更多...
Codeforces Round 816 (Div. 2)（C. Monoblock）

Codeforces Round 816 (Div. 2)（C. Monoblock）

传送门 题意&#xff1a; 定义一个数组&#xff0c;定义它 相等连续段的个数 如1,2,2,1&#xff0c;相等连续段有[1],[2,2],[1]&#xff0c;共3个 如2,2,2,1&#xff0c;相等连续段有[2,2,2],[1]&#xff0c;共2个 给定一个数组a&#xff0c;定义g(l,r)为&#xff0c;子数组中…
阅读更多...
fisco bcos 2.0+ 版本浏览器搭建

fisco bcos 2.0+ 版本浏览器搭建

Fisco Bcos当前已经发布了3版本&#xff0c;在3版本中&#xff0c;Fisco摒弃了原有的区块链浏览器&#xff0c;而是提供了具有更为丰富功能的图形化区块链管理工具WeBase&#xff0c;后续也会提供对于WeBase的搭建部署教程。 文章目录 一、准备条件二、下载浏览器安装脚本1. 下…
阅读更多...
ChatGPT4 的体验 一站式 AI工具箱 -—Poe（使用教程）

ChatGPT4 的体验 一站式 AI工具箱 -—Poe（使用教程）

最近由于人工智能聊天机器人的爆火(ChatGPT)&#xff0c;因此各种各样的AI助手流行与网络&#xff0c;各种各样的都有&#xff0c;不论是什么样的其实都是为了我们更方便的解决问题&#xff0c;今天介绍一款AI工具箱——Poe将多种AI集成与一个界面&#xff0c;大家可以一次感受…
阅读更多...
Mac系统brew报错“The GitHub credentials in the macOS keychain may be invalid”解决

Mac系统brew报错“The GitHub credentials in the macOS keychain may be invalid”解决

报错信息如下&#xff1a; $ brew search nginx Warning: Error searching on GitHub: GitHub API Error: Requires authentication The GitHub credentials in the macOS keychain may be invalid. Clear them with: printf "protocolhttps\nhostgithub.com\n" | …
阅读更多...
Python双向链表的操作

Python双向链表的操作

目录 一、双向链表 双向链表示例图 二、双向链表的操作 1、判断链表是否为空 2&#xff0c;链表长度 3&#xff0c;遍历整个链表 4&#xff0c;在链表头部添加元素 5、链表尾部添加元素 6&#xff0c;在指定位置插入元素 7&#xff0c;修改指定位置的元素 8&#xff0…
阅读更多...
百度ai智能写作工具-百度ai自动写文章

百度ai智能写作工具-百度ai自动写文章

百度AI智能写作工具&#xff1a;让创作更快捷、高效&#xff01; 在当今竞争激烈的文化创意市场中&#xff0c;创作一篇高质量的文章需要投入大量时间和精力。然而&#xff0c;有了百度AI智能写作工具&#xff0c;创作变得更快捷、高效了。 百度AI智能写作工具采用最先进的人…
阅读更多...
JVM之垃圾回收器概述

JVM之垃圾回收器概述

目录 垃圾收集器分类 按线程数分 按照工作模式分 ​编辑 按碎片处理方式分 按工作的内存区间分 评估GC的性能指标 吞吐量 暂停时间 吞吐量 vs 暂停时间 垃圾回收器概述 垃圾收集器没有在规范中进行过多的规定&#xff0c;可以由不同的厂商、不同版本的JVM来实现。 由…
阅读更多...
一起学 WebGL：感受三维世界之视图矩阵

一起学 WebGL：感受三维世界之视图矩阵

大家好&#xff0c;我是前端西瓜哥。之前绘制的图形都是在 XY 轴所在的平面上&#xff0c;这次我们来加入一点深度信息 z&#xff0c;带你走入三维的世界。 视图矩阵 对于一个立方体来说&#xff0c;我们从它的正前方看&#xff0c;不管距离它多远&#xff0c;也只能看到一个…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023