Hadoop中的DBInputFormat

1. create table student(id INTEGER NOT NULL PRIMARY KEY,name VARCHAR(32) NOT NULL);  

(3)：插入数据

1. insert into student values(1,"lavimer");  

(4)编写MapReduce程序，我这里使用的版本是hadoop1.2.1，相关知识点都写在注释中了，如下：

1. /** 
2.  * 使用DBInputFormat和DBOutputFormat 
3.  * 要把数据库的jdbc驱动放到各个TaskTracker节点的lib目录下 
4.  * 重启集群 
5.  * @author 廖钟民 
6.  * time : 2015年1月15日下午12:50:55 
7.  * @version 
8.  */  
9. public class MyDBInputFormat {  
10.   
11.     //定义输出路径  
12.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
13.       
14.     public static void main(String[] args) {  
15.           
16.         try {  
17.             //创建配置信息  
18.             Configuration conf = new Configuration();  
19.               
20.             /*//对Map端的输出进行压缩 
21.             conf.setBoolean("mapred.compress.map.output", true); 
22.             //设置map端输出使用的压缩类 
23.             conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class); 
24.             //对reduce端输出进行压缩 
25.             conf.setBoolean("mapred.output.compress", true); 
26.             //设置reduce端输出使用的压缩类 
27.             conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);*/  
28.               
29.             // 添加配置文件(我们可以在编程的时候动态配置信息，而不需要手动去改变集群)  
30.             /* 
31.             * conf.addResource("classpath://hadoop/core-site.xml");  
32.             * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
33.             * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
34.             */  
35.               
36.             //通过conf创建数据库配置信息  
37.             DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://liaozhongmin:3306/myDB","root","134045");  
38.               
39.             //创建文件系统  
40.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
41.               
42.             //如果输出目录存在就删除  
43.             if (fileSystem.exists(new Path(OUT_PATH))){  
44.                 fileSystem.delete(new Path(OUT_PATH),true);  
45.             }  
46.               
47.             //创建任务  
48.             Job job = new Job(conf,MyDBInputFormat.class.getName());  
49.               
50.             //1.1 设置输入数据格式化的类和设置数据来源  
51.             job.setInputFormatClass(DBInputFormat.class);  
52.             DBInputFormat.setInput(job, Student.class, "student", null, null, new String[]{"id","name"});  
53.               
54.             //1.2 设置自定义的Mapper类和Mapper输出的key和value的类型  
55.             job.setMapperClass(MyDBInputFormatMapper.class);  
56.             job.setMapOutputKeyClass(Text.class);  
57.             job.setMapOutputValueClass(Text.class);  
58.               
59.             //1.3 设置分区和reduce数量(reduce的数量和分区的数量对应，因为分区只有一个，所以reduce的个数也设置为一个)  
60.             job.setPartitionerClass(HashPartitioner.class);  
61.             job.setNumReduceTasks(1);  
62.               
63.             //1.4 排序、分组  
64.             //1.5 归约  
65.             //2.1 Shuffle把数据从Map端拷贝到Reduce端  
66.               
67.             //2.2 指定Reducer类和输出key和value的类型  
68.             job.setReducerClass(MyDBInputFormatReducer.class);  
69.             job.setOutputKeyClass(Text.class);  
70.             job.setOutputValueClass(Text.class);  
71.               
72.             //2.3 指定输出的路径和设置输出的格式化类  
73.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
74.             job.setOutputFormatClass(TextOutputFormat.class);  
75.               
76.             //提交作业 然后关闭虚拟机正常退出  
77.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
78.               
79.         } catch (Exception e) {  
80.             e.printStackTrace();  
81.         }  
82.     }  
83.       
84.     /** 
85.      * 自定义Mapper类 
86.      * @author 廖钟民 
87.      * time : 2015年1月15日下午1:22:57 
88.      * @version 
89.      */  
90.     public static class MyDBInputFormatMapper extends Mapper<LongWritable, Student, Text, Text>{  
91.         //创建map输出时的key类型  
92.          private Text mapOutKey = new Text();  
93.          //创建map输出时的value类型  
94.          private Text mapOutValue = new Text();  
95.            
96.         @Override  
97.         protected void map(LongWritable key, Student value, Mapper<LongWritable, Student, Text, Text>.Context context) throws IOException, InterruptedException {  
98.               
99.             //创建输出的key:把id当做key  
100.             mapOutKey.set(String.valueOf(value.getId()));  
101.             //创建输出的value：把name当做value  
102.             mapOutValue.set(value.getName());  
103.               
104.             //通过context写出去  
105.             context.write(mapOutKey, mapOutValue);  
106.         }  
107.     }  
108.       
109.     /** 
110.      * 自定义Reducer类 
111.      * @author 廖钟民 
112.      * time : 2015年1月15日下午1:23:28 
113.      * @version 
114.      */  
115.     public static class MyDBInputFormatReducer extends Reducer<Text, Text, Text, Text>{  
116.           
117.         @Override  
118.         protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {  
119.               
120.             //遍历把结果写到HDFS中  
121.             for (Text t : values){  
122.                   
123.                 context.write(key, t);  
124.             }  
125.         }  
126.     }  
127. }  
128.   
129. /** 
130.  * 自定义实体类 用于对应数据库表中的字段 
131.  * @author 廖钟民 
132.  * time : 2015年1月15日下午12:52:58 
133.  * @version 
134.  */  
135. class Student implements Writable,DBWritable{  
136.   
137.     //学生id字段  
138.     private Integer id;  
139.     //学生姓名  
140.     private String name;  
141.       
142.       
143.     //无参构造方法  
144.     public Student() {  
145.     }  
146.   
147.     //有参构造方法  
148.     public Student(Integer id, String name) {  
149.         this.id = id;  
150.         this.name = name;  
151.     }  
152.   
153.   
154.     public Integer getId() {  
155.         return id;  
156.     }  
157.   
158.     public void setId(Integer id) {  
159.         this.id = id;  
160.     }  
161.   
162.     public String getName() {  
163.         return name;  
164.     }  
165.   
166.     public void setName(String name) {  
167.         this.name = name;  
168.     }  
169.   
170.     //实现DBWritable接口要实现的方法  
171.     public void readFields(ResultSet resultSet) throws SQLException {  
172.         this.id = resultSet.getInt(1);  
173.         this.name = resultSet.getString(2);  
174.     }  
175.       
176.     //实现DBWritable接口要实现的方法  
177.     public void write(PreparedStatement preparedStatement) throws SQLException {  
178.         preparedStatement.setInt(1, this.id);  
179.         preparedStatement.setString(2, this.name);  
180.     }  
181.   
182.     //实现Writable接口要实现的方法  
183.     public void readFields(DataInput dataInput) throws IOException {  
184.         this.id = dataInput.readInt();  
185.         this.name = Text.readString(dataInput);  
186.     }  
187.       
188.     //实现Writable接口要实现的方法  
189.     public void write(DataOutput dataOutput) throws IOException {  
190.         dataOutput.writeInt(this.id);  
191.         Text.writeString(dataOutput, this.name);  
192.     }  
193.   
194.       
195.     @Override  
196.     public int hashCode() {  
197.         final int prime = 31;  
198.         int result = 1;  
199.         result = prime * result + ((id == null) ? 0 : id.hashCode());  
200.         result = prime * result + ((name == null) ? 0 : name.hashCode());  
201.         return result;  
202.     }  
203.   
204.     @Override  
205.     public boolean equals(Object obj) {  
206.         if (this == obj)  
207.             return true;  
208.         if (obj == null)  
209.             return false;  
210.         if (getClass() != obj.getClass())  
211.             return false;  
212.         Student other = (Student) obj;  
213.         if (id == null) {  
214.             if (other.id != null)  
215.                 return false;  
216.         } else if (!id.equals(other.id))  
217.             return false;  
218.         if (name == null) {  
219.             if (other.name != null)  
220.                 return false;  
221.         } else if (!name.equals(other.name))  
222.             return false;  
223.         return true;  
224.     }  
225.   
226.     @Override  
227.     public String toString() {  
228.         return "Student [id=" + id + ", name=" + name + "]";  
229.     }  
230.       
231.       
232. }  

程序运行的结果是数据库中的数据成功导入到HDFS中，如下：

注：程序运行时，会碰到一个常见的数据库远程连接错误，大致如下：

1. Access denied for user ‘root‘@‘%‘ to database ‘xxxx’  

原因：创建完数据库后，需要进行授权(在本地访问一般不会出现这个问题)

解决方法就是进行授权：

1. grant all on xxxx.* to ‘root‘@‘%‘ identified by ‘password‘ with grant option;  
2.   
3. xxxx代表创建的数据库;  
4. password为用户密码，在此为root的密码  

另外一个常见的错误就是MYSQL驱动没有导入到hadoop/lib目录下，解决方案有两种，传统的方式我就不多说了，这里说另外一种方式：

(1)：把包上传到集群上

1. hadoop fs -put mysql-connector-java-5.1.0- bin.jar /lib  

(2)：在MR程序提交job前，添加语句：

1. DistributedCache.addFileToClassPath(new Path("/lib/mysql- connector-java- 5.1.0-bin.jar"), conf);  

Hadoop中的DBInputFormat

标签：