Hadoop MapReduce同步MySQL数据到HDFS提示少驱动JAR包

近期因为一个需求，要把MySQL中的数据按照指定格式同步到HDFS中，以作后续分析使用(实际可以使用sqoop，但这里暂不考虑)，实现方法是使用MapReduce程序来实现，但在使用过程中，遇到了java.lang.ClassNotFoundException: com.mysql.jdbc.Driver错误，各种测试、搜索无果。

下面贴出代码

LoginDaysMapReduce

package com.zlikun.learning.mysql.logins;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * 登录日期mapreduce程序，输出$userId + $date到HDFS中，以供后续使用
 * @author zlikun <zlikun-dev@hotmail.com>
 * @date 2018-01-16 11:39
 */
public class LoginDaysMapReduce extends Configured implements Tool {

    /**
     * 将数据库中数据映射为：$userId + $date(与1970的相关天数) 格式
     */
    public static class LoginDaysMapper extends Mapper<LongWritable, TblRecord, Text, NullWritable> {

        @Override
        protected void map(LongWritable key, TblRecord value, Context context) throws IOException, InterruptedException {
            if (value.days > 0) {
                context.write(new Text(value.toString()), NullWritable.get());
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(this.getConf(), this.getClass().getSimpleName());
        job.setJarByClass(LoginDaysMapReduce.class);
        // 配置MapReduce程序
        job.setMapperClass(LoginDaysMapper.class);
        job.setNumReduceTasks(0);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setInputFormatClass(DBInputFormat.class);
        FileOutputFormat.setOutputPath(job, new Path("hdfs://zlikun:9000/login_days/example"));
        job.addArchiveToClassPath(new Path("/lib/mysql/mysql-connector-java-5.1.45.jar"));
        DBConfiguration.configureDB(job.getConfiguration(),
                "com.mysql.jdbc.Driver",
                "jdbc:mysql://192.168.xxx.xxx:3306/logins",
                "root",
                "******");
        String [] fields = {"USER_ID", "LOGIN_TIME"};
        DBInputFormat.setInput(job, TblRecord.class,
                "TBL_LOGIN_LOG", null, null, fields);
        return job.waitForCompletion(true) ? 1 : 0 ;
    }

    /**
     * args[0]  jdbcUrl
     * args[1]  username
     * args[2]  password
     * args[3]  tableName
     * args[4]  outputPath
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        int status = ToolRunner.run(conf, new LoginDaysMapReduce(), args);
        System.exit(status);
    }

}

TblRecord

package com.zlikun.learning.mysql.logins;

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;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.ZoneId;

/**
 * MySQL 表记录，用于读写MySQL表数据使用
 * @author zlikun <zlikun-dev@hotmail.com>
 * @date 2018-01-16 11:53
 */
public class TblRecord implements Writable, DBWritable {

    Long userId ;   // 用户ID
    Long days;      // 登录日期距1970/1/1天数(方便后续按序列处理数据，以统计连续登录天数)

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(this.userId);
        dataOutput.writeLong(this.days);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.userId = dataInput.readLong();
        this.days = dataInput.readLong();
    }

    @Override
    public void write(PreparedStatement statement) throws SQLException {
        statement.setLong(1, this.userId);
        statement.setLong(2, this.days);
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.userId = resultSet.getLong(1);
        Timestamp loginTime = resultSet.getTimestamp(2);
        if (loginTime != null) {
            this.days = Instant.ofEpochMilli(loginTime.getTime()).atZone(ZoneId.systemDefault()).toLocalDate().toEpochDay();
        } else {
            this.days = -1L;
        }
    }

    @Override
    public String toString() {
        return this.userId + ":" + this.days;
    }
}

打包后(mr.jar)，在Hadoop（版本：2.7.4）上执行，报出如下错误：

$ hadoop jar mr.jar com.zlikun.learning.mysql.logins.LoginDaysMapReduce
18/01/16 17:17:14 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
18/01/16 17:17:15 INFO mapreduce.JobSubmitter: Cleaning up the staging area /tmp/hadoop-yarn/staging/zlikun/.staging/job_1516093190077_0003
Exception in thread "main" java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
    at org.apache.hadoop.mapreduce.lib.db.DBInputFormat.setConf(DBInputFormat.java:171)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:76)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:136)
    at org.apache.hadoop.mapreduce.JobSubmitter.writeNewSplits(JobSubmitter.java:299)
    at org.apache.hadoop.mapreduce.JobSubmitter.writeSplits(JobSubmitter.java:318)
    at org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:196)
    at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1290)
    at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1287)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:422)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1746)
    at org.apache.hadoop.mapreduce.Job.submit(Job.java:1287)
    at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1308)
    at com.zlikun.learning.mysql.logins.LoginDaysMapReduce.run(LoginDaysMapReduce.java:62)
    at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)
    at com.zlikun.learning.mysql.logins.LoginDaysMapReduce.main(LoginDaysMapReduce.java:76)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:136)
Caused by: java.lang.RuntimeException: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
    at org.apache.hadoop.mapreduce.lib.db.DBInputFormat.createConnection(DBInputFormat.java:205)
    at org.apache.hadoop.mapreduce.lib.db.DBInputFormat.setConf(DBInputFormat.java:164)
    ... 21 more
Caused by: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at java.lang.Class.forName0(Native Method)
    at java.lang.Class.forName(Class.java:264)
    at org.apache.hadoop.mapreduce.lib.db.DBConfiguration.getConnection(DBConfiguration.java:148)
    at org.apache.hadoop.mapreduce.lib.db.DBInputFormat.createConnection(DBInputFormat.java:198)
    ... 22 more

实际上驱动JAR包已经上传至HDFS上了

$ hdfs dfs -ls /lib/mysql/mysql-connector-java-5.1.45.jar
-rw-r--r--   1 zlikun supergroup     999808 2018-01-16 16:35 /lib/mysql/mysql-connector-java-5.1.45.jar

Hadoop MapReduce同步MySQL数据到HDFS提示少驱动JAR包

下面贴出代码

你尚未登录，登录后可以

Spring Boot + PostgreSQL 查询报错：Error attempting to get column 'xxx' from result set解决办法？

hive需要在配置文件中指定hadoop集群namenode的ip和端口吗，如果不需要，那他们是如何建立连接的呢？

Shuffle错误对MapReduce数据处理结果是否会有影响？