hadoop 从mysql中读取数据写到hdfs_mysql 数据转换成hadoop

1. 定义类

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.Writable;
import org.apache.hadoop.mapred.lib.db.DBWritable;

/**
 * @author DELL_pc
 *  @date 2017年6月27日
 */
public class StudentRecord implements Writable,DBWritable{
     int id;
     String name;

    /**
     * @return the id
     */
    public int getId() {
        return id;
    }


    /**
     * @param id the id to set
     */
    public void setId(int id) {
        this.id = id;
    }


    /**
     * @return the name
     */
    public String getName() {
        return name;
    }


    /**
     * @param name the name to set
     */
    public void setName(String name) {
        this.name = name;
    }


    @Override
    public void write(PreparedStatement statement) throws SQLException {
        // TODO Auto-generated method stub
        statement.setInt(1, this.id);
        statement.setString(2, this.name);
    }


    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        // TODO Auto-generated method stub
        this.id=resultSet.getInt(1);
        this.name=resultSet.getString(2);
    }


    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        out.writeInt(this.id);
        out.writeUTF(this.name);
    }


    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        this.id=in.readInt();
        this.name=in.readUTF();
    }



    @Override
    public String toString() {
        return "StudentRecord [id=" + id + ", name=" + name + "]";
    }


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

2.mr程序

/**
 * @author DELL_pc
 *  @date 2017年6月27日
 * 
 */
package com.beifeng.test;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
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.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.db.DBConfiguration;
import org.apache.hadoop.mapred.lib.db.DBInputFormat;

/**
 * @author DELL_pc
 *  @date 2017年6月27日
 */
public class DbaMysql {
     public static class DBAccessMapper extends MapReduceBase    implements Mapper<LongWritable,StudentRecord , IntWritable, Text>
     {


        @Override
        public void map(LongWritable key, StudentRecord value, OutputCollector<IntWritable, Text> output,
                Reporter reporter) throws IOException {
            // TODO Auto-generated method stub
            output.collect(new IntWritable(value.id), new Text(value.toString()));

        }


     }
      public static class DBAccessReduce extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text>
      {


        @Override
        public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable, Text> output,
                Reporter reporter) throws IOException {
              while (values.hasNext()) {
                  output.collect(key, values.next());
                }
        }

      }
     public static void main(String[] args) {
        Configuration configuration=new Configuration();
        JobConf jobConf=new JobConf(configuration);

        jobConf.setOutputKeyClass(IntWritable.class);
        jobConf.setOutputValueClass(Text.class);
        jobConf.setInputFormat(DBInputFormat.class);
        String[] fields={"id","name"};
        DBInputFormat.setInput(jobConf, StudentRecord.class, "user", "length(name)>2", "",fields );
        DBConfiguration.configureDB(jobConf, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/test","root","123456");
        jobConf.setMapperClass(DBAccessMapper.class);
        jobConf.setReducerClass(DBAccessReduce.class);
        FileOutputFormat.setOutputPath(jobConf,new Path("/data/out"));
        try {
            JobClient.runJob(jobConf);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80