Hadoop与关系数据库

Hadoop对关系数据库无非两种操作，即从关系数据库输入到HDFS和从HDFS输出到关系数据库。Hadoop中分别提供了DBInputFormat类和DBOutputFormat类，前者用于从关系数据库输入到HDFS,该类将关系数据库中的一条记录作为向Mapper输入的value值，后者用于将HDFS中的文件输出到关系数据库，该类将Reducer输出的key值存储到数据库。我们只要在主程序中设置job的输入输出格式为这两个类中的一种，就可以让Hadoop从关系数据库输入或者向关系数据库输出。
正如我上面提到的，我们在操作的过程中使用了“记录”这个对象，因此需要写一个类对应到关系数据库中我们要操作的那个表，这个类要实现DBWritable接口和Writable接口，具体参见HadoopAPI。
具体代码参见文档。

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.lib.db.*;
import java.sql.*;
import java.io.*;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.fs.Path;
public class SDBConnInput {
public static class CustomerRecord implements Writable,DBWritable{
String customerID;
String customerName;
String phoneNumber;
public void readFields(ResultSet resultSet)  throws SQLException{
customerID=resultSet.getString(1);
customerName=resultSet.getString(2);
phoneNumber=resultSet.getString(3);
}
public void write(PreparedStatement statement)  throws SQLException{
statement.setString(1, customerID);
statement.setString(2, customerName);
statement.setString(3,phoneNumber);
}
public void readFields(DataInput in) throws IOException{
customerID=in.readUTF();
customerName=in.readUTF();
phoneNumber=in.readUTF();
}
public void write(DataOutput out) throws IOException{
out.writeUTF(customerID);
out.writeUTF(customerName);
out.writeUTF(phoneNumber);
}
public void setCustomerID(String customerID){
this.customerID=customerID;
}
public void setCustomerName(String customerName){
this.customerName=customerName;
}
public void setPhoneNumber(String phoneNumber){
this.phoneNumber=phoneNumber;
}
public String toString(){
return this.customerID+","+this.customerName+","+this.phoneNumber;
}
}
public static class MapperClass extends MapReduceBase implements Mapper<LongWritable,CustomerRecord,LongWritable,Text>{
Text result= new Text();
public void map(LongWritable key, CustomerRecord value,OutputCollector<LongWritable, Text> collector, Reporter reporter) throws IOException{
result.set(value.toString());
collector.collect(key, result);
}
}
public static class ReducerClass extends MapReduceBase implements Reducer<LongWritable, Text,NullWritable,Text>{
public void reduce(LongWritable key, Iterator<Text> values, OutputCollector<NullWritable,Text> output, Reporter reporter) throws IOException{
String str="";
while(values.hasNext()){
str+=values.next().toString();
}
output.collect(null, new Text(str));
}
}
public static void main(String [] args) throws Exception{
/**
* 从关系数据库读取数据到HDFS
*/
JobConf job = new JobConf();
job.setJarByClass(SDBConnInput.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setInputFormat(DBInputFormat.class);
FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/user/xuyizhen/out"));
DBConfiguration.configureDB(job, "com.mysql.jdbc.Driver",
"jdbc:mysql://192.168.0.25:3306/hadoop","root","1117");
String fieldNames []={"customerID","customerName","phoneNumber"};
DBInputFormat.setInput(job, CustomerRecord.class,"customers",null,"customerID", fieldNames);
job.setMapperClass(MapperClass.class);
job.setReducerClass(ReducerClass.class);
JobClient.runJob(job);
}
}

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.lib.db.*;
import java.sql.*;
import java.io.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.filecache.*;
public class SDBConnOutput {
public static class CustomerRecord implements Writable,DBWritable{
String customerID;
String customerName;
String phoneNumber;
public void readFields(ResultSet resultSet)  throws SQLException{
customerID=resultSet.getString(1);
customerName=resultSet.getString(2);
phoneNumber=resultSet.getString(3);
}
public void write(PreparedStatement statement)  throws SQLException{
statement.setString(1, customerID);
statement.setString(2, customerName);
statement.setString(3,phoneNumber);
}
public void readFields(DataInput in) throws IOException{
customerID=in.readUTF();
customerName=in.readUTF();
phoneNumber=in.readUTF();
}
public void write(DataOutput out) throws IOException{
out.writeUTF(customerID);
out.writeUTF(customerName);
out.writeUTF(phoneNumber);
}
public void setCustomerID(String customerID){
this.customerID=customerID;
}
public void setCustomerName(String customerName){
this.customerName=customerName;
}
public void setPhoneNumber(String phoneNumber){
this.phoneNumber=phoneNumber;
}
public String toString(){
return this.customerID+","+this.customerName+","+this.phoneNumber;
}
}
public static class MapperClass extends MapReduceBase implements Mapper<LongWritable,Text,CustomerRecord,Text>{
CustomerRecord customer=new CustomerRecord();
public void map(LongWritable key, Text value,OutputCollector<CustomerRecord,Text> collector, Reporter reporter)  throws IOException{
String [] strs=value.toString().split(",");
customer.setCustomerID(strs[0]);
customer.setCustomerName(strs[1]);
customer.setPhoneNumber(strs[2]);
collector.collect( customer,value);
}
}
/**
*将HDFS中的文件输出到数据库
*/
public static void main(String [] args) throws Exception{
/**
* 从关系数据库读取数据到HDFS
*/
JobConf job = new JobConf(SDBConnInput.class);
//DBOutputFormat类只会将MapReduce框架输出结果的K值输出到关系数据库中
job.setOutputFormat(DBOutputFormat.class);
FileInputFormat.addInputPath(job, new Path("hdfs://master:9000/user/xuyizhen/in/customer.txt"));
DBConfiguration.configureDB(job, "com.mysql.jdbc.Driver",
"jdbc:mysql://192.168.0.25:3306/hadoop","root","1117");
String fieldNames []={"customerID","customerName","phoneNumber"};
DBOutputFormat.setOutput(job, "customers", fieldNames);
job.setMapperClass(MapperClass.class);
job.setNumReduceTasks(0);
JobClient.runJob(job);
}
}

注意：运行MapReduce时候报错：
java.io.IOException: com.mysql.jdbc.Driver
一般是由于程序找不到mysql驱动包。解决方法是让每个tasktracker运行MapReduce程序时都可以找到该驱动包。
添加包有两种方式：
1.在每个节点下的${HADOOP_HOME}/lib下添加该包，然后重启集群，这是比较原始的方法。
2.把包传到集群上:hadoop fs -put mysql驱动jar包名称/lib,并且在提交job前,添加语句DistributedCache.addFileToClassPath(new Path("/lib/mysql驱动jar包名称"),conf);
以上方法使用与所有需要额外jar包的MapReduce代码。

mysql-connector-java-5.1.22-bin.jar (813.4 KB)
下载次数: 3

Hadoop与关系数据库相关推荐

Hadoop冷热数据转换工具Sqoop
Sqoop是Apache基金下的开源项目,目的是完成关系数据库和Hadoop的转化,实现双向导入. 通常的大型数据系统使用之实现较少,主要原因是因为其在CDH3才开始支持,而且该方式是通过JDBC驱动 ...
Hadoop辅助工具——Flume、Sqoop
前言在一个完整的离线大数据处理系统中,除了hdfs+mapreduce+hive组成分析系统的核心之外,还需要数据采集.结果数据导出.任务调度等不可或缺的辅助系统,而这些辅助工具在hadoop生态体 ...
Hadoop生态系统各组件功能
参考: Spark编程基础(Scala版) 林子雨赖永炫陶继平人民邮电出版社出版 2018-07-01 1.HDFS 分布式文件系统 Hadoop分布式文件系统HDFS是针对谷歌分布式文件系统( ...
Hadoop生态系统功能组件，主要包括哪些？
经过多年的发展,Hadoop生态系统不断完善和成熟,目前已经包括了多个子项目,除了核心的HDFS和MapReduce以外,Hadoop生态系统还包括要ZoopKer.HBase.Hive.Pig.Ma ...
Hadoop笔记-01概述
文章目录 1 什么是大数据? 1.1 大数据计算模式及代表产品 1.2 云计算与物联网 1.2.1 云计算 1.2.1.1 虚拟化 1.2.1.2 分布式存储 1.2.1.3 分布式计算 1.2.1. ...
Hadoop 和 Spark 知识点整理汇总
文章目录前言一.LINUX 系统常用命令汇总二.Hadoop 常用命令汇总三.Hadoop 基本概念 1. Hadoop 特性 2. Hadoop 架构 2.1 Hadoop 集群 2.2 H ...
hadoop生态圈各产品基本概念梳理
hbase:是一个适合于非结构化数据存储的数据库,是基于列的而不是基于行的模式,HBase利用Hadoop MapReduce来处理HBase中的海量数据. HDFS: 是GFS的一种实现,他的完整名 ...
急性子的开源大数据，第 1 部分: Hadoop 教程：Hello World 与 Java、Pig、Hive、Flume、Fuse、Oozie，以及 Sqoop 与 Informix、DB2 和
如何开始使用 Hadoop 和自己喜欢的数据库本文的重点是解释大数据,然后在 Hadoop 中提供简单的工作示例,Hadoop 是在大数据领域的主要开源选手.您会很高兴地听到,Hadoop 并不是 ...
Hadoop离线 day18 sqoop数据迁移和java执行shell命令
sqoop数据迁移和java执行shell命令 3. sqoop数据迁移 3.1.概述 3.2.sqoop1与sqoop2架构对比 3.3.工作机制 3.4 .sqoop实战及原理 3.4.1 sqo ...

Hadoop与关系数据库

Hadoop与关系数据库相关推荐

最新文章

热门文章