MapReduce直接连接MySQL获取数据(2)

@Override
                public void write(DataOutput out) throws IOException {
                        // TODO Auto-generated method stub
                        Text.writeString(out, this.tbl_name);
                        Text.writeString(out, this.tbl_type);
                }

@Override
                public void readFields(DataInput in) throws IOException {
                        // TODO Auto-generated method stub
                        this.tbl_name = Text.readString(in);
                        this.tbl_type = Text.readString(in);
                }

public String toString() {
                        return new String(this.tbl_name + " " + this.tbl_type);
                }

}

public static class ConnMysqlMapper extends Mapper<LongWritable,TblsRecord,Text,Text> {
                public void map(LongWritable key,TblsRecord values,Context context)
                                throws IOException,InterruptedException {
                        context.write(new Text(values.tbl_name), new Text(values.tbl_type));
                }
        }
       
        public static class ConnMysqlReducer extends Reducer<Text,Text,Text,Text> {
                public void reduce(Text key,Iterable<Text> values,Context context)
                                throws IOException,InterruptedException {
                        for(Iterator<Text> itr = values.iterator();itr.hasNext();) {
                                context.write(key, itr.next());
                        }
                }
        }
       
        public static void main(String[] args) throws Exception {
                Path output = new Path("/user/linuxidc/output/");
               
                FileSystem fs = FileSystem.get(URI.create(output.toString()), conf);
                if (fs.exists(output)) {
                        fs.delete(output);
                }
               
                //mysql的jdbc驱动
                DistributedCache.addFileToClassPath(new Path( 
                          "hdfs://hd022-test.nh.sdo.com/user/liuxiaowen/mysql-connector-java-5.1.13-bin.jar"), conf); 
               
                DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", 
                          "jdbc:mysql://10.133.103.22:3306/hive", "hive", "hive"); 
               
                Job job = new Job(conf,"test mysql connection");
                job.setJarByClass(ConnMysql.class);
               
                job.setMapperClass(ConnMysqlMapper.class);
                job.setReducerClass(ConnMysqlReducer.class);
               
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(Text.class);
               
                job.setInputFormatClass(DBInputFormat.class);
                FileOutputFormat.setOutputPath(job, output);
               
                //列名
                String[] fields = { "TBL_NAME", "TBL_TYPE" };
                //六个参数分别为：
                //1.Job;2.Class<? extends DBWritable>
                //3.表名;4.where条件
                //5.order by语句;6.列名
                DBInputFormat.setInput(job, TblsRecord.class,
                    "linuxidc_tbls", "TBL_NAME like 'linuxidc%'", "TBL_NAME", fields); 
               
                System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
       
}

运行结果：