MapReduce操作Mysql的方法是什么

這篇文章主要講解了“MapReduce操作Mysql的方法是什么”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“MapReduce操作Mysql的方法是什么”吧！

hadoop技術推出一首曾遭到關系數據庫研究者的挑釁和批評，認為MapReduce不具有關系數據庫中的結構化數據存儲和處理能力。為此，hadoop社區和研究人員做了多的努力，在hadoop0.19版支持MapReduce訪問關系數據庫，如：mysql，MySQL、PostgreSQL、Oracle 等幾個數據庫系統。

1. 從Mysql讀出數據

Hadoop訪問關系數據庫主要通過一下接口實現的：DBInputFormat類，包所在位置：org.apache.hadoop.mapred.lib.db 中。DBInputFormat 在 Hadoop 應用程序中通過數據庫供應商提供的 JDBC接口來與數據庫進行交互,并且可以使用標準的 SQL 來讀取數據庫中的記錄。學習DBInputFormat首先必須知道二個條件。

1. 在使用 DBInputFormat 之前,必須將要使用的 JDBC 驅動拷貝到分布式系統各個節點的$HADOOP_HOME/lib/目錄下。

2. MapReduce訪問關系數據庫時，大量頻繁的從MapReduce程序中查詢和讀取數據，這大大的增加了數據庫的訪問負載，因此，DBInputFormat接口僅僅適合讀取小數據量的數據，而不適合處理數據倉庫。要處理數據倉庫的方法有：利用數據庫的Dump工具將大量待分析的數據輸出為文本，并上傳的Hdfs中進行處理

 DBInputFormat 類中包含以下三個內置類

1. protected class DBRecordReader implementsRecordReader<LongWritable, T>:用來從一張數據庫表中讀取一條條元組記錄。

2. 2.public static class NullDBWritable implements DBWritable,Writable:主要用來實現 DBWritable 接口。DBWritable接口要實現二個函數，第一是write，第二是readFileds，這二個函數都不難理解，一個是寫，一個是讀出所有字段。原型如下：
   

   public void write(PreparedStatement statement) throwsSQLException;public void readFields(ResultSet resultSet) throws SQLException;

3. protected static class DBInputSplit implements InputSplit:主要用來描述輸入元組集合的范圍,包括 start 和 end 兩個屬性,start 用來表示第一條記錄的索引號,end 表示最后一條記錄的索引號.

下面對怎樣使用 DBInputFormat 讀取數據庫記錄進行詳細的介紹,具體步驟如下:

1. DBConfiguration.configureDB (JobConf job, StringdriverClass, String dbUrl, String userName, String passwd)函數，配置JDBC 驅動,數據源,以及數據庫訪問的用戶名和密碼。MySQL 數據庫的 JDBC 的驅動為“com.mysql.jdbc.Driver”,數據源為“jdbc:mysql://localhost/testDB”,其中testDB為訪問的數據庫。useName一般為“root”，passwd是你數據庫的密碼。

2. DBInputFormat.setInput(JobConf job, Class<?extends DBWritable> inputClass, String tableName, String conditions,String orderBy, String... fieldNames)，這個方法的參數很容易看懂，inputClass實現DBWritable接口。，string tableName表名， conditions表示查詢的條件，orderby表示排序的條件，fieldNames是字段，這相當與把sql語句拆分的結果。當然也可以用sql語句進行重載。etInput(JobConf job, Class<?extends DBWritable> inputClass, String inputQuery, StringinputCountQuery)。

3. 編寫MapReduce函數,包括Mapper 類、Reducer 類、輸入輸出文件格式等,然后調用JobClient.runJob(conf)。

上面講了理論，下面舉個例子：假設 MySQL 數據庫中有數據庫student，假設數據庫中的字段有“id”，“name”，“gender","number"。

第一步要實現DBwrite和write數據接口。代碼如下：

        public class StudentRecord implements Writable, DBWritable{            int id;
            String name;
            String gender;
            String number;
            @Override        public void readFields(DataInput in) throws IOException {            // TODO Auto-generated method stub
            this.id = in.readInt();            this.gender = Text.readString(in);            this.name = in.readString();            this.number = in.readString();
        }
            @Override        public void write(DataOutput out) throws IOException {            // TODO Auto-generated method stub
            out.writeInt(this.id);
            Text.writeString(out,this.name);            out.writeInt(this.gender);            out.writeInt(this.number);
        }
            
            @Override        public void readFields(ResultSet result) throws SQLException {            // TODO Auto-generated method stub
            this.id = result.getInt(1);            this.name = result.getString(2);            this.gender = result.getString(3);            this.number = result.getString(4);
        }
            
            @Override        public void write(PreparedStatement stmt) throws SQLException{            // TODO Auto-generated method stub
            stmt.setInt(1, this.id);
            stmt.setString(2, this.name);
            stmt.setString(3, this.gender);
            stmt.setString(4, this.number);
        }
            @Override        public String toString() {            // TODO Auto-generated method stub
            return new String(this.name + " " + this.gender + " " +this.number);
        }

第二步，實現Map和Reduce類

    public class DBAccessMapper extends MapReduceBase implements
            Mapper<LongWritable, TeacherRecord, LongWritable, Text> {
        @Override        public void map(LongWritable key, TeacherRecord value,
                OutputCollector<LongWritable, Text> collector, Reporter reporter)
                throws IOException {            // TODO Auto-generated method stub
            new collector.collect(new LongWritable(value.id), new Text(value
                    .toString()));
        }
    }

第三步：主函數的實現，函數

public class DBAccessReader {    public static void main(String[] args) throws IOException {
        JobConf conf = new JobConf(DBAccessReader.class);
        conf.setOutputKeyClass(LongWritable.class);
        conf.setOutputValueClass(Text.class);
        conf.setInputFormat(DBInputFormat.class);
        FileOutputFormat.setOutputPath(conf, new Path("dboutput"));
        DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver",        "jdbc:mysql://localhost/school","root","123456");
        String [] fields = {"id", "name", "gender", "number"};
        DBInputFormat.setInput(conf, StudentRecord.class,"Student",null "id", fields);
        conf.setMapperClass(DBAccessMapper.class);
        conf.setReducerClass(IdentityReducer.class);
        JobClient.runJob(conf);
        }

}

2.寫數據

 往往對于數據處理的結果的數據量一般不會太大，可能適合hadoop直接寫入數據庫中。hadoop提供了相應的數據庫直接輸出的計算發結果。

1. DBOutFormat: 提供數據庫寫入接口。

2. DBRecordWriter:提供向數據庫中寫入的數據記錄的接口。

3. DBConfiguration:提供數據庫配置和創建鏈接的接口。

DBOutFormat提供一個靜態方法setOutput(job,String table,String ...filedNames);該方法的參數很容易看懂。假設要插入一個Student的數據，其代碼為

    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        JobConf conf = new JobConf();
        conf.setOutputFormat(DBOutputFormat.class);
        DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver",                "jdbc:mysql://localhost/school","root","123456");
        DBOutputFormat.setOutput(conf,"Student", 456, "liqizhou", "man", "20004154578");
        JobClient.runJob(conf);

感謝各位的閱讀，以上就是“MapReduce操作Mysql的方法是什么”的內容了，經過本文的學習后，相信大家對MapReduce操作Mysql的方法是什么這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！