java - Hadoop MapReduce程序连接到数据库

Question

当我在Eclipse中使用hadoop运行时执行程序时，我得到了一个Mapper类未发现错误，但是当我通过创建jar来执行时，它的工作正常

我检查了所有存在于classpath中的类。
我需要将此MapReduce程序从eclipse自身连接到后端过程。

这是我的代码

public class DBDriver extends Configured implements Tool  {

    public int run(String[] args) throws Exception {
        // TODO Auto-generated method stub
        //Configuration conf = new Configuration();

        String dbName = "bigdata";
        String tableName = "cpr_transaction";

        Job job = Job.getInstance(getConf());

        HCatInputFormat.setInput(job, dbName, tableName);

        job.setJarByClass(DBDriver.class);
        //job.setJar("/home/murali/workspace/hadoop/DBReader/target/DBReader-0.0.1-SNAPSHOT.jar");
        job.setMapperClass(DBMapper.class);
        job.setReducerClass(DBReducer.class);

        job.setInputFormatClass(HCatInputFormat.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        FileOutputFormat.setOutputPath(job, new Path("/user/murali/output/HiveIO"));

        return(job.waitForCompletion(true)? 0:1);
    }

    public static void main(String[] args) throws Exception {
        int exitStatus = ToolRunner.run(new DBDriver(), args);
        System.exit(exitStatus);
    }   

    public static class DBMapper extends Mapper<Writable, HCatRecord, Text, Text> {

        String amount;
        String outlet;

        public void map(Writable key, HCatRecord value, Mapper<Writable, HCatRecord, Text, Text>.Context context) throws IOException, InterruptedException {

            amount = (String) value.get(8);
            outlet = (String) value.get(3);

            context.write(new Text(outlet), new Text(amount));
        }
    }

    public class DBReducer extends Reducer<Text, Text, Text, DoubleWritable> {

        public void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {

    double totalAmount = 0.0;

            for (Text text : values) {
                String textStr = text.toString();
                textStr = textStr.replaceAll(",", "");
                if (!(textStr.equalsIgnoreCase(null))) {
                    totalAmount += Double.parseDouble(textStr.toString());
                }
            }

            context.write(key, new DoubleWritable(totalAmount));
        }
    }
}

错误，我得到的是

14/12/11 14:57:04 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
14/12/11 14:57:04 INFO hive.metastore: Trying to connect to metastore with URI thrift://192.168.0.65:9083
14/12/11 14:57:04 INFO hive.metastore: Connected to metastore.
14/12/11 14:57:05 INFO client.RMProxy: Connecting to ResourceManager at hdm/192.168.0.93:8032
14/12/11 14:57:06 WARN mapreduce.JobSubmitter: No job jar file set.  User classes may not be found. See Job or Job#setJar(String).
14/12/11 14:57:06 INFO Configuration.deprecation: mapred.input.dir is deprecated. Instead, use mapreduce.input.fileinputformat.inputdir
14/12/11 14:57:06 INFO mapred.FileInputFormat: Total input paths to process : 1
14/12/11 14:57:06 INFO mapreduce.JobSubmitter: number of splits:1
14/12/11 14:57:06 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1418017495640_0166
14/12/11 14:57:06 INFO mapred.YARNRunner: Job jar is not present. Not adding any jar to the list of resources.
14/12/11 14:57:06 INFO impl.YarnClientImpl: Submitted application application_1418017495640_0166
14/12/11 14:57:06 INFO mapreduce.Job: The url to track the job: http://hdm:8088/proxy/application_1418017495640_0166/
14/12/11 14:57:06 INFO mapreduce.Job: Running job: job_1418017495640_0166
14/12/11 14:57:11 INFO mapreduce.Job: Job job_1418017495640_0166 running in uber mode : false
14/12/11 14:57:11 INFO mapreduce.Job:  map 0% reduce 0%
14/12/11 14:57:14 INFO mapreduce.Job: Task Id : attempt_1418017495640_0166_m_000000_0, Status : FAILED
Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1905)
    at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:722)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1614)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:163)
Caused by: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1811)
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1903)
    ... 8 more

14/12/11 14:57:18 INFO mapreduce.Job: Task Id : attempt_1418017495640_0166_m_000000_1, Status : FAILED
Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1905)
    at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:722)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1614)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:163)
Caused by: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1811)
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1903)
    ... 8 more

14/12/11 14:57:22 INFO mapreduce.Job: Task Id : attempt_1418017495640_0166_m_000000_2, Status : FAILED
Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1905)
    at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:722)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1614)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:163)
Caused by: java.lang.ClassNotFoundException: Class com.cosn.DBReader.DBDriver$DBMapper not found
    at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1811)
    at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1903)
    ... 8 more

14/12/11 14:57:27 INFO mapreduce.Job:  map 100% reduce 100%
14/12/11 14:57:27 INFO mapreduce.Job: Job job_1418017495640_0166 failed with state FAILED due to: Task failed task_1418017495640_0166_m_000000
Job failed as tasks failed. failedMaps:1 failedReduces:0

14/12/11 14:57:27 INFO mapreduce.Job: Counters: 9
    Job Counters 
        Failed map tasks=4
        Launched map tasks=4
        Other local map tasks=3
        Data-local map tasks=1
        Total time spent by all maps in occupied slots (ms)=8314
        Total time spent by all reduces in occupied slots (ms)=0
        Total time spent by all map tasks (ms)=8314
        Total vcore-seconds taken by all map tasks=8314
        Total megabyte-seconds taken by all map tasks=8513536

Answer 1

尝试在 reducer 中也包括“静态”

public static class DBReducer extends Reducer<Text, Text, Text, DoubleWritable> {

        public void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {