Hadoop项目实战之将MapReduce的结果写入到Mysql

一.环境配置

1.本次实验的主要配置环境如下:

  • 物理机:windows 10
  • 虚拟机:VMware pro 12,用其分别创建了三个虚拟机,其ip地址分别为192.168.211.3
  • hadoop2.6.4
  • Server version: 5.7.21 MySQL Community Server (GPL)

二.具体需求

MapReduce 实现 WordCount 功能,并将输出写入到mysql中

三.代码实现

  • WordCountMapper
package MapReduce.three; 
 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 
 
import java.io.IOException; 
 
public class WordCountMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
   
     
    @Override 
    protected void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
   
     
        //获取每一个输入行 
        String line = value.toString(); 
        //get every separated word 
        String [] word = line.split(" "); 
        for(int i = 0;i< word.length;i++){
   
     
            context.write(new Text(word[i]),new IntWritable(1)); 
        } 
    } 
} 
  • WordCountReducer
package MapReduce.three; 
 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Reducer; 
 
import java.io.IOException; 
 
public class WordCountReducer extends Reducer<Text,IntWritable,ReceiveTable,NullWritable> {
   
     
    @Override 
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) 
            throws IOException, InterruptedException {
   
     
        int sum = 0; 
        for(IntWritable intW : values){
   
     
            sum += intW.get(); 
        } 
        ReceiveTable receiveTable = new ReceiveTable(key.toString(),sum); 
        context.write(receiveTable,null); 
    } 
} 
  • WordCountJob
package MapReduce.three; 
 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration; 
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
 
import java.io.IOException; 
 
public class WordCountJob {
   
     
    public static String driverClass = "com.mysql.jdbc.Driver"; 
    public static String dbUrl = "jdbc:mysql://192.168.211.3:3306/mydatabase"; 
    public static String userName = "root"; 
    public static String passwd = "xxxx"; 
    public static String inputFilePath = "hdfs://192.168.211.3:9000/input/word.txt"; 
    public static String tableName = "keyWord"; 
    public static String [] fields = {
   
    "word","total"}; 
 
    public static void main(String[] args) {
   
     
        Configuration conf = new Configuration(); 
        DBConfiguration.configureDB(conf,driverClass,dbUrl,userName,passwd); 
        try {
   
     
            Job job = Job.getInstance(conf); 
 
            job.setJarByClass(WordCountJob.class); 
            job.setMapOutputValueClass(IntWritable.class); 
            job.setMapOutputKeyClass(Text.class); 
 
            job.setMapperClass(WordCountMapper.class); 
            job.setReducerClass(WordCountReducer.class); 
 
            job.setJobName("MyWordCountDB"); 
 
            FileInputFormat.setInputPaths(job,new Path(inputFilePath)); 
            DBOutputFormat.setOutput(job,tableName,fields); 
 
            job.waitForCompletion(true); 
        } catch (IOException e) {
   
     
            e.printStackTrace(); 
        } catch (InterruptedException e) {
   
     
            e.printStackTrace(); 
        } catch (ClassNotFoundException e) {
   
     
            e.printStackTrace(); 
        } 
    } 
} 
  • ReceiveTable类【非常重要的实现】
package MapReduce.three; 
 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.io.Writable; 
import org.apache.hadoop.mapred.lib.db.DBWritable; 
 
import java.io.DataInput; 
import java.io.DataOutput; 
import java.io.IOException; 
import java.sql.PreparedStatement; 
import java.sql.ResultSet; 
import java.sql.SQLException; 
 
public class ReceiveTable implements Writable,DBWritable{
   
     
    //column1:keyword  column2:number 
    private String keyWord; 
    private int number; 
 
    public ReceiveTable(){
   
     
 
    } 
    public ReceiveTable(String keyWord,int number){
   
     
        this.keyWord = keyWord; 
        this.number = number; 
    } 
    /**Writable  only serializable and deseiralizable 
     * 
     * @param out 
     * @throws IOException 
     */ 
    @Override 
    public void write(DataOutput out) throws IOException {
   
     
        out.writeInt(this.number); 
        /*1.将this.keyWord以UTF8的编码方式写入到out中[Write a UTF8 encoded string to out] 
        2.其实这个效果和out.writeInt(this.number)是一样的,只不过是DataOutput类型没有writeString()这个方法, 
        所以借用了Text.writeString(...)这个方法 
         */ 
        Text.writeString(out, this.keyWord); 
    } 
 
    @Override 
    public void readFields(DataInput in) throws IOException {
   
     
        this.number = in.readInt(); 
        this.keyWord = in.readUTF(); 
    } 
 
 
    /**DBWritable 
     * write data to mysql 
     * @param statement 
     * @throws SQLException 
     */ 
    @Override 
    public void write(PreparedStatement statement) throws SQLException {
   
     
        statement.setString(1,this.keyWord); 
        statement.setInt(2,this.number); 
    } 
 
