MapReduce 程序mysql JDBC驱动类找不到原因及学习hadoop写入数据到Mysql数据库的方法

报错：ClassNotFoundException: com.mysql.jdbc.Driver

需求描述：

hadoop需要动态加载个三方jar包（比如mysql JDBC 驱动包），是在MR结束后，使用hadoop 的filesystem系统读取HDFS文件，调用JDBC驱动类插入数据库，但是运行时报错找不到驱动类。

第一个方法：加到HADOOP_HOME/lib下不可行，集群需要重启（集群再用，队列有任务进行中）。
第二个方法：job2.addFileToClassPath(file)和DistributedCache.addFileToClassPath()

以及利用hadoop jar xx.jar -libjars $yourpath/mysql-connector-java-5.0.3-bin.jar <other args.....> 这种原理上是一样的，

这种方式加入了Mapreduce的classpath（最后是加载到了每个节点 map container 或reduce container的JVM的classpath中），MR结束后通过filesystem调用JDBC，是不可行的，我的代码在Diver类中调用，即客户端拿不到这个类。
第三个方法：所有依赖三方包和MR程序打包到一个jar中,打包完接近180MB。jar包太大了。
第四个方法：依赖的第三方jar打包到lib中，本次任务不需要的jar包可以全删掉 (这里注意：工程的jar和集群jar版本不一致可能导致冲突)。可以运行，导出的lib也小。

总结一下：

1.报错Diver类里缺少三方包，扔到lib下（不是鼠标拖进lib目录下,是打包打进去）

2.报错Map或Reduce缺少三方包，分发到集群即可；可以用上面第二个方法中的2个方法（job.addFileToClassPath()这个方法好像只有hadoop 2.7版本以上才有）或者直接用-libjars $yourpath/your-jar.jar （一定要写在其他参数前面）把三方包分发到集群

other：

后来在网上搜了一下，发现可以直接使用hadoop的DBOutputFormat 和DBInputFormat类来直接对数据库进行操作。

Diver类中的设置项：

　　　　　conf.set(DBConfiguration.DRIVER_CLASS_PROPERTY,"com.mysql.jdbc.Driver");  
        conf.set(DBConfiguration.URL_PROPERTY,  
                        "jdbc:mysql://x.x.x.x:3306/test");  
        conf.set(DBConfiguration.USERNAME_PROPERTY, "xxx");  
        conf.set(DBConfiguration.PASSWORD_PROPERTY, "xxxx"); 

        //这三行代码是本机测试用的
//        conf.set(DBConfiguration.URL_PROPERTY,  
//        "jdbc:mysql://localhost:3306/test"); 
//        conf.set(DBConfiguration.PASSWORD_PROPERTY, "xxx"); 
        Job job3 = Job.getInstance(conf,"step3:insertToMySqlDB");
        job3.setJarByClass(ECMerchantMapReduceV2.class);
        job3.setMapOutputKeyClass(DBOutputKey.class);
        job3.setMapOutputValueClass(NullWritable.class);
        job3.setMapperClass(ECMerchantThirdMap.class);
        job3.setOutputFormatClass(DBOutputFormat.class);
        job3.setNumReduceTasks(0);
        job3.addFileToClassPath(new Path("/xxxx/mysql-connector-java-5.0.3-bin.jar"));
        DBOutputFormat.setOutput(job3, "yourtablename", "column1","column2","column3","column4");
        FileInputFormat.addInputPath(job3, new Path(otherArgs[3]));
        if(!job3.waitForCompletion(true)) return ;

public static class DBOutputKey implements WritableComparable<DBOutputKey>,DBWritable {

        private String city = "";
        private String ec="";
        private String account="";
        private Date date2=null;　　　　
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        public DBOutputKey() {
            
        }

        public DBOutputKey(String city, String ec, String account, Date date2) {
            super();
            this.city = city;
            this.ec = ec;
            this.account = account;
            this.date2 = date2;
        }

        public void set(String city, String ec, String account, Date date2) {
            this.city = city;
            this.ec = ec;
            this.account = account;
            this.date2 = date2;
        }


        @Override
        public void readFields(DataInput in) throws IOException {
            this.city = in.readUTF();
            this.ec = in.readUTF();
            this.account = in.readUTF();
            try {
                this.date2 = new Date(format.parse(in.readUTF()).getTime());
            } catch (ParseException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeUTF(city);
            out.writeUTF(ec);
            out.writeUTF(account);
            out.writeUTF(date2.toString());
        }

        @Override
        public int compareTo(DBOutputKey other) {
            if (this.city.compareTo(other.city) != 0) {
                return this.city.compareTo(other.city);
            } else if (!this.ec .equals(other.ec) ) {
                return ec .compareTo(other.ec);
            } else if (!this.account.equals(other.account)) {
                return account.compareTo(other.account);
            }else {
                return 0;
            }
        }

        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this);
        }
//下面这2个方法是DBWritable接口的方法
        @Override
        public void write(PreparedStatement statement) throws SQLException {
            statement.setString(1, city);
            statement.setString(2, ec);
            statement.setString(3, account);
            statement.setDate(4, date2);
        }

        @Override
        public void readFields(ResultSet resultSet) throws SQLException {
            this.city=resultSet.getString(1);
            this.ec=resultSet.getString(2);
            this.account=resultSet.getString(3);
            this.date2=resultSet.getDate(4);
        }
    }

注意：
1.DBOutputFormat写入数据库，是按照key输出的，没有value，key要实现WritableComparable和DBWritable接口；
2.数据库表字段有个为Sql.date类型private Date date2=null，readFields(DataInput in)和write(DataOutput out)输入输出流不支持Sql.Date类型，通过文中红色代码部分转换格式，即可使Hadoop序列化的时候支持sql.Date数据类型。

    public static class ECMerchantThirdMap extends Mapper<LongWritable, Text, DBOutputKey, NullWritable> {
        DBOutputKey dbKey=new DBOutputKey();
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        String date;
        @Override
        protected void setup(Mapper<LongWritable, Text, DBOutputKey, NullWritable>.Context context)
                throws IOException, InterruptedException {
            date=context.getConfiguration().get("eradiusDate");
        }
        public void map(LongWritable ikey, Text ivalue, Context context) throws IOException, InterruptedException {
            String[] strs = ivalue.toString().split("\\s+");
            if(strs.length!=3) return;
            
            Date date2=null;
            try {
                date2 = new Date(format.parse(date).getTime());
            } catch (ParseException e) {
                e.printStackTrace();
            }
            dbKey.set(strs[0], strs[1], strs[2], date2);
            context.write(dbKey, NullWritable.get());
        }

    }

Map类直接输出key就可以，不需要reduce类，可以设置reduce数量为0；

在Diver类中job3.addFileToClassPath(new Path("/xxxx/mysql-connector-java-5.0.3-bin.jar"));

测试后发现JDBC驱动包在ECMerchantThirdMap这个类写数据到Mysql数据库时是可以调用的，

而通过job3.addFileToClassPath（）方法添加JDBC驱动包，在客户端是无法调用的。

所以该方法是添加jar包到运行Map或Reduce的节点的Classpath中。