Hadoop主要用来对非结构化或半结构化(HBase)数据进行存储和分析,而结构化的数据则一般使用数据库来进行存储和访问。本文的主要内容则是讲述如何将Hadoop与现有的数据库结合起来,在Hadoop应用程序中访问数据库中的文件。
1.DBInputFormat
DBInputFormat是Hadoop从0.19.0开始支持的一种输入格式,包含在包org.apache.hadoop.mapred.lib.db中,主要用来与现有的数据库系统进行交互,包括MySQL、PostgreSQL、Oracle等几个数据库系统。DBInputFormat在Hadoop应用程序中通过数据库供应商提供的JDBC接口来与数据库进行交互,并且可以使用标准的SQL来读取数据库中的记录。在使用DBInputFormat之前,必须将要使用的JDBC驱动拷贝到分布式系统各个节点的$HADOOP_HOME/lib/目录下。
在DBInputFormat类中包含以下三个内置类:
1.protected class DBRecordReader
implements RecordReader<LongWritable, T>:用来从一张数据库表中读取一条条元组记录。
2.public static
class NullDBWritable implements DBWritable, Writable:主要用来实现DBWritable接口。
3.protected static
class DBInputSplit implements InputSplit:主要用来描述输入元组集合的范围,包括start和end两个属性,start用来表示第一条记录的索引号,end表示最后一条记录的索引号。
其中DBWritable接口与Writable接口比较类似,也包含write和readFields两个函数,只是函数的参数有所不同。DBWritable中的两个函数分别为:
public void
write(PreparedStatement statement) throws SQLException;
public void
readFields(ResultSet resultSet) throws SQLException;
这两个函数分别用来给java.sql.PreparedStatement设置参数,以及从java.sql.ResultSet中读取一条记录,熟悉Java JDBC用法的应该对这两个类的用法比较了解。
2.使用DBInputFormat读取数据库表中的记录
上文已经对DBInputFormat以及其中的相关内置类作了简单介绍,下面对怎样使用DBInputFormat读取数据库记录进行详细的介绍,具体步骤如下:
1.使用DBConfiguration.configureDB (JobConf
job, String driverClass, String dbUrl, String userName, String passwd)函数配置JDBC驱动,数据源,以及数据库访问的用户名和密码。例如MySQL数据库的JDBC的驱动为“com.mysql.jdbc.Driver”,数据源可以设置为“jdbc:mysql://localhost/mydb”,其中mydb可以设置为所需要访问的数据库。
2.使用DBInputFormat.setInput(JobConf
job, Class<? extends DBWritable> inputClass, String tableName, String
conditions, String orderBy, String... fieldNames)函数对要输入的数据进行一些初始化设置,包括输入记录的类名(必须实现了DBWritable接口)、数据表名、输入数据满足的条件、输入顺序、输入的属性列。也可以使用重载的函数setInput(JobConf
job, Class<? extends DBWritable> inputClass, String inputQuery, String
inputCountQuery)进行初始化,区别在于后者可以直接使用标准SQL进行初始化,具体可以参考Hadoop API中的讲解。
3.按照普通Hadoop应用程序的格式进行配置,包括Mapper类、Reducer类、输入输出文件格式等,然后调用JobClient.runJob(conf)。
3.使用示例
假设MySQL数据库中有数据库school,其中的teacher数据表定义如下:
DROP TABLE IF
EXISTS `school`.`teacher`;
CREATE TABLE `school`.`teacher` (
`id` int(11) default NULL,
`name` char(20) default NULL,
`age` int(11) default NULL,
`departmentID` int(11) default NULL
) ENGINE=InnoDB
DEFAULT CHARSET=latin1;
首先给出实现了DBWritable接口的TeacherRecord类:
public class TeacherRecord implements
Writable, DBWritable{
int id;
String name;
int age;
int departmentID;
@Override
public void
readFields(DataInput in) throws IOException {
// TODO Auto-generated
method stub
this.id = in.readInt();
this.name = Text.readString(in);
this.age = in.readInt();
this.departmentID =
in.readInt();
}
@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.age);
out.writeInt(this.departmentID);
}
@Override
public void
readFields(ResultSet result) throws SQLException {
// TODO Auto-generated
method stub
this.id = result.getInt(1);
this.name =
result.getString(2);
this.age = result.getInt(3);
this.departmentID =
result.getInt(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.setInt(3, this.age);
stmt.setInt(4, this.departmentID);
}
@Override
public String toString() {
// TODO Auto-generated
method stub
return new String(this.name
+ " " + this.age + " " + this.departmentID);
}
}
利用DBAccessMapper读取一条条记录:
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
collector.collect(new
LongWritable(value.id),
new Text(value.toString()));
}
}
Main函数如下:
public class DBAccess {
public static void
main(String[] args) throws IOException {
JobConf conf = new
JobConf(DBAccess.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", "age", "departmentID"};
DBInputFormat.setInput(conf,
TeacherRecord.class, "teacher",
null, "id", fields);
conf.setMapperClass(DBAccessMapper.class);
conf.setReducerClass(IdentityReducer.class);
JobClient.runJob(conf);
}
}
该示例从teacher表中读取所有记录,并以TextOutputFormat的格式输出到dboutput目录下,输出格式为<”id”, “name
age departmentID”>。
4.使用DBOutputFormat向数据库中写记录
DBOutputFormat将计算结果写回到一个数据库,同样先调用DBConfiguration.configureDB()函数进行数据库配置,然后调用函数DBOutputFormat.setOutput (JobConf job, String tableName, String... fieldNames)进行初始化设置,包括数据库表名和属性列名。同样,在将记录写回数据库之前,要先实现DBWritable接口。每个DBWritable的实例在传递给Reducer中的OutputCollector时都将调用其中的write(PreparedStatement
stmt)方法。在Reduce过程结束时,PreparedStatement中的对象将会被转化成SQL语句中的INSERT语句,从而插入到数据库中。
5.总结
DBInputFormat和DBOutputFormat提供了一个访问数据库的简单接口,虽然接口简单,但应用广泛。例如,可以将现有数据库中的数据转储到Hadoop中,由Hadoop进行分布式计算,通过Hadoop对海量数据进行分析,然后将分析后的结果转储到数据库中。在搜索引擎的实现中,可以通过Hadoop将爬行下来的网页进行链接分析,评分计算,建立倒排索引,然后存储到数据库中,通过数据库进行快速搜索。虽然上述的数据库访问接口已经能满足一般的数据转储功能,但是仍然存在一些限制不足,例如并发访问、数据表中的键必须要满足排序要求等,还需Hadoop社区的人员进行改进和优化。
分享到:
相关推荐
使用hadoop来访问数据库,详细说明了访问数据库的原理和步骤
hadoop连接数据库查询数据,并添加到hdfs;从hdfs进行mapreduce数据导入到数据库 hadoop连接数据库查询数据,并添加到hdfs;从hdfs进行mapreduce数据导入到数据库 hadoop连接数据库查询数据,并添加到hdfs;从hdfs...
HBase 是基于 Apache Hadoop 的面向列的 NoSQL 数据库,是 Google 的 BigTable 的开源实现。HBase 是一个针对半结构化数据的开源的、多版本的、可伸缩的、高可靠的、高性能的、分布式的和面向列的动态模式数据库。 ...
Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展。Hbase中的数据是面向列的数据库,其中结构化数据存储在键值对中。Hbase用Java编写。Hbase的灵感来自Google Paper-“大表:结构化数据的分布式...
修改 src/main/resources/config/hadoop.properties 中的 HDFS 和 namenode 配置 修改 CommentServiceImpl 类 analyzeComment 方法下的日志输出路径、Hadoop Example 路径 访问 http://localhost:8080 主要功能 员工...
第五天 hadoop2.x中HA机制的原理和全分布式集群安装部署及维护 01-zookeeper.avi 02-zookeeper2.avi 03-NN高可用方案的要点1.avi 04-hadoop-HA机制的配置文件.avi 05-hadoop分布式集群HA模式部署.avi 06-hdfs...
大数据 数据库hadoop-2.5.1 、2.7.1组件包大集合 数据库访问驱动数据库开发驱动
基于Hadoop部署实践对网站日志分析 1. 项目概述 本次要实践的数据日志来源于国内某技术学习论坛,该...NameNode它是Hadoop 中的主服务器,管理文件系统名称空间和对集群中存储的文件的访问。 Secondary NameNode.....
毕业设计,基于Python+Spark+Hadoop大数据开发的用户画像电影推荐系统,内含Python完整源代码,数据库脚本 Python+Spark+Hadoop大数据基于用户画像电影推荐系统设计 程序开发软件: Pycharm + Python3.7 + Django...
第1章 Hadoop简介 1.1 为什么写《Hadoop 实战》 1.2 什么是Hadoop 1.3 了解分布式系统和Hadoop 1.4 比较SQL 数据库和Hadoop 1.5 理解MapReduce 1.5.1 动手扩展一个简单程序 1.5.2 相同程序在MapReduce中的...
Hadoop中的数据库访问 MapReduce中多文件输出的使用 Zookeeper使用与分析 浅析一种分类数据模型 Sector框架分析 Run on Hadoop
本系统是以Django作为基础框架,采用MTV模式,数据库使用MySQL和Redis,以从豆瓣平台爬取的电影数据作为基础数据源,主要基于用户的基本信息和使用操作记录等行为信息来开发用户标签,并使用Hadoop、Spark大数据...
基于Hadoop的商品推荐系统源码+数据库文件+使用教程.zip已获导师指导并通过的高分项目 安装教程 创建数据库并导入sql脚本 运行com.cy.store.StoreApplication 主程序 打开浏览器 localhost:8081 访问 大数据运行: ...
1 Hadoop 中的数据库访问 2 MapReduce 中多文件输出的使用 3 Zookeeper 使用与分析 4 浅析一种分类数据模型 5 Sector 框架分析 6 Run on Hadoop
一个大型的分布式数据库,这个数据库不是关系式的数据库。像它的名字一样,就是一个巨大的表格,用来存储结构化的数据。 以上三个设施Google均有论文发表。 开源实现 Hadoop是项目的总称,起源于作者儿子的一...
关系数据库已经流行很多年,并且Hadoop已经有了HDFS和MapReduce,为什么需要HBase? Hadoop可以很好地解决大规模数据的离线批量处理问题,但是,受限于Hadoop MapReduce编程框架的高延迟数据处理机制,使得Hadoop...
本系统是以Django作为基础框架,采用MTV模式,数据库使用MySQL和Redis,以从豆瓣平台爬取的电影数据作为基础数据源,主要基于用户的基本信息和使用操作记录等行为信息来开发用户标签,并使用Hadoop、Spark大数据...
hadoop 中文文档index Prefetch chain 预处理链 Pre selector Precondition Enforcer Fetch chain 提取链 Fetch DNS Fetch Http Extractor chain抽取链 Extractor HTML Extractor JS Write Chain 写链 ARC ...
最后在Hadoop数据服务平台上实现了面向软件客户端的数据访问服务。设计了适合通信社会网络数据特色的 Map/Reduce算法。通过MaP和Reduee函数实现了数据的并行查询和处理。在数据查询过程中,还将数据遍历过程放Reduce...
1669.2.3 准备SSH密钥对 1689.3 在EC2上安装Hadoop 1699.3.1 配置安全参数 1699.3.2 配置集群类型 1699.4 在EC2上运行MapReduce程序 1719.4.1 将代码转移到Hadoop集群上 1719.4.2 访问Hadoop集群上的...