    /**DBWritable 
     * get data from resultset.And set in your fields 
     * @param resultSet 
     * @throws SQLException 
     */ 
    @Override 
    public void readFields(ResultSet resultSet) throws SQLException {
   
     
        this.keyWord = resultSet.getString(1); 
        this.number = resultSet.getInt(2); 
    } 
} 
  • 建表语句
CREATE TABLE `keyWord` ( 
  `word` varchar(10) NOT NULL, 
  `total` int(10) NOT NULL 
) 

四.测试运行

17:48:45.517 [Thread-4] WARN  o.a.hadoop.mapred.LocalJobRunner - job_local294354826_0001 
java.lang.Exception: java.io.IOException: Access denied for user 'root'@'%' to database 'mydatabase' 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462) ~[hadoop-mapreduce-client-common-2.6.5.jar:na] 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529) ~[hadoop-mapreduce-client-common-2.6.5.jar:na] 
java.io.IOException: Access denied for user 'root'@'%' to database 'mydatabase' 
	at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat.getRecordWriter(DBOutputFormat.java:185) ~[hadoop-mapreduce-client-core-2.6.4.jar:na] 
	at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.<init>(ReduceTask.java:540) ~[hadoop-mapreduce-client-core-2.6.4.jar:na] 
	at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:614) ~[hadoop-mapreduce-client-core-2.6.4.jar:na] 
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389) ~[hadoop-mapreduce-client-core-2.6.4.jar:na] 
	at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319) ~[hadoop-mapreduce-client-common-2.6.5.jar:na] 
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_77] 
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_77] 
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_77] 
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0_77] 
	at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_77] 

这个报错一看就应该知道是无权访问 mysql 数据库,所以将访问用户设置为%,并且给予相应的库,表权限即可。这里的相应操作如下:

mysql> create user 'root'@'%' identified by 'root'; 
Query OK, 0 rows affected (0.01 sec) 
 
mysql> grant all privileges on *.* to 'root'@'%'; 
Query OK, 0 rows affected (0.01 sec) 

详细mysql 操作可见我的mysql 相关博客。
接着再次执行,再遇到下面的报错:

11:05:38 WARN mapred.LocalJobRunner: job_local1320561409_0001 
java.lang.Exception: java.io.IOException: Data truncation: Data too long for column 'word' at row 1 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529) 
Caused by: java.io.IOException: Data truncation: Data too long for column 'word' at row 1 
	at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat$DBRecordWriter.close(DBOutputFormat.java:103) 
	at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.close(ReduceTask.java:550) 
	at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:629) 
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319) 
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
	at java.lang.Thread.run(Thread.java:745)	 

短眼一看,就知道这是因为数据库中的表keyWord中的某个字段设置的太短,导致出错。所以只需要将word.txt中的每个单词控制在10个字符长度之内就可以啦。
看一下最后的运行结果:

mysql> select *from keyWord; 
+-----------+-------+ 
| word      | total | 
+-----------+-------+ 
| LittleLaw |     1 | 
| hello     |     4 | 
| java      |     1 | 
| scala     |     1 | 
| spark     |     1 | 
+-----------+-------+ 
5 rows in set (0.00 sec) 

五.日志分析

具体分析一下这个执行日志,【待完善】

