将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 = "....";//【这个保密哈】
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
)
四.测试运行
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()
六.总结
【这一部分,我晚点再总结,这个内容我悟了好久,大概一个星期之久吧,才大概弄懂为啥如此编写代码。哈哈哈。不过最后还是开心。等我分享哦!】