博客
关于我
强烈建议你试试无所不能的chatGPT，快点击我
如何使用MR来读取数据库的数据，并写入HDFS上
阅读量：4185 次
发布时间：2019-05-26

本文共 23782 字，大约阅读时间需要 79 分钟。

在我们的一些应用程序中，常常避免不了要与数据库进行交互，而在我们的hadoop中，有时候也需要和数据库进行交互，比如说，数据分析的结果存入数据库，或者是，读取数据库的信息写入HDFS上，不过直接使用MapReduce操作数据库，这种情况在现实开发还是比较少，一般我们会采用Sqoop来进行数据的迁入，迁出，使用Hive分析数据集，大多数情况下，直接使用Hadoop访问关系型数据库，可能产生比较大的数据访问压力，尤其是在数据库还是单机的情况下，情况可能更加糟糕，在集群的模式下压力会相对少一些。

那么，今天散仙就来看下，如何直接使用Hadoop1.2.0的MR来读写操作数据库，hadoop的API提供了DBOutputFormat和DBInputFormat这两个类，来进行与数据库交互，除此之外，我们还需要定义一个类似JAVA Bean的实体类，来与数据库的每行记录进行对应，通常这个类要实现Writable和DBWritable接口，来重写里面的4个方法以对应获取每行记录里面的各个字段信息。
下面，我们先来看下如何使用MR来读取数据库的数据，并写入HDFS上，
数据表的截图如下所示，
实体类定义代码：

Java代码  
  1. package com.qin.operadb;  
  2.   
  3. import java.io.DataInput;  
  4. import java.io.DataOutput;  
  5. import java.io.IOException;  
  6. import java.sql.PreparedStatement;  
  7. import java.sql.ResultSet;  
  8. import java.sql.SQLException;  
  9.   
  10. import org.apache.hadoop.io.Text;  
  11. import org.apache.hadoop.io.Writable;  
  12. import org.apache.hadoop.mapreduce.lib.db.DBWritable;  
  13.    
  14. /*** 
  15.  * 封装数据库实体信息 
  16.  * 的记录 
  17.  *  
  18.  * 搜索大数据技术交流群：376932160 
  19.  *  
  20.  * **/  
  21. public class PersonRecoder implements Writable,DBWritable {  
  22.   
  23.     public int id;//对应数据库中id字段  
  24.     public String name;//对应数据库中的name字段  
  25.     public int age;//对应数据库中的age字段  
  26.       
  27.       
  28.       
  29.       
  30.     @Override  
  31.     public void readFields(ResultSet result) throws SQLException {  
  32.       
  33.           
  34.         this.id=result.getInt(1);  
  35.         this.name=result.getString(2);  
  36.         this.age=result.getInt(3);  
  37.           
  38.     }  
  39.   
  40.     @Override  
  41.     public void write(PreparedStatement stmt) throws SQLException {  
  42.           
  43.         stmt.setInt(1, id);  
  44.         stmt.setString(2, name);  
  45.         stmt.setInt(3, age);  
  46.           
  47.     }  
  48.   
  49.     @Override  
  50.     public void readFields(DataInput arg0) throws IOException {  
  51.         // TODO Auto-generated method stub  
  52.         this.id=arg0.readInt();  
  53.         this.name=Text.readString(arg0);  
  54.         this.age=arg0.readInt();  
  55.           
  56.     }  
  57.   
  58.     @Override  
  59.     public void write(DataOutput out) throws IOException {  
  60.         // TODO Auto-generated method stub  
  61.         out.writeInt(id);  
  62.         Text.writeString(out, this.name);  
  63.         out.writeInt(this.age);  
  64.     }  
  65.       
  66.       
  67.     @Override  
  68.     public String toString() {  
  69.         // TODO Auto-generated method stub  
  70.         return "id: "+id+"  年龄: "+age+"   名字："+name;  
  71.     }  
  72.       
  73.   
  74. }  
