关于hadoop:Hadoop-入门笔记-二十一-MapReduce-DB操作

一. 背景常识

通常组织会应用关系型数据来存储业务相干的数据，但随着数据的规模越来越大，尤其是像MySQL这种，在单表超过5千万条记录时，只管对表应用了特定的存储引擎和索引优化，但仍然不可避免的存在性能降落问题。
此时，咱们**能够通过应用MapReduce从MySQL中定期迁徙应用频率较低的历史数据到HDFS中，一方面能够升高对MySQL的存储和计算负载，另一方面，通过分布式计算引擎能够更加高效的解决过来的历史数据。

对于MapReduce框架来说，应用inputform进行数据读取操作，读取的数据首先由mapper解决，而后交给reducer解决，最终应用outputformat进行数据的输入操作。默认状况下，输入输出的组件实现都是针对文本数据处理的，别离是TextInputFormat、TextOutputFormat。
为了不便 MapReduce 间接拜访关系型数据库（Mysql,Oracle），Hadoop提供了DBInputFormat和DBOutputFormat两个类。其中DBInputFormat负责从数据库中读取数据，而DBOutputFormat负责把数据最终写入数据库中。

二. 读取数据库操作

1. 需要

在mysql中itcast_shop数据库下创立表itheima_goods并加载数据到表中。要求应用MapReduce程序将表中的数据导出寄存在指定的文件下。
数据库:
链接：https://pan.baidu.com/s/1ImrI…
提取码：pz9b

因为波及到java操作mysql，因而须要在pom依赖中额定增加mysql-jdbc驱动。

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.32</version>
    </dependency>

2. DBInputFormat类

InputFormat类用于从SQL表读取数据。DBInputFormat底层一行一行读取表中的数据，返回<k,v>键值对。其中k是LongWritable类型，表中数据的记录行号，从0开始，v是DBWritable类型，示意该行数据对应的对象类型。
此外还须要应用setInput办法设置SQL查问的语句相干信息。

3. 代码实现

1. 编写GoodsBean类

定义GoodsBean的实体类，用于封装查问返回的后果（如果要查问表的所有字段，那么属性就跟表的字段一一对应即可）。并且须要实现序列化机制Writable。
此外，从数据库读取/写入数据库的对象应实现DBWritable。 DBWritable与Writable类似，区别在于write（PreparedStatement）办法采纳PreparedStatement，而readFields（ResultSet）采纳ResultSet。

package com.uuicon.sentiment_upload.db;


import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.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;

public class GoodsBean implements Writable, DBWritable {
    private Long goodsId;
    private String goodsSn;
    private String goodsName;
    private Double marketPrice;
    private Double shopPrice;
    private Long saleNum;

    @Override
    public String toString() {
        return goodsId + "\t" + goodsSn + '\t' + goodsName + "\t" + marketPrice + "\t" + shopPrice + "\t" + saleNum;
    }

    public Long getGoodsId() {
        return goodsId;
    }

    public void setGoodsId(Long goodsId) {
        this.goodsId = goodsId;
    }

    public String getGoodsSn() {
        return goodsSn;
    }

    public void setGoodsSn(String goodsSn) {
        this.goodsSn = goodsSn;
    }

    public String getGoodsName() {
        return goodsName;
    }

    public void setGoodsName(String goodsName) {
        this.goodsName = goodsName;
    }

    public Double getMarketPrice() {
        return marketPrice;
    }

    public void setMarketPrice(Double marketPrice) {
        this.marketPrice = marketPrice;
    }

    public Double getShopPrice() {
        return shopPrice;
    }

    public void setShopPrice(Double shopPrice) {
        this.shopPrice = shopPrice;
    }

    public Long getSaleNum() {
        return saleNum;
    }

    public void setSaleNum(Long saleNum) {
        this.saleNum = saleNum;
    }

    public GoodsBean() {

    }

    public GoodsBean(Long goodsId, String goodsSn, String goodsName, Double marketPrice, Double shopPrice, Long saleNum) {
        this.goodsId = goodsId;
        this.goodsSn = goodsSn;
        this.goodsName = goodsName;
        this.marketPrice = marketPrice;
        this.shopPrice = shopPrice;
        this.saleNum = saleNum;
    }

