使用DBOutputFormat把MapReduce产生的结果集导入到 MySQL 中

以你之姓@ 2023-01-14 04:58 101阅读 0赞

数据在HDFS和关系型数据库之间的迁移，主要有以下两种方式：

1.  按照数据库要求的文件格式生成文件，然后由数据库提供的导入工具进行导入
2.  采用JDBC的方式进行导入

MapReduce 默认提供了 `DBInputFormat` 和`DBOutputFormat`，分别用于数据库的`读取`和数据库的`写入`

**1.需求**

下面使用 DBOutputFormat，将 MapReduce 处理后的学生信息导入到 mysql 中

**2.数据集**

> 张明明 45  
> 李成友 78  
> 张辉灿 56  
> 王露 56  
> 陈东明 67  
> 陈果 31  
> 李华明 32  
> 张明东 12  
> 李明国 34  
> 陈道亮 35  
> 陈家勇 78  
> 陈旻昊 13  
> 陈潘 78  
> 陈学澄 18

**3.实现**

package com.buaa;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import java.sql.PreparedStatement;
    import java.sql.ResultSet;
    import java.sql.SQLException;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.filecache.DistributedCache;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.io.Writable;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
    import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
    import org.apache.hadoop.mapreduce.lib.db.DBWritable;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    /** * @ProjectName DBOutputormatDemo * @PackageName com.buaa * @ClassName MysqlDBOutputormatDemo * @Description TODO * @Author 刘吉超 * @Date 2016-05-06 09:15:57 */
    @SuppressWarnings({  "unused", "deprecation" })
    public class MysqlDBOutputormatDemo extends Configured implements Tool { 
         /** * 实现DBWritable * * TblsWritable需要向mysql中写入数据 */  
        public static class TblsWritable implements Writable, DBWritable {   
            String tbl_name;  
            int tbl_age;  
      
            public TblsWritable() {   
            }  
      
            public TblsWritable(String name, int age) {   
                this.tbl_name = name;  
                this.tbl_age = age;  
            }  
     
            @Override  
            public void write(PreparedStatement statement) throws SQLException {   
                statement.setString(1, this.tbl_name);  
                statement.setInt(2, this.tbl_age);  
            }  
     
            @Override  
            public void readFields(ResultSet resultSet) throws SQLException {   
                this.tbl_name = resultSet.getString(1);  
                this.tbl_age = resultSet.getInt(2);  
            }  
     
            @Override  
            public void write(DataOutput out) throws IOException {   
                out.writeUTF(this.tbl_name);  
                out.writeInt(this.tbl_age);  
            }  
     
            @Override  
            public void readFields(DataInput in) throws IOException {   
                this.tbl_name = in.readUTF();  
                this.tbl_age = in.readInt();  
            }  
      
            public String toString() {   
                return new String(this.tbl_name + " " + this.tbl_age);  
            }  
        }  
        
        public static class StudentMapper extends Mapper<LongWritable, Text, LongWritable, Text>{ 
            @Override
            protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException { 
                context.write(key, value);
            }
        }
        
        public static class StudentReducer extends Reducer<LongWritable, Text, TblsWritable, TblsWritable> { 
            @Override
            protected void reduce(LongWritable key, Iterable<Text> values,Context context) throws IOException, InterruptedException { 
                // values只有一个值，因为key没有相同的
                StringBuilder value = new StringBuilder();
                for(Text text : values){ 
                    value.append(text);
                }
                
                String[] studentArr = value.toString().split("\t");
                
                if(StringUtils.isNotBlank(studentArr[0])){ 
                    /* * 姓名 年龄（中间以tab分割） * 张明明 45 */
                    String name = studentArr[0].trim();
                    
                    int age = 0;
                    try{ 
                        age = Integer.parseInt(studentArr[1].trim());
                    }catch(NumberFormatException e){ 
                    }
                    
                    context.write(new TblsWritable(name, age), null);  
                }
            }
        }
        
        public static void main(String[] args) throws Exception { 
            // 数据输入路径和输出路径
            String[] args0 = { 
                "hdfs://ljc:9000/buaa/student/student.txt"
            };
            int ec = ToolRunner.run(new Configuration(), new MysqlDBOutputormatDemo(), args0);
            System.exit(ec);
        }
        
        @Override
        public int run(String[] arg0) throws Exception { 
            // 读取配置文件
            Configuration conf = new Configuration();
            
            DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver",  
                    "jdbc:mysql://172.26.168.2:3306/test", "hadoop", "123");  
            
            // 新建一个任务
            Job job = new Job(conf, "DBOutputormatDemo");
            // 设置主类
            job.setJarByClass(MysqlDBOutputormatDemo.class);
            
            // 输入路径
            FileInputFormat.addInputPath(job, new Path(arg0[0]));
            
            // Mapper
            job.setMapperClass(StudentMapper.class);
            // Reducer
            job.setReducerClass(StudentReducer.class);
            
            // mapper输出格式
            job.setOutputKeyClass(LongWritable.class);
            job.setOutputValueClass(Text.class);
            
            // 输入格式，默认就是TextInputFormat
    // job.setInputFormatClass(TextInputFormat.class);
            // 输出格式
            job.setOutputFormatClass(DBOutputFormat.class);  
            
            // 输出到哪些表、字段
            DBOutputFormat.setOutput(job, "student", "name", "age");
    
            // 添加mysql数据库jar
    // job.addArchiveToClassPath(new Path("hdfs://ljc:9000/lib/mysql/mysql-connector-java-5.1.31.jar"));
    // DistributedCache.addFileToClassPath(new Path("hdfs://ljc:9000/lib/mysql/mysql-connector-java-5.1.31.jar"), conf);
            //提交任务
            return job.waitForCompletion(true)?0:1;
        }
    }