package com.qin.operadb;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.Text;import org.apache.hadoop.io.Writable;import org.apache.hadoop.mapreduce.lib.db.DBWritable; /*** * 封装数据库实体信息 * 的记录 *  * 搜索大数据技术交流群：376932160 *  * **/public class PersonRecoder implements Writable,DBWritable {	public int id;//对应数据库中id字段	public String name;//对应数据库中的name字段	public int age;//对应数据库中的age字段					@Override	public void readFields(ResultSet result) throws SQLException {					this.id=result.getInt(1);		this.name=result.getString(2);		this.age=result.getInt(3);			}	@Override	public void write(PreparedStatement stmt) throws SQLException {				stmt.setInt(1, id);		stmt.setString(2, name);		stmt.setInt(3, age);			}	@Override	public void readFields(DataInput arg0) throws IOException {		// TODO Auto-generated method stub		this.id=arg0.readInt();		this.name=Text.readString(arg0);		this.age=arg0.readInt();			}	@Override	public void write(DataOutput out) throws IOException {		// TODO Auto-generated method stub		out.writeInt(id);		Text.writeString(out, this.name);		out.writeInt(this.age);	}			@Override	public String toString() {		// TODO Auto-generated method stub		return "id: "+id+"  年龄: "+age+"   名字："+name;	}	}

MR类的定义代码，注意是一个Map Only作业：

Java代码  
  1. package com.qin.operadb;  
  2.   
  3. import java.io.IOException;  
  4.   
  5. import org.apache.hadoop.conf.Configuration;  
  6. import org.apache.hadoop.fs.FileSystem;  
  7. import org.apache.hadoop.fs.Path;  
  8. import org.apache.hadoop.io.LongWritable;  
  9. import org.apache.hadoop.io.Text;  
  10. import org.apache.hadoop.mapred.JobConf;  
  11. import org.apache.hadoop.mapred.lib.IdentityReducer;  
  12. import org.apache.hadoop.mapreduce.Job;  
  13. import org.apache.hadoop.mapreduce.Mapper;  
  14. import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;  
  15. import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;  
  16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
  17.   
  18. public class ReadMapDB {  
  19.       
  20.       
  21.       
  22.     /** 
  23.      * Map作业读取数据记录数 
  24.      *  
  25.      * **/  
  26.     private static class DBMap extends Mapper<LongWritable, PersonRecoder    , LongWritable, Text>{  
  27.         @Override  
  28.         protected void map(LongWritable key, PersonRecoder value,Context context)  
  29.                 throws IOException, InterruptedException {  
  30.            
  31.             context.write(new LongWritable(value.id), new Text(value.toString()));  
  32.            
  33.         }  
  34.     }  
  35.       
  36.     public static void main(String[] args)throws Exception {  
  37.           
  38.           
  39.          JobConf conf=new JobConf(ReadMapDB.class);  
  40.         //Configuration conf=new Configuration();  
  41.        //   conf.set("mapred.job.tracker","192.168.75.130:9001");  
  42.         //读取person中的数据字段  
  43.      // conf.setJar("tt.jar");  
  44.            
  45.         //注意这行代码放在最前面，进行初始化，否则会报  
  46.         DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test", "root", "qin");  
  47.           
  48.         /**要读取的字段信息**/  
  49.         String fileds[]=new String[]{
    "id","name","age"};  
  50.         /**Job任务**/  
  51.         Job job=new Job(conf, "readDB");  
  52.         System.out.println("模式：  "+conf.get("mapred.job.tracker"));  
  53.           
  54.         /**设置数据库输入格式的一些信息**/  
  55.         DBInputFormat.setInput(job, PersonRecoder.class, "person", null, "id", fileds);  
  56.         /***设置输入格式*/  
  57.         job.setInputFormatClass(DBInputFormat.class);  
  58.         job.setOutputKeyClass(LongWritable.class);  
  59.         job.setOutputValueClass(Text.class);  
  60.         job.setMapperClass(DBMap.class);  
  61.         String path="hdfs://192.168.75.130:9000/root/outputdb";  
  62.         FileSystem fs=FileSystem.get(conf);  
  63.         Path p=new Path(path);  
  64.         if(fs.exists(p)){  
  65.             fs.delete(p, true);  
  66.             System.out.println("输出路径存在，已删除！");  
  67.         }  
  68.         FileOutputFormat.setOutputPath(job,p );  
  69.         System.exit(job.waitForCompletion(true) ? 0 : 1);    
  70.           
  71.           
  72.            
  73.           
  74.           
  75.           
  76.           
  77.     }  
  78.       
  79.       
  80.       
  81.       
  82.       
  83.       
  84.   
  85. }  
package com.qin.operadb;import java.io.IOException;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.Text;import org.apache.hadoop.mapred.JobConf;import org.apache.hadoop.mapred.lib.IdentityReducer;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;public class ReadMapDB {				/**	 * Map作业读取数据记录数	 * 	 * **/	private static class DBMap extends Mapper
    
     {		@Override		protected void map(LongWritable key, PersonRecoder value,Context context)				throws IOException, InterruptedException {		 			context.write(new LongWritable(value.id), new Text(value.toString()));		 		}	}		public static void main(String[] args)throws Exception {						 JobConf conf=new JobConf(ReadMapDB.class);		//Configuration conf=new Configuration();	   //	conf.set("mapred.job.tracker","192.168.75.130:9001");		//读取person中的数据字段	 //	conf.setJar("tt.jar");		 		//注意这行代码放在最前面，进行初始化，否则会报		DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test", "root", "qin");				/**要读取的字段信息**/		String fileds[]=new String[]{"id","name","age"};		/**Job任务**/		Job job=new Job(conf, "readDB");		System.out.println("模式：  "+conf.get("mapred.job.tracker"));				/**设置数据库输入格式的一些信息**/		DBInputFormat.setInput(job, PersonRecoder.class, "person", null, "id", fileds);		/***设置输入格式*/		job.setInputFormatClass(DBInputFormat.class);		job.setOutputKeyClass(LongWritable.class);		job.setOutputValueClass(Text.class);		job.setMapperClass(DBMap.class);		String path="hdfs://192.168.75.130:9000/root/outputdb";		FileSystem fs=FileSystem.get(conf);		Path p=new Path(path);		if(fs.exists(p)){			fs.delete(p, true);			System.out.println("输出路径存在，已删除！");		}		FileOutputFormat.setOutputPath(job,p );		System.exit(job.waitForCompletion(true) ? 0 : 1);  						 									}						}

写入到HDFS目录下数据集：
读取相对比较简单，需要注意的第一注意JDBC的驱动jar包要在各个节点上分别上传一份，第二是在main方法里的驱动类的编写顺序，以及数据信息的完整，才是正确连接数据库并读取的关键。
下面来看下，如何使用MR，分析完数据后的结果，写入在数据库中，散仙本篇测试的是一个简单的WordCount的统计。我们先来看下数据库表的信息：
实体类定义代码：

Java代码  
  1. package com.qin.operadb;  
  2.   
  3. import java.io.DataInput;  
  4. import java.io.DataOutput;  
  5. import java.io.IOException;  
  6. import java.sql.PreparedStatement;  
  7. import java.sql.ResultSet;  
  8. import java.sql.SQLException;  
  9.   
  10. import org.apache.hadoop.io.Text;  
  11. import org.apache.hadoop.io.Writable;  
  12. import org.apache.hadoop.mapreduce.lib.db.DBWritable;  
  13.   
  14. public class WordRecoder implements Writable,DBWritable {  
  15.   
  16.      public String word;  
  17.      public int count;  
  18.       
  19.       
  20.     @Override  
  21.     public void readFields(ResultSet rs) throws SQLException {  
  22.           
  23.         this.word=rs.getString(1);  
  24.         this.count=rs.getInt(2);  
  25.           
  26.     }  
  27.   
  28.     @Override  
  29.     public void write(PreparedStatement ps) throws SQLException {  
  30.           
  31.         ps.setString(1, this.word);  
  32.         ps.setInt(2, this.count);  
  33.           
  34.     }  
  35.   
  36.     @Override  
  37.     public void readFields(DataInput in) throws IOException {  
  38.        
  39.           this.word=Text.readString(in);  
  40.           this.count=in.readInt();  
  41.           
  42.     }  
  43.   
  44.     @Override  
  45.     public void write(DataOutput out) throws IOException {  
  46.         
  47.              
  48.         Text.writeString(out, this.word);  
  49.         out.writeInt(count);  
  50.            
  51.           
  52.           
  53.     }  
  54.       
  55.       
  56.   
  57. }  
package com.qin.operadb;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.Text;import org.apache.hadoop.io.Writable;import org.apache.hadoop.mapreduce.lib.db.DBWritable;public class WordRecoder implements Writable,DBWritable {	 public String word;	 public int count;			@Override	public void readFields(ResultSet rs) throws SQLException {				this.word=rs.getString(1);		this.count=rs.getInt(2);			}	@Override	public void write(PreparedStatement ps) throws SQLException {				ps.setString(1, this.word);		ps.setInt(2, this.count);			}	@Override	public void readFields(DataInput in) throws IOException {	 		  this.word=Text.readString(in);		  this.count=in.readInt();			}	@Override	public void write(DataOutput out) throws IOException {	  		   		Text.writeString(out, this.word);		out.writeInt(count);		 					}		}

统计的2个文件的内容所示：
MR的核心类代码：

Java代码  
  1. package com.qin.operadb;  
  2.   
  3. import java.io.IOException;  
  4. import java.util.StringTokenizer;  
  5.   
  6. import org.apache.hadoop.fs.FileSystem;  
  7. import org.apache.hadoop.fs.Path;  
  8. import org.apache.hadoop.io.IntWritable;  
  9. import org.apache.hadoop.io.LongWritable;  
  10. import org.apache.hadoop.io.Text;  
  11. import org.apache.hadoop.mapred.JobConf;  
  12. import org.apache.hadoop.mapreduce.Job;  
  13. import org.apache.hadoop.mapreduce.Mapper;  
  14. import org.apache.hadoop.mapreduce.Reducer;  
  15. import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;  
  16. import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;  
  17. import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;  
  18. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
  19. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
  20. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
  21.    
  22.   
  23. public class WriteMapDB {  
  24.       
  25.       
  26.       
  27.     private static class WMap extends Mapper<LongWritable, Text, Text, IntWritable>{  
  28.           
  29.           
  30.         /*** 
  31.          * Mapper的参数类型介绍 
  32.          * K,V,K,V分别依次代表 
  33.          * Map作业输入类型的K，输入类型的V 
  34.          * 后面两个是输出类型的K，输出类型的V 
  35.          * 后面的两个与 context.write(word, one); 
  36.          * 的两个参数是对应的 
  37.          * **/  
  38.         private Text word=new Text();  
  39.         private IntWritable one=new IntWritable(1);  
  40.           
  41.         @Override  
  42.         protected void map(LongWritable key, Text value,Context context)  
  43.                 throws IOException, InterruptedException {  
  44.        
  45.              String line=value.toString();  
  46.             //处理记事本UTF-8的BOM问题  
  47.   
  48.                 if (line.getBytes().length > 0) {  
  49.   
  50.                     if ((int) line.charAt(0) == 65279) {  
  51.   
  52.                         line = line.substring(1);  
  53.   
  54.                     }  
  55.   
  56.                 }  
  57.                
  58.              StringTokenizer st=new StringTokenizer(line);  
  59.              while(st.hasMoreTokens()){  
  60.                  word.set(st.nextToken());//设置单词  
  61.                  context.write(word, one);  
  62.              }  
  63.               
  64.               
  65.                
  66.         }  
  67.           
  68.           
  69.           
  70.     }  
  71.       
  72.       
  73.     /*** 
  74.      * 由于在reduce中，需要向数据库里写入 
  75.      * 数据，所以跟combine，不能共用 
  76.      *  
  77.      *  
  78.      *  
  79.      * ***/  
  80.     private static class WCombine extends Reducer<Text, IntWritable, Text, IntWritable>{  
  81.           
  82.         @Override  
  83.         protected void reduce(Text text, Iterable<IntWritable> value,Context context)  
  84.                 throws IOException, InterruptedException {  
  85.               
  86.               
  87.             int sum=0;  
  88.               
  89.             for(IntWritable iw:value){  
  90.                 sum+=iw.get();  
  91.             }  
  92.             context.write(text, new IntWritable(sum));  
  93.               
  94.                
  95.         }  
  96.           
  97.           
  98.           
  99.           
  100.     }  
  101.       
  102.       
  103.       
  104.       
  105.       
  106.     /** 
  107.      *  
  108.      * Reduce类 
  109.      *  
  110.      * **/  
  111.     private static class WReduce extends Reducer<Text, IntWritable, WordRecoder, Text>{  
  112.           
  113.           
  114.         @Override  
  115.         protected void reduce(Text key, Iterable<IntWritable> values,Context context)  
  116.                 throws IOException, InterruptedException {  
  117.                
  118.               int sum=0;  
  119.               
  120.               for(IntWritable s:values){  
  121.                   sum+=s.get();  
  122.               }  
  123.                 
  124.                 
  125.               WordRecoder wr=new WordRecoder();  
  126.               wr.word=key.toString();  
  127.               wr.count=sum;  
  128.                 
  129.               //写出到数据库里  
  130.               context.write(wr, new Text());  
  131.               
  132.               
  133.               
  134.               
  135.         }  
  136.           
  137.           
  138.           
  139.     }  
  140.       
  141.       
  142.       
  143.       
  144.       
  145.       
  146.       
  147.     public static void main(String[] args)throws Exception {  
  148.           
  149.   
  150.          JobConf conf=new JobConf(WriteMapDB.class);  
  151.         //Configuration conf=new Configuration();  
  152.        // conf.set("mapred.job.tracker","192.168.75.130:9001");  
  153.         //读取person中的数据字段  
  154.         //conf.setJar("tt.jar");  
  155.         // conf.setNumReduceTasks(1);  
  156.         //注意这行代码放在最前面，进行初始化，否则会报  
  157.          /**建立数据库连接**/  
  158.         DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test?characterEncoding=utf-8", "root", "qin");  
  159.         String fileds[]=new String[]{
    "word","count"};  
  160.           
  161.         Job job=new Job(conf, "writeDB");  
  162.           
  163.         System.out.println("运行模式：  "+conf.get("mapred.job.tracker"));  
  164.    
  165.         /**设置输出表的的信息  第一个参数是job任务，第二个参数是表名，第三个参数字段项**/  
  166.         DBOutputFormat.setOutput(job, "wordresult", fileds);  
  167.           
  168.         /**设置DB的输入路径**/  
  169.         job.setInputFormatClass(TextInputFormat.class);  
  170.          /**设置DB的输出路径**/  
  171.         job.setOutputFormatClass(DBOutputFormat.class);  
  172.           
  173.           
  174.         /***设置Reduce的个数为1，可以得到全局统计的数字 
  175.          * 但，需要注意，在分布式环境下，最好不要设置为1，Reduce的个数 
  176.          * 正是Hadoop并发能力的体现 
  177.          *  
  178.          * **/  
  179.     //  job.setNumReduceTasks(1);  
  180.           
  181.           
  182.         /**设置输出K路径**/  
  183.         job.setOutputKeyClass(Text.class);  
  184.         /**设置输出V路径**/  
  185.         job.setOutputValueClass(IntWritable.class);  
  186.           
  187.           
  188.         /**设置Map类**/  
  189.         job.setMapperClass(WMap.class);  
  190.         /**设置Combiner类**/  
  191.         job.setCombinerClass(WCombine.class);  
  192.         /**设置Reduce类**/  
  193.         job.setReducerClass(WReduce.class);  
  194.           
  195.           
  196.           
  197.         /**设置输入路径*/  
  198.         FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.75.130:9000/root/input"));  
  199.        
  200.   
  201.       
  202.         System.exit(job.waitForCompletion(true) ? 0 : 1);    
  203.           
  204.           
  205.     }  
  206.       
  207.   
  208. }  
package com.qin.operadb;import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapred.JobConf;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.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WriteMapDB {				private static class WMap extends Mapper
    
     {						/***		 * Mapper的参数类型介绍		 * K,V,K,V分别依次代表		 * Map作业输入类型的K，输入类型的V		 * 后面两个是输出类型的K，输出类型的V		 * 后面的两个与 context.write(word, one);		 * 的两个参数是对应的		 * **/		private Text word=new Text();		private IntWritable one=new IntWritable(1);				@Override		protected void map(LongWritable key, Text value,Context context)				throws IOException, InterruptedException {	 			 String line=value.toString();			//处理记事本UTF-8的BOM问题	            if (line.getBytes().length > 0) {	                if ((int) line.charAt(0) == 65279) {	                    line = line.substring(1);	                }	            }			 			 StringTokenizer st=new StringTokenizer(line);			 while(st.hasMoreTokens()){				 word.set(st.nextToken());//设置单词				 context.write(word, one);			 }									 		}							}			/***	 * 由于在reduce中，需要向数据库里写入	 * 数据，所以跟combine，不能共用	 * 	 * 	 * 	 * ***/	private static class WCombine extends Reducer
     
      {				@Override		protected void reduce(Text text, Iterable
      
        value,Context context)				throws IOException, InterruptedException {									int sum=0;						for(IntWritable iw:value){				sum+=iw.get();			}			context.write(text, new IntWritable(sum));						 		}									}						/**	 * 	 * Reduce类	 * 	 * **/	private static class WReduce extends Reducer
       
        {						@Override		protected void reduce(Text key, Iterable
        
          values,Context context)				throws IOException, InterruptedException {			 			  int sum=0;						  for(IntWritable s:values){				  sum+=s.get();			  }			  			  			  WordRecoder wr=new WordRecoder();			  wr.word=key.toString();			  wr.count=sum;			  			  //写出到数据库里			  context.write(wr, new Text());														}							}								public static void main(String[] args)throws Exception {				 JobConf conf=new JobConf(WriteMapDB.class);		//Configuration conf=new Configuration();	   // conf.set("mapred.job.tracker","192.168.75.130:9001");		//读取person中的数据字段	  	//conf.setJar("tt.jar");		// conf.setNumReduceTasks(1);		//注意这行代码放在最前面，进行初始化，否则会报		 /**建立数据库连接**/		DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test?characterEncoding=utf-8", "root", "qin");		String fileds[]=new String[]{"word","count"};				Job job=new Job(conf, "writeDB");				System.out.println("运行模式：  "+conf.get("mapred.job.tracker")); 		/**设置输出表的的信息  第一个参数是job任务，第二个参数是表名，第三个参数字段项**/		DBOutputFormat.setOutput(job, "wordresult", fileds);				/**设置DB的输入路径**/		job.setInputFormatClass(TextInputFormat.class);		 /**设置DB的输出路径**/		job.setOutputFormatClass(DBOutputFormat.class);						/***设置Reduce的个数为1，可以得到全局统计的数字		 * 但，需要注意，在分布式环境下，最好不要设置为1，Reduce的个数		 * 正是Hadoop并发能力的体现		 * 		 * **/	//	job.setNumReduceTasks(1);						/**设置输出K路径**/		job.setOutputKeyClass(Text.class);		/**设置输出V路径**/		job.setOutputValueClass(IntWritable.class);						/**设置Map类**/		job.setMapperClass(WMap.class);		/**设置Combiner类**/		job.setCombinerClass(WCombine.class);		/**设置Reduce类**/		job.setReducerClass(WReduce.class);								/**设置输入路径*/		FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.75.130:9000/root/input"));	 			System.exit(job.waitForCompletion(true) ? 0 : 1);  					}	}

运行状态如下所示：

Java代码  
  1. 运行模式：  192.168.75.130:9001  
  2. 14/03/26 20:26:59 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.  
  3. 14/03/26 20:27:01 INFO input.FileInputFormat: Total input paths to process : 2  
  4. 14/03/26 20:27:01 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable  
  5. 14/03/26 20:27:01 WARN snappy.LoadSnappy: Snappy native library not loaded  
  6. 14/03/26 20:27:01 INFO mapred.JobClient: Running job: job_201403262328_0006  
  7. 14/03/26 20:27:02 INFO mapred.JobClient:  map 0% reduce 0%  
  8. 14/03/26 20:27:10 INFO mapred.JobClient:  map 50% reduce 0%  
  9. 14/03/26 20:27:11 INFO mapred.JobClient:  map 100% reduce 0%  
  10. 14/03/26 20:27:18 INFO mapred.JobClient:  map 100% reduce 33%  
  11. 14/03/26 20:27:19 INFO mapred.JobClient:  map 100% reduce 100%  
  12. 14/03/26 20:27:20 INFO mapred.JobClient: Job complete: job_201403262328_0006  
  13. 14/03/26 20:27:20 INFO mapred.JobClient: Counters: 28  
  14. 14/03/26 20:27:20 INFO mapred.JobClient:   Job Counters   
  15. 14/03/26 20:27:20 INFO mapred.JobClient:     Launched reduce tasks=1  
  16. 14/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_MAPS=10345  
  17. 14/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all reduces waiting after reserving slots (ms)=0  
  18. 14/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all maps waiting after reserving slots (ms)=0  
  19. 14/03/26 20:27:20 INFO mapred.JobClient:     Launched map tasks=2  
  20. 14/03/26 20:27:20 INFO mapred.JobClient:     Data-local map tasks=2  
  21. 14/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_REDUCES=8911  
  22. 14/03/26 20:27:20 INFO mapred.JobClient:   File Output Format Counters   
  23. 14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Written=0  
  24. 14/03/26 20:27:20 INFO mapred.JobClient:   FileSystemCounters  
  25. 14/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_READ=158  
  26. 14/03/26 20:27:20 INFO mapred.JobClient:     HDFS_BYTES_READ=325  
  27. 14/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=182065  
  28. 14/03/26 20:27:20 INFO mapred.JobClient:   File Input Format Counters   
  29. 14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Read=107  
  30. 14/03/26 20:27:20 INFO mapred.JobClient:   Map-Reduce Framework  
  31. 14/03/26 20:27:20 INFO mapred.JobClient:     Map output materialized bytes=164  
  32. 14/03/26 20:27:20 INFO mapred.JobClient:     Map input records=6  
  33. 14/03/26 20:27:20 INFO mapred.JobClient:     Reduce shuffle bytes=164  
  34. 14/03/26 20:27:20 INFO mapred.JobClient:     Spilled Records=24  
  35. 14/03/26 20:27:20 INFO mapred.JobClient:     Map output bytes=185  
  36. 14/03/26 20:27:20 INFO mapred.JobClient:     Total committed heap usage (bytes)=336338944  
  37. 14/03/26 20:27:20 INFO mapred.JobClient:     CPU time spent (ms)=2850  
  38. 14/03/26 20:27:20 INFO mapred.JobClient:     Combine input records=20  
  39. 14/03/26 20:27:20 INFO mapred.JobClient:     SPLIT_RAW_BYTES=218  
  40. 14/03/26 20:27:20 INFO mapred.JobClient:     Reduce input records=12  
  41. 14/03/26 20:27:20 INFO mapred.JobClient:     Reduce input groups=8  
  42. 14/03/26 20:27:20 INFO mapred.JobClient:     Combine output records=12  
  43. 14/03/26 20:27:20 INFO mapred.JobClient:     Physical memory (bytes) snapshot=464982016  
  44. 14/03/26 20:27:20 INFO mapred.JobClient:     Reduce output records=8  
  45. 14/03/26 20:27:20 INFO mapred.JobClient:     Virtual memory (bytes) snapshot=2182836224  
  46. 14/03/26 20:27:20 INFO mapred.JobClient:     Map output records=20  
运行模式：  192.168.75.130:900114/03/26 20:26:59 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.14/03/26 20:27:01 INFO input.FileInputFormat: Total input paths to process : 214/03/26 20:27:01 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable14/03/26 20:27:01 WARN snappy.LoadSnappy: Snappy native library not loaded14/03/26 20:27:01 INFO mapred.JobClient: Running job: job_201403262328_000614/03/26 20:27:02 INFO mapred.JobClient:  map 0% reduce 0%14/03/26 20:27:10 INFO mapred.JobClient:  map 50% reduce 0%14/03/26 20:27:11 INFO mapred.JobClient:  map 100% reduce 0%14/03/26 20:27:18 INFO mapred.JobClient:  map 100% reduce 33%14/03/26 20:27:19 INFO mapred.JobClient:  map 100% reduce 100%14/03/26 20:27:20 INFO mapred.JobClient: Job complete: job_201403262328_000614/03/26 20:27:20 INFO mapred.JobClient: Counters: 2814/03/26 20:27:20 INFO mapred.JobClient:   Job Counters 14/03/26 20:27:20 INFO mapred.JobClient:     Launched reduce tasks=114/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_MAPS=1034514/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all reduces waiting after reserving slots (ms)=014/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all maps waiting after reserving slots (ms)=014/03/26 20:27:20 INFO mapred.JobClient:     Launched map tasks=214/03/26 20:27:20 INFO mapred.JobClient:     Data-local map tasks=214/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_REDUCES=891114/03/26 20:27:20 INFO mapred.JobClient:   File Output Format Counters 14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Written=014/03/26 20:27:20 INFO mapred.JobClient:   FileSystemCounters14/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_READ=15814/03/26 20:27:20 INFO mapred.JobClient:     HDFS_BYTES_READ=32514/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=18206514/03/26 20:27:20 INFO mapred.JobClient:   File Input Format Counters 14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Read=10714/03/26 20:27:20 INFO mapred.JobClient:   Map-Reduce Framework14/03/26 20:27:20 INFO mapred.JobClient:     Map output materialized bytes=16414/03/26 20:27:20 INFO mapred.JobClient:     Map input records=614/03/26 20:27:20 INFO mapred.JobClient:     Reduce shuffle bytes=16414/03/26 20:27:20 INFO mapred.JobClient:     Spilled Records=2414/03/26 20:27:20 INFO mapred.JobClient:     Map output bytes=18514/03/26 20:27:20 INFO mapred.JobClient:     Total committed heap usage (bytes)=33633894414/03/26 20:27:20 INFO mapred.JobClient:     CPU time spent (ms)=285014/03/26 20:27:20 INFO mapred.JobClient:     Combine input records=2014/03/26 20:27:20 INFO mapred.JobClient:     SPLIT_RAW_BYTES=21814/03/26 20:27:20 INFO mapred.JobClient:     Reduce input records=1214/03/26 20:27:20 INFO mapred.JobClient:     Reduce input groups=814/03/26 20:27:20 INFO mapred.JobClient:     Combine output records=1214/03/26 20:27:20 INFO mapred.JobClient:     Physical memory (bytes) snapshot=46498201614/03/26 20:27:20 INFO mapred.JobClient:     Reduce output records=814/03/26 20:27:20 INFO mapred.JobClient:     Virtual memory (bytes) snapshot=218283622414/03/26 20:27:20 INFO mapred.JobClient:     Map output records=20

最后，我们就可以去数据库里，查看统计的信息了，截图如下：
至此，我们就完成了使用MR来读写数据库了，注意测试前，先确保自己的hadoop集群，可以正常工作。

转载地址：http://bfjoi.baihongyu.com/

你可能感兴趣的文章
Optional类包含的方法
查看>>
如何使用MR来读取数据库的数据，并写入HDFS上
查看>>
mapred-site.xml里面配置运行日志的输出目录
查看>>
DistributedCache是Hadoop的一个分布式文件缓存类
查看>>
FileSplit:文件的子集--文件分割体
查看>>
使用Hadoop的MapReduce来完成大表join
查看>>
常用的算法
查看>>
Mina框架
查看>>
Spring MVC 和 Servlet 一样，都不是线程安全的
查看>>
Java线程：线程的同步与锁
查看>>
Mac、Windows可以互相远程
查看>>
oracle提示 ORA-12154: TNS: 无法解析指定的连接标识符
查看>>
oracle 插入数据时提示没有足够的值
查看>>
Oracle Net Manager的使用及配置
查看>>
镜像文件
查看>>
苹果笔记本桌面下面的工具栏没了怎么调出来
查看>>
CSS原理与CSS经验分享
查看>>
oracle中int与number的区别
查看>>
php不用jsonp也能跨域
查看>>
solr作为一种开源的搜索服务器
查看>>
喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！-- 愿君每日到此一游！
当前时间: 2024-09-24 16:34:38   当前IP: 18.223.213.238   联系邮箱:javaeecc@qq.com     Copyright © 2020 - 2022 baihongyu.com 京ICP备2021015314号-2
强烈建议你试试无所不能的CHAT-GPT，快点击我
强烈建议你试试无所不能的CHAT-GPT，快点击我