11:11:38 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this. 
11:11:38 WARN mapreduce.JobResourceUploader: No job jar file set.  User classes may not be found. See Job or Job#setJar(String). 
11:11:38 INFO mapred.LocalJobRunner: OutputCommitter set in config null 
11:11:38 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter 
11:11:38 WARN output.FileOutputCommitter: Output Path is null in setupJob() 
11:11:38 INFO mapred.LocalJobRunner: Waiting for map tasks 
11:11:38 INFO mapred.LocalJobRunner: Starting task: attempt_local1144070395_0001_m_000000_0 
11:11:38 INFO mapred.Task:  Using ResourceCalculatorProcessTree : org.apache.hadoop.yarn.util.WindowsBasedProcessTree@68cbb3fc 
11:11:38 INFO mapred.MapTask: Processing split: hdfs://192.168.211.3:9000/input/word.txt:0+55 
11:11:38 INFO mapred.MapTask: (EQUATOR) 0 kvi 26214396(104857584) 
11:11:38 INFO mapred.MapTask: mapreduce.task.io.sort.mb: 100 
11:11:38 INFO mapred.MapTask: soft limit at 83886080 
11:11:38 INFO mapred.MapTask: bufstart = 0; bufvoid = 104857600 
11:11:38 INFO mapred.MapTask: kvstart = 26214396; length = 6553600 
11:11:38 INFO mapred.MapTask: Map output collector class = org.apache.hadoop.mapred.MapTask$MapOutputBuffer 
11:11:38 INFO mapred.LocalJobRunner:  
11:11:38 INFO mapred.MapTask: Starting flush of map output 
11:11:38 INFO mapred.MapTask: Spilling map output 
11:11:38 INFO mapred.MapTask: bufstart = 0; bufend = 83; bufvoid = 104857600 
11:11:38 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 26214368(104857472); length = 29/6553600 
11:11:38 INFO mapred.MapTask: Finished spill 0 
11:11:38 INFO mapred.Task: Task:attempt_local1144070395_0001_m_000000_0 is done. And is in the process of committing 
11:11:38 INFO mapred.LocalJobRunner: map 
11:11:38 INFO mapred.Task: Task 'attempt_local1144070395_0001_m_000000_0' done. 
11:11:38 INFO mapred.LocalJobRunner: Finishing task: attempt_local1144070395_0001_m_000000_0 
11:11:38 INFO mapred.LocalJobRunner: map task executor complete. 
11:11:38 INFO mapred.LocalJobRunner: Waiting for reduce tasks 
11:11:38 INFO mapred.LocalJobRunner: Starting task: attempt_local1144070395_0001_r_000000_0 
11:11:38 INFO mapred.Task:  Using ResourceCalculatorProcessTree : org.apache.hadoop.yarn.util.WindowsBasedProcessTree@7e6c1d28 
11:11:38 INFO mapred.ReduceTask: Using ShuffleConsumerPlugin: org.apache.hadoop.mapreduce.task.reduce.Shuffle@2804e4c1 
11:11:38 INFO mapred.LocalJobRunner: 1 / 1 copied. 
11:11:38 INFO mapred.Merger: Merging 1 sorted segments 
11:11:38 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 89 bytes 
11:11:38 INFO mapred.Merger: Merging 1 sorted segments 
11:11:38 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 89 bytes 
11:11:38 INFO mapred.LocalJobRunner: 1 / 1 copied. 
11:11:39 INFO mapred.Task: Task:attempt_local1144070395_0001_r_000000_0 is done. And is in the process of committing 
11:11:39 INFO mapred.LocalJobRunner: reduce > reduce 
11:11:39 INFO mapred.Task: Task 'attempt_local1144070395_0001_r_000000_0' done. 
11:11:39 INFO mapred.LocalJobRunner: Finishing task: attempt_local1144070395_0001_r_000000_0 
11:11:39 INFO mapred.LocalJobRunner: reduce task executor complete. 
11:11:39 WARN output.FileOutputCommitter: Output Path is null in commitJob() 

六. 常见错误

  • 如果有多个字段想写进mysql的表中,就不能使用Text字段了,而是使用自己定义的一个类型。你需要这么思考。简单情况下,比如WordCount程序,因为结果是单词,【key值】,却是Text类型的对象。所以在reduce阶段的输出key类型是Text类型。所以如果想输出一个负责的对象就应该是一个其他对象所属的类型。【这个类必须实现DBWritable,Writable两个接口】
java.lang.Exception: java.lang.ClassCastException: org.apache.hadoop.io.Text cannot be cast to org.apache.hadoop.mapreduce.lib.db.DBWritable 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529) 
Caused by: java.lang.ClassCastException: org.apache.hadoop.io.Text cannot be cast to org.apache.hadoop.mapreduce.lib.db.DBWritable 
	at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat$DBRecordWriter.write(DBOutputFormat.java:66) 
	at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:558) 
	at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:89) 
	at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.write(WrappedReducer.java:105) 
	at mapReduce.FromHBToMys.HBMyReducer.reduce(HBMyReducer.java:22) 
	at mapReduce.FromHBToMys.HBMyReducer.reduce(HBMyReducer.java:9) 
	at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171) 
	at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:627) 
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319) 
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
	at java.lang.Thread.run(Thread.java:745) 
  • 容易出现数据库中的表并非是自己的目的表的错误,需要细心些。这时就会报无法识别的字段错。
java.lang.Exception: java.io.IOException: Unknown column 'duration' in 'field list' 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529) 
Caused by: java.io.IOException: Unknown column 'duration' in 'field list' 
	at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat$DBRecordWriter.close(DBOutputFormat.java:103) 
	at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.close(ReduceTask.java:550) 
	at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:629) 
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389) 
	at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319) 
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
	at java.lang.Thread.run(Thread.java:745) 

七. 总结

这一部分,我晚点再总结,这个内容我悟了好久,大概一个星期之久吧,才大概弄懂为啥如此编写代码。哈哈哈。不过最后还是开心。等我分享哦!

  • step 1
ReceiveTable receiveTable = new ReceiveTable(key.toString(),sum); 
context.write(receiveTable,null); 

注意这里的实现由简单的context.write(key,new IntWritable(count)); 变成了 context.write(receiveTable,null); 这里不仅仅是写一个简单的<Key,Value>了,写的是一个java 对象。后面的键就让其空着为null。

  • step 2
    当然如果做了step1,就需要预定义一个ReceiveTable类了。而ReceiveTable这个类需要序列化和DB操作,所以就实现了WritableDBWritable这两个接口。

评论关闭
IT虾米网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

HDFS基本命令行操作