mr程序很简单，只是读取文件内容，在这里我们主要关注的是怎么将 mr 处理后的结果集导入MySQL中的。

数据库中表是`student`，为student表编写对应的bean类TblsWritable，该类需要实现`Writable接口` 和 `DBWritable接口`。

**1. Writable接口**

@Override  
    public void write(DataOutput out) throws IOException {   
        out.writeUTF(this.tbl_name);  
        out.writeInt(this.tbl_age);  
    }  
    
    @Override  
    public void readFields(DataInput in) throws IOException {   
        this.tbl_name = in.readUTF();  
        this.tbl_age = in.readInt();  
    }

上面两个方法对应着Writable接口，用对象序列化，这里不再多说，为什么要序列化，参考：[Hadoop序列化][Hadoop]

**2.DBWritable接口**

@Override  
    public void write(PreparedStatement statement) throws SQLException {   
           statement.setString(1, this.tbl_name);  
           statement.setInt(2, this.tbl_age);  
     }  
     
     @Override  
     public void readFields(ResultSet resultSet) throws SQLException {   
           this.tbl_name = resultSet.getString(1);  
           this.tbl_age = resultSet.getInt(2);  
     }

上面两个方法对应着 DBWriteable 接口。readFields 方法负责从结果集中读取数据库数据（注意 ResultSet 的下标是从1开始的），一次读取查询 SQL中筛选的某一列。Write 方法负责将数据写入到数据库，将每一行的每一列依次写入。

**最后进行Job的一些配置，具体如下面代码所示**

DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://172.26.168.2:3306/test", "hadoop", "123")

上面的配置主要包括以下几项：

1.  数据库驱动的名称：com.mysql.jdbc.Driver
2.  数据库URL：jdbc:mysql://172.26.168.2:3306/test
3.  用户名：hadoop
4.  密码：123

还有以下几项需要配置：

1.  数据库表以及每列的名称：DBOutputFormat.setOutput(job, “student”, “name”, “age”);
2.  输出格式改为：job.setOutputFormatClass(DBOutputFormat.class);

> **提示：**需要提醒的是 DBOutputFormat 以 MapReduce 的方式运行，会并行的连接数据库。在这里需要合适的设置map、reduce的个数，以便将并行连接的数量控制在合理的范围之内

**4.运行效果**  
![在这里插入图片描述][20210420095545646.png]  
**5.注意事项**

运行项目可能会报如下错误  
![在这里插入图片描述][202104200956206.png]

解决方法：

共有3种解决方法，但我喜欢第三种。

> 1、在每个节点下的$\{HADOOP\_HOME\}/lib下添加该包。重启集群，一般是比较原始的方法。  
>    
> 2、把jar包传到集群上，命令如下
> 
>     hadoop fs -put mysql-connector-java-5.1.31.jar /lib/mysql
> 
>  在mr程序提交job前，添加如下两个语句中一个就行
> 
>     // 第一个
>     DistributedCache.addFileToClassPath(new Path(“hdfs://ljc:9000/lib/mysql/mysql-connector-java-5.1.31.jar”), conf);
> 
> 这条语句不推荐使用了，建议使用下面这条语句
> 
>     // 第二个
>     job.addArchiveToClassPath(new Path("hdfs://ljc:9000/lib/mysql/mysql-connector-java-5.1.31.jar"));
> 
> 注意：用这种方式，在本地运行，依然报“java.io.IOException: com.mysql.jdbc.Driver”，但放到hadoop运行环境就可以啦  
>    
> 3、把依赖的jar打到项目中，然后配置MANIFEST.MF文件中Class-Path选项  
> ![在这里插入图片描述][20210420100052282.png]  
> 具体配置，请参考[“通过指定manifest.mf文件的打包”][manifest.mf]

--------------------

> 本文系转载内容  
> 原文地址：[https://www.cnblogs.com/codeOfLife/p/5464613.html][https_www.cnblogs.com_codeOfLife_p_5464613.html]

> 下一篇：[自定义 OutputFormat 输出，指定数据写入到不同文件下][OutputFormat]

[Hadoop]: https://blog.csdn.net/lzb348110175/article/details/115672431
[20210420095545646.png]: /images/20221022/709f27a9b8d147e8bc073d14061087d2.png
[202104200956206.png]: /images/20221022/a4c89fba8a884eafb9b99c4a47649a4d.png
[20210420100052282.png]: /images/20221022/b866095056de4d13a8f6198013560197.png
[manifest.mf]: https://gitee.com/ljc520313/notesummary/tree/master/computersoftware/software/jdt/eclipse/pack/%E9%80%9A%E8%BF%87%E6%8C%87%E5%AE%9Amanifest.mf%E6%96%87%E4%BB%B6%E7%9A%84%E6%89%93%E5%8C%85
[https_www.cnblogs.com_codeOfLife_p_5464613.html]: https://www.cnblogs.com/codeOfLife/p/5464613.html
[OutputFormat]: https://blog.csdn.net/lzb348110175/article/details/115892660