    public void set(Long goodsId, String goodsSn, String goodsName, Double marketPrice, Double shopPrice, Long saleNum) {
        this.goodsId = goodsId;
        this.goodsSn = goodsSn;
        this.goodsName = goodsName;
        this.marketPrice = marketPrice;
        this.shopPrice = shopPrice;
        this.saleNum = saleNum;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(goodsId);
        out.writeUTF(goodsSn);
        out.writeUTF(goodsName);
        out.writeDouble(marketPrice);
        out.writeDouble(shopPrice);
        out.writeLong(saleNum);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.goodsId = in.readLong();
        this.goodsSn = in.readUTF();
        this.goodsName = in.readUTF();
        this.marketPrice = in.readDouble();
        this.shopPrice = in.readDouble();
        this.saleNum = in.readLong();
    }

    @Override
    public void write(PreparedStatement ps) throws SQLException {
        ps.setLong(1, goodsId);
        ps.setString(2, goodsSn);
        ps.setString(3, goodsName);
        ps.setDouble(4, marketPrice);
        ps.setDouble(5, shopPrice);
        ps.setLong(6, saleNum);
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.goodsId = resultSet.getLong(1);
        this.goodsSn = resultSet.getString(2);
        this.goodsName = resultSet.getString(3);
        this.marketPrice = resultSet.getDouble(4);
        this.shopPrice = resultSet.getDouble(5);
        this.saleNum = resultSet.getLong(6);
    }
}

2. 编写Mapper类

package com.uuicon.sentiment_upload.db;

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

import java.io.IOException;

public class ReadDBMapper extends Mapper<LongWritable, GoodsBean, LongWritable, Text> {
    Text outValue = new Text();

    @Override
    protected void map(LongWritable key, GoodsBean value, Context context) throws IOException, InterruptedException {
        outValue.set(value.toString());
        context.write(key, outValue);
    }
}

3. 创立程序驱动类

package com.uuicon.sentiment_upload.db;


import com.uuicon.sentiment_upload.covidtopn.CovidTopNDriver;
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.lib.output.FileOutputFormat;
import org.apache.hadoop.mapred.lib.db.DBConfiguration;
import org.apache.hadoop.mapred.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.Job;

import java.io.File;
import java.io.IOException;

public class ReadDBDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        //数据库信息
        DBConfiguration.configureDB(conf,
                "com.mysql.jdbc.Driver",
                "jdbc:mysql://localhost:3306/itcast_goods",
                "root",
                "root"
        );
        // 创立作业类
        Job job = Job.getInstance(conf, ReadDBDriver.class.getSimpleName());
        //设置mr 驱动类
        job.setJarByClass(ReadDBDriver.class);

        //设置mapper 类
        job.setMapperClass(ReadDBMapper.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        // todo 设置输出组件
        FileOutputFormat.setOutputPath(job,
                new Path("E:\\ml\\hadoop\\mysqlout"));

        //设置Reducer 类 ,todo 本例不须要reduce ,操作形式是将tasknumber 设置为 0
        job.setNumReduceTasks(0);

        // todo 设置输出组件
        job.setInputFormatClass(DBInputFormat.class);
        //增加读取数据库相干参数

        DBInputFormat.setInput(
                job,
                GoodsBean.class,
                "select goodsId ,goodsSn,goodsName,marketPrice ,shopPrice , saleNum from itheima_goods",
                "select count(goodsId) from itheima_goods"
        );
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);

    }
}

4. 运行程序

间接在驱动类中右键运行main办法，应用MapReduce的本地模式执行。也能够将程序应用maven插件打包成jar包，提交到yarn上进行分布式运行。

3. 输入到数据库操作

1. 需要

有一份结构化的数据文件，数据内容对应着mysql中一张表的内容，要求应用MapReduce程序将文件的内容读取写入到mysql中。
就以上例的输入文件作为结构化文件，上面在mysql中创立对应的表构造。
表构造:

CREATE TABLE `itheima_goods_mr_write` (
  `goodsId` bigint(11) NOT NULL AUTO_INCREMENT COMMENT '商品id',
  `goodsSn` varchar(20) NOT NULL COMMENT '商品编号',
  `goodsName` varchar(200) NOT NULL COMMENT '商品名称',
  `marketPrice` decimal(11,2) NOT NULL DEFAULT '0.00' COMMENT '市场价',
  `shopPrice` decimal(11,2) NOT NULL DEFAULT '0.00' COMMENT '门店价',
  `saleNum` int(11) NOT NULL DEFAULT '0' COMMENT '总销售量',
  PRIMARY KEY (`goodsId`)
) ENGINE=InnoDB AUTO_INCREMENT=115909 DEFAULT CHARSET=utf8;

2. DBOutputFormat类

OutputFormat，它将reduce输入发送到SQL表。DBOutputFormat承受<key，value>键值对，其中key必须具备扩大DBWritable的类型。
此外还须要应用setOutput办法设置SQL插入语句相干信息，比方表、字段等。

3. 代码实现

1. 编写GoodsBean类

定义GoodsBean的实体类，用于封装插入表中的数据（对象属性跟表的字段一一对应即可）。并且须要实现序列化机制Writable。
此外，从数据库读取/写入数据库的对象应实现DBWritable。 DBWritable与Writable类似，区别在于write（PreparedStatement）办法采纳PreparedStatement，而readFields（ResultSet）采纳ResultSet。

package com.uuicon.sentiment_upload.db;


import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.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;

public class GoodsBean implements Writable, DBWritable {
    private Long goodsId;
    private String goodsSn;
    private String goodsName;
    private Double marketPrice;
    private Double shopPrice;
    private Long saleNum;

    @Override
    public String toString() {
        return goodsId + "\t" + goodsSn + '\t' + goodsName + "\t" + marketPrice + "\t" + shopPrice + "\t" + saleNum;
    }

    public Long getGoodsId() {
        return goodsId;
    }

    public void setGoodsId(Long goodsId) {
        this.goodsId = goodsId;
    }

    public String getGoodsSn() {
        return goodsSn;
    }

    public void setGoodsSn(String goodsSn) {
        this.goodsSn = goodsSn;
    }

    public String getGoodsName() {
        return goodsName;
    }

    public void setGoodsName(String goodsName) {
        this.goodsName = goodsName;
    }

    public Double getMarketPrice() {
        return marketPrice;
    }

    public void setMarketPrice(Double marketPrice) {
        this.marketPrice = marketPrice;
    }

    public Double getShopPrice() {
        return shopPrice;
    }

    public void setShopPrice(Double shopPrice) {
        this.shopPrice = shopPrice;
    }

    public Long getSaleNum() {
        return saleNum;
    }

    public void setSaleNum(Long saleNum) {
        this.saleNum = saleNum;
    }

    public GoodsBean() {

    }

    public GoodsBean(Long goodsId, String goodsSn, String goodsName, Double marketPrice, Double shopPrice, Long saleNum) {
        this.goodsId = goodsId;
        this.goodsSn = goodsSn;
        this.goodsName = goodsName;
        this.marketPrice = marketPrice;
        this.shopPrice = shopPrice;
        this.saleNum = saleNum;
    }

    public void set(Long goodsId, String goodsSn, String goodsName, Double marketPrice, Double shopPrice, Long saleNum) {
        this.goodsId = goodsId;
        this.goodsSn = goodsSn;
        this.goodsName = goodsName;
        this.marketPrice = marketPrice;
        this.shopPrice = shopPrice;
        this.saleNum = saleNum;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(goodsId);
        out.writeUTF(goodsSn);
        out.writeUTF(goodsName);
        out.writeDouble(marketPrice);
        out.writeDouble(shopPrice);
        out.writeLong(saleNum);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.goodsId = in.readLong();
        this.goodsSn = in.readUTF();
        this.goodsName = in.readUTF();
        this.marketPrice = in.readDouble();
        this.shopPrice = in.readDouble();
        this.saleNum = in.readLong();
    }

