向MySQL数据库写入数据时map reduce作业中的类强制转换异常

Question

我正在尝试一个map reduce作业来加载mysql数据库中的数据，但是我面临一个类强制转换异常错误，下面是我使用的过程：

我首先创建了一个实现可写和可写接口的DBOutputWritable类。然后我使用我的reduce作业将数据写入数据库，但是当我运行该作业时，它会因为出现错误而失败：

java.lang.ClassCastException: com.amalwa.hadoop.DataBaseLoadMapReduce.DBOutputWritable cannot be cast to org.apache.hadoop.mapreduce.lib.db.DBWritable

at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat$DBRecordWriter.write(DBOutputFormat.java:66)
    at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:601)
    at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
    at com.amalwa.hadoop.DataBaseLoadMapReduce.DBMapReduce$DBReducer.reduce(DBMapReduce.java:58)
    at com.amalwa.hadoop.DataBaseLoadMapReduce.DBMapReduce$DBReducer.reduce(DBMapReduce.java:53)
    at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
    at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:663)
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:426)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
    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:1132)
    at org.apache.hadoop.mapred.Child.main(Child.java:249)

我很难弄清楚，如果我的类实现了使用映射reduce作业向DB写入所需的接口，那么为什么会出现类强制转换异常。我正在实现所需的所有功能。

谢了。

DBOutputWritable
package com.amalwa.hadoop.DataBaseLoadMapReduce;

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


public class DBOutputWritable implements Writable, DBWritable{

 private String keyValue;
 private String response;

 public DBOutputWritable(String keyValue, String response){
 this.keyValue = keyValue;
 this.response = response; 
 }

 public void readFields(DataInput resultSet) throws IOException {

 }

 public void readFields(ResultSet resultSet) throws SQLException {
 keyValue = resultSet.getString(1);
 response = resultSet.getString(2);
 }

 public void write(PreparedStatement preparedStatement) throws SQLException {
 preparedStatement.setString(1, keyValue);
 preparedStatement.setString(2, response);
 }

 public void write(DataOutput dataOutput) throws IOException {

 }

}

减速器：

public static class DBReducer extends Reducer<Text, Text, DBOutputWritable, NullWritable>{

 public void reduce(Text requestKey, Iterable<Text> response, Context context){
 for(Text responseSet: response){
 try{
 context.write(new DBOutputWritable(requestKey.toString(), responseSet.toString()), NullWritable.get());
 }catch(IOException e){
 System.err.println(e.getMessage());
 }
 catch(InterruptedException e){
 System.err.println(e.getMessage());
 }
 }
 }
 }

    public void map(LongWritable key, Text value, Context context) throws IOException{
        String tweetInfo = value.toString();
        String[] myTweetData = tweetInfo.split(",", 2);
        String requestKey = myTweetData[0];
        String response = myTweetData[1];
        try {
            context.write(new Text(requestKey), new Text(response));
        } catch (InterruptedException e) {
            System.err.println(e.getMessage());;
        }
    }
}

public static void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
        DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://ec2-54-152-254-194.compute-1.amazonaws.com/TWEETS", "user", "password");
        Job job = new Job(conf);
        job.setJarByClass(DBMapReduce.class);
        job.setMapperClass(DBMapper.class);
        job.setReducerClass(DBReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(DBOutputWritable.class);
        job.setOutputValueClass(NullWritable.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(DBOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[1]));
        DBOutputFormat.setOutput(job, "TWEET_INFO", new String[] { "REQUESTKEY", "TWEET_DETAILS" });
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

Answer 1

似乎是将旧的(org.apache.hadoop.mapred.*)和新的(org.apache.hadoop.MapReduce.*)MapReduce API混合在一起，这导致了冲突。我怀疑您的dbreducer类正在从新API扩展reducer类，但您的dboutputwritable正在从旧API实现dbwritable。

您应该在实现中只选择一个API，这意味着所有导入的MapReduce类型都以相同的包前缀开头。

注意，通常在使用旧API时实现MapReduce接口，在使用新API时扩展MapReduce基类。