166
通過虛擬機的hdfs讀取本地mysql上的文件
讀取時會出現一些問題
比如 虛擬機上無法ping 本地ip
還有就是需要在類中加
job.addFileToClassPath(new Path("/mysql-connector-java-5.1.43-bin.jar"));
關於這個jar包需要放在hdfs中
在解壓安裝的hadoop中的share/hadoop/common/lib/下也要加入
且每個子節點都得配置一下
避免出錯
public class ReadFormDB {
public static class ReadFromDBMap extends Mapper<LongWritable, WordCountDBWritable, Text, NullWritable> {
private final NullWritable outValue = NullWritable.get();
private Text outKey = new Text();
@Override
protected void map(LongWritable key, WordCountDBWritable value,
Mapper<LongWritable, WordCountDBWritable, Text, NullWritable>.Context context)
throws IOException, InterruptedException {
outKey.set(value.toString());
context.write(outKey, outValue);
}
}
//GRANT ALL PRIVILEGES ON *.* TO 'root'@'%'IDENTIFIED BY '123456' WITH GRANT OPTION;
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration configuration =new Configuration();
// DBConfiguration.configureDB(configuration, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost/test","root","123456");
DBConfiguration.configureDB(configuration, "com.mysql.jdbc.Driver",
"jdbc:mysql://192.168.111.1:3306/test?useSSL=false&autoReconnect=true&failOverReadOnly=false","root","123456");
Job job=Job.getInstance(configuration);
job.setJarByClass(ReadFormDB.class);
job.setJobName("讀取從mysql中");
job.setMapperClass(ReadFromDBMap.class);
job.setNumReduceTasks(0);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.addFileToClassPath(new Path("/mysql-connector-java-5.1.43-bin.jar"));
DBInputFormat.setInput(job, WordCountDBWritable.class, "word_count", "wc_count","wc_count asc","*");
Path outputPath =new Path("/ReadFormD");
outputPath.getFileSystem(configuration).delete(outputPath, true);
FileOutputFormat.setOutputPath(job, outputPath);
System.exit(job.waitForCompletion(true)?0:1);
}
}
把wordcount 的結果寫入到mysql中
public class WriteToDB {
// 對應表 word_count create table(wc_word varchar(255) )
public static class WordCountDBWritable implements DBWritable, Writable {
private String word;
private int count;
public String getWord() {
return word;
}
public void setWord(String word) {
this.word = word;
}
public int getCount() {
return count;
}
public void setCount(int count) {
this.count = count;
}
@Override
public String toString() {
return word + "," + count;
}
// 把數據寫到數據庫中
// insert into word_count(wc_word,wc_count)value(?,?)
public void write(PreparedStatement statement) throws SQLException {
statement.setString(1, this.word);
statement.setInt(2, this.count);
}
// 從數據庫中讀取數據
public void readFields(ResultSet resultSet) throws SQLException {
this.word = resultSet.getString("wc_word");
this.count = resultSet.getInt("wc_count");
}
public void write(DataOutput out) throws IOException {
out.writeUTF(this.word);
out.writeInt(this.count);
}
public void readFields(DataInput in) throws IOException {
this.word = in.readUTF();
this.count = in.readInt();
}
}
public static class WriteToDBMap extends Mapper<LongWritable, Text, Text, IntWritable> {
private final IntWritable ONE = new IntWritable(1);
private Text oKey = new Text();
private String[] infos;
上麵是讀文件的方式
下麵是關於如何寫入到本地mysql中
最後更新：2017-10-19 21:03:43