    @Override
    public void write(PreparedStatement ps) throws SQLException {
        ps.setLong(1, goodsId);
        ps.setString(2, goodsSn);
        ps.setString(3, goodsName);
        ps.setDouble(4, marketPrice);
        ps.setDouble(5, shopPrice);
        ps.setLong(6, saleNum);
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.goodsId = resultSet.getLong(1);
        this.goodsSn = resultSet.getString(2);
        this.goodsName = resultSet.getString(3);
        this.marketPrice = resultSet.getDouble(4);
        this.shopPrice = resultSet.getDouble(5);
        this.saleNum = resultSet.getLong(6);
    }
}

2. Mapper 类

package com.uuicon.sentiment_upload.db;

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;

public class WriteDBMapper extends Mapper<LongWritable, Text, NullWritable, GoodsBean> {
    NullWritable outKey = NullWritable.get();
    GoodsBean outValue = new GoodsBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        Counter sc = context.getCounter("mr_to_sql", "SUCCESS");
        Counter fc = context.getCounter("mr_to_sql", "FAILED");
        String[] fields = value.toString().split("\\s+");
        if (fields.length > 6) {
            // 失常数据
            outValue.set(
                    Long.parseLong(fields[1]),
                    fields[2],
                    fields[3],
                    Double.parseDouble(fields[4]),
                    Double.parseDouble(fields[5]),
                    Long.parseLong(fields[6])
            );
            context.write(outKey, outValue);
            sc.increment(1);
        } else {
            // 异样数据
            fc.increment(1);
        }
    }
}

3. reudce 类

package com.uuicon.sentiment_upload.db;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * todo 在应用DBOutputFormat 的时候,要求输入的key 必须实现DBWritable 因为只会把key写入数据库
 */
public class WriteDBReducer extends Reducer<NullWritable, GoodsBean, GoodsBean, NullWritable> {
    @Override
    protected void reduce(NullWritable key, Iterable<GoodsBean> values, Context context) throws IOException, InterruptedException {
        for (GoodsBean value : values) {
            context.write(value, key);
        }
    }
}

4. 驱动类

package com.uuicon.sentiment_upload.db;


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

public class WriteDBDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        //数据库信息
        DBConfiguration.configureDB(conf,
                "com.mysql.jdbc.Driver",
                "jdbc:mysql://localhost:3306/itcast_goods?useUnicode=true&characterEncoding=utf8",
                "root",
                "root"
        );
        // 创立作业类
        Job job = Job.getInstance(conf, WriteDBDriver.class.getSimpleName());
        //设置mr 驱动类
        job.setJarByClass(WriteDBDriver.class);

        //设置mapper 类
        job.setMapperClass(WriteDBMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(GoodsBean.class);

        //设置 Reduce 相干
        job.setReducerClass(WriteDBReducer.class);
        job.setOutputKeyClass(GoodsBean.class);
        job.setOutputValueClass(NullWritable.class);

        // 设置以后作业的文件门路
        FileInputFormat.setInputPaths(job, new Path("E:\\ml\\hadoop\\mysqlout"));
        // todo 设置程序输入类
        job.setOutputFormatClass(DBOutputFormat.class);
        // 配置以后作业,写入数据库表 itheima_goods_mr_write
        DBOutputFormat.setOutput(
                job,
                "itheima_goods_mr_write",
                "goodsId","goodsSn","goodsName","marketPrice","shopPrice","saleNum"
                );
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);

    }
}

关于hadoop:Hadoop-入门笔记-二十一-MapReduce-DB操作

一. 背景常识

二. 读取数据库操作

1. 需要

2. DBInputFormat类

3. 代码实现

1. 编写GoodsBean类

2. 编写Mapper类

3. 创立程序驱动类

4. 运行程序

3. 输入到数据库操作

1. 需要

2. DBOutputFormat类

3. 代码实现

1. 编写GoodsBean类

2. Mapper 类

3. reudce 类

4. 驱动类

5. 运行后果

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于hadoop:Hadoop-入门笔记-二十一-MapReduce-DB操作

一. 背景常识

二. 读取数据库操作

1. 需要

2. DBInputFormat类

3. 代码实现

1. 编写GoodsBean类

2. 编写Mapper类

3. 创立程序驱动类

4. 运行程序

3. 输入到数据库操作

1. 需要

2. DBOutputFormat类

3. 代码实现

1. 编写GoodsBean类

2. Mapper 类

3. reudce 类

4. 驱动类

5. 运行后果

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复