DBInputFormat的使用

  1 package InputFormat;
  2 
  3 import java.io.DataInput;
  4 import java.io.DataOutput;
  5 import java.io.IOException;
  6 import java.net.URI;
  7 import java.sql.PreparedStatement;
  8 import java.sql.ResultSet;
  9 import java.sql.SQLException;
 10 
 11 
 12 import org.apache.hadoop.conf.Configuration;
 13 import org.apache.hadoop.fs.FileSystem;
 14 import org.apache.hadoop.fs.Path;
 15 import org.apache.hadoop.io.LongWritable;
 16 import org.apache.hadoop.io.NullWritable;
 17 import org.apache.hadoop.io.Text;
 18 import org.apache.hadoop.io.Writable;
 19 import org.apache.hadoop.mapreduce.Job;
 20 import org.apache.hadoop.mapreduce.Mapper;
 21 import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
 22 import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
 23 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
 24 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 25 
 26 /**
 27  * 将Mysql的驱动包放在TaskTracker的lib下 重启集群
 28  * @author Administrator
 29  *
 30  */
 31 
 32 public class DBInputFormatApp {
 33     public static void main(String[] args) throws Exception {
 34         
 35         Configuration conf = new Configuration();
 36         
 37         DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://hadoop:3306/test", "root", "123456");
 38         
 39         FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), conf);
 40         boolean delete = fileSystem.delete(new Path("hdfs://hadoop:9000/out"), true);
 41         
 42         if (delete) {
 43             System.out.println("==============ooooooooookkkkkkkkkkkkkkkkkkkkkkks==================");
 44             
 45         }
 46         
 47         
 48         Job job = new Job(conf, DBInputFormatApp.class.getSimpleName());
 49         job.setJarByClass(DBInputFormatApp.class);
 50 
 51         job.setInputFormatClass(DBInputFormat.class);
 52         DBInputFormat.setInput(job, MyUser.class, "myuser", null, null, "id","name");
 53         job.setMapperClass(MyDBMapper.class);
 54         job.setMapOutputKeyClass(Text.class);
 55         job.setOutputValueClass(NullWritable.class);
 56         
 57         job.setNumReduceTasks(0);
 58         job.setOutputKeyClass(Text.class);
 59         job.setOutputValueClass(NullWritable.class);
 60         
 61         FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop:9000/out"));
 62         
 63         job.waitForCompletion(true);
 64         
 65         
 66     }
 67     
 68     public static class MyDBMapper extends Mapper<LongWritable, MyUser, Text, NullWritable>{
 69         @Override
 70         protected void map(LongWritable key, MyUser value,
 71                 org.apache.hadoop.mapreduce.Mapper<LongWritable, MyUser, Text, NullWritable>.Context context)
 72                 throws IOException, InterruptedException {
 73             
 74             context.write(new Text(value.toString()), NullWritable.get());
 75             
 76         }
 77         
 78     }
 79 
 80     
 81     public static class MyUser implements Writable,DBWritable{
 82         
 83         int id;
 84         String name;
 85 
 86         @Override
 87         public void write(PreparedStatement statement) throws SQLException {
 88             statement.setInt(1, id);
 89             statement.setString(2, name);
 90             
 91         }
 92 
 93         @Override
 94         public void readFields(ResultSet resultSet) throws SQLException {
 95             this.id=resultSet.getInt(1);
 96             this.name=resultSet.getString(2);
 97             
 98         }
 99 
100         @Override
101         public void write(DataOutput out) throws IOException {
102             out.writeInt(id);
103             Text.writeString(out, name);
104             
105         }
106 
107         @Override
108         public void readFields(DataInput in) throws IOException {
109             this.id=in.readInt();
110             this.name=Text.readString(in);
111             
112         }
113 
114         @Override
115         public String toString() {
116             
117             return  id + ", \t" + name ;
118         }
119         
120         
121         
122     }
123     
124 }