DBWritable的使用

Posted 2020-10-14 求知cvip

package com.neworigin.db;

 

import java.io.IOException;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

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.Reducer;

import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;

import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;

import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;

import org.apache.hadoop.mapreduce.lib.db.DBWritable;

import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

public class DBJob {

public static class TableUers implements DBWritable{//表的映射

 

int id;

String name;

int age;

 

   public int getId() {

   return id;

}

 

public void setId(int id) {

   this.id = id;

}

 

public String getName() {

   return name;

}

 

public void setName(String name) {

   this.name = name;

}

 

public int getAge() {

   return age;

}

 

public void setAge(int age) {

   this.age = age;

}

 

   public TableUers(int id, String name, int age) {

   this.id = id;

   this.name = name;

   this.age = age;

}

 

   public TableUers() {

   }

//将表的映射对象中的属性设置进（写入数据库）

   public void write(PreparedStatement statement) throws SQLException {

      statement.setInt(1, id);

      statement.setString(2, name);

      statement.setInt(3, age);

   }

//从结果集通过与字段匹配的赋值给表的映射对象（读取数据库）

   public void readFields(ResultSet resultSet) throws SQLException {//获取表中的属性

      this.id=resultSet.getInt("id");

      this.name=resultSet.getString("name");

      this.age=resultSet.getInt("age");

     

   }

 

   @Override

   public String toString() {

      return  id + "  " + name + " " + age;

   }

  

}

public static class MyMapper extends Mapper<Object ,TableUers,TableUers,NullWritable>{

 

   protected void map(Object key, TableUers value, Mapper<Object, TableUers, TableUers, NullWritable>.Context context)

          throws IOException, InterruptedException {

      String s = value.toString();

      System.out.println(s);

     

     

      context.write(value,NullWritable.get());

     

   }

  

}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

   Path out=new Path("file:///F:/安装/java工程/MR/data/db/out");

   Job job = Job.getInstance();

   Configuration conf = job.getConfiguration();

/* FileSystem fs =FileSystem.get(conf);

   if(fs.exists(out))

   {

      fs.delete(out);

   }*/

  

   job.setInputFormatClass(DBInputFormat.class);

   DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://127.0.0.1/mydb","root","123456");

   DBInputFormat.setInput(job, TableUers.class, "select id,name,age from user", "select count(*) from user");

  

   job.setOutputFormatClass(DBOutputFormat.class);

   DBOutputFormat.setOutput(job, "aaa", "id","name","age");

  

  

   job.setMapperClass(MyMapper.class);

   job.setMapOutputKeyClass(TableUers.class);

   job.setMapOutputValueClass(NullWritable.class);

  

  

   job.setOutputKeyClass(TableUers.class);

   job.setOutputValueClass(NullWritable.class);

   job.setNumReduceTasks(0);

  

   //FileOutputFormat.setOutputPath(job, out);

  

   job.waitForCompletion(true);

}

}