`
郑云飞
  • 浏览: 814652 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

如何使用Hadoop读写数据库

 
阅读更多

在我们的一些应用程序中,常常避免不了要与数据库进行交互,而在我们的hadoop中,有时候也需要和数据库进行交互,比如说,数据分析的结果存入数据库,或者是,读取数据库的信息写入HDFS上,不过直接使用MapReduce操作数据库,这种情况在现实开发还是比较少,一般我们会采用Sqoop来进行数据的迁入,迁出,使用Hive分析数据集,大多数情况下,直接使用Hadoop访问关系型数据库,可能产生比较大的数据访问压力,尤其是在数据库还是单机的情况下,情况可能更加糟糕,在集群的模式下压力会相对少一些。 

那么,今天散仙就来看下,如何直接使用Hadoop1.2.0的MR来读写操作数据库,hadoop的API提供了DBOutputFormat和DBInputFormat这两个类,来进行与数据库交互,除此之外,我们还需要定义一个类似JAVA Bean的实体类,来与数据库的每行记录进行对应,通常这个类要实现Writable和DBWritable接口,来重写里面的4个方法以对应获取每行记录里面的各个字段信息。 

下面,我们先来看下如何使用MR来读取数据库的数据,并写入HDFS上, 
数据表的截图如下所示, 



 
实体类定义代码: 
<pre name="code" class="java">package com.qin.operadb; 


import java.io.DataInput; 
import java.io.DataOutput; 
import java.io.IOException; 
import java.sql.PreparedStatement; 
import java.sql.ResultSet; 
import java.sql.SQLException; 


import org.apache.hadoop.io.Text; 
import org.apache.hadoop.io.Writable; 
import org.apache.hadoop.mapreduce.lib.db.DBWritable; 


/*** 
* 封装数据库实体信息 
* 的记录 
* 搜索大数据技术交流群:376932160 
* **/ 
public class PersonRecoder implements Writable,DBWritable { 


public int id;//对应数据库中id字段 
public String name;//对应数据库中的name字段 
public int age;//对应数据库中的age字段 








@Override 
public void readFields(ResultSet result) throws SQLException { 




this.id=result.getInt(1); 
this.name=result.getString(2); 
this.age=result.getInt(3); 




@Override 
public void write(PreparedStatement stmt) throws SQLException { 


stmt.setInt(1, id); 
stmt.setString(2, name); 
stmt.setInt(3, age); 




@Override 
public void readFields(DataInput arg0) throws IOException { 
// TODO Auto-generated method stub 
this.id=arg0.readInt(); 
this.name=Text.readString(arg0); 
this.age=arg0.readInt(); 




@Override 
public void write(DataOutput out) throws IOException { 
// TODO Auto-generated method stub 
out.writeInt(id); 
Text.writeString(out, this.name); 
out.writeInt(this.age); 




@Override 
public String toString() { 
// TODO Auto-generated method stub 
return "id: "+id+"  年龄: "+age+"   名字:"+name; 




</pre> 
MR类的定义代码,注意是一个Map Only作业: 
<pre name="code" class="java">package com.qin.operadb; 


import java.io.IOException; 


import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapred.JobConf; 
import org.apache.hadoop.mapred.lib.IdentityReducer; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration; 
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 


public class ReadMapDB { 






/** 
* Map作业读取数据记录数 
* **/ 
private static class DBMap extends Mapper<LongWritable, PersonRecoder , LongWritable, Text>{ 
@Override 
protected void map(LongWritable key, PersonRecoder value,Context context) 
throws IOException, InterruptedException { 


context.write(new LongWritable(value.id), new Text(value.toString())); 




public static void main(String[] args)throws Exception { 




JobConf conf=new JobConf(ReadMapDB.class); 
//Configuration conf=new Configuration(); 
   // conf.set("mapred.job.tracker","192.168.75.130:9001"); 
//读取person中的数据字段 
// conf.setJar("tt.jar"); 


//注意这行代码放在最前面,进行初始化,否则会报 
DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test", "root", "qin"); 


/**要读取的字段信息**/ 
String fileds[]=new String[]{"id","name","age"}; 
/**Job任务**/ 
Job job=new Job(conf, "readDB"); 
System.out.println("模式:  "+conf.get("mapred.job.tracker")); 


/**设置数据库输入格式的一些信息**/ 
DBInputFormat.setInput(job, PersonRecoder.class, "person", null, "id", fileds); 
/***设置输入格式*/ 
job.setInputFormatClass(DBInputFormat.class); 
job.setOutputKeyClass(LongWritable.class); 
job.setOutputValueClass(Text.class); 
job.setMapperClass(DBMap.class); 
String path="hdfs://192.168.75.130:9000/root/outputdb"; 
FileSystem fs=FileSystem.get(conf); 
Path p=new Path(path); 
if(fs.exists(p)){ 
fs.delete(p, true); 
System.out.println("输出路径存在,已删除!"); 
FileOutputFormat.setOutputPath(job,p ); 
System.exit(job.waitForCompletion(true) ? 0 : 1);  




























</pre> 
写入到HDFS目录下数据集: 


 
读取相对比较简单,需要注意的第一注意JDBC的驱动jar包要在各个节点上分别上传一份,第二是在main方法里的驱动类的编写顺序,以及数据信息的完整,才是正确连接数据库并读取的关键。 




下面来看下,如何使用MR,分析完数据后的结果,写入在数据库中,散仙本篇测试的是一个简单的WordCount的统计。我们先来看下数据库表的信息: 


 


实体类定义代码: 
<pre name="code" class="java">package com.qin.operadb; 


import java.io.DataInput; 
import java.io.DataOutput; 
import java.io.IOException; 
import java.sql.PreparedStatement; 
import java.sql.ResultSet; 
import java.sql.SQLException; 


import org.apache.hadoop.io.Text; 
import org.apache.hadoop.io.Writable; 
import org.apache.hadoop.mapreduce.lib.db.DBWritable; 


public class WordRecoder implements Writable,DBWritable { 


public String word; 
public int count; 




@Override 
public void readFields(ResultSet rs) throws SQLException { 


this.word=rs.getString(1); 
this.count=rs.getInt(2); 




@Override 
public void write(PreparedStatement ps) throws SQLException { 


ps.setString(1, this.word); 
ps.setInt(2, this.count); 




@Override 
public void readFields(DataInput in) throws IOException { 


  this.word=Text.readString(in); 
  this.count=in.readInt(); 




@Override 
public void write(DataOutput out) throws IOException { 
  
   
Text.writeString(out, this.word); 
out.writeInt(count); 












</pre> 


统计的2个文件的内容所示: 


 
MR的核心类代码: 
<pre name="code" class="java">package com.qin.operadb; 


import java.io.IOException; 
import java.util.StringTokenizer; 


import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapred.JobConf; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration; 
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat; 
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 




public class WriteMapDB { 






private static class WMap extends Mapper<LongWritable, Text, Text, IntWritable>{ 




/*** 
* Mapper的参数类型介绍 
* K,V,K,V分别依次代表 
* Map作业输入类型的K,输入类型的V 
* 后面两个是输出类型的K,输出类型的V 
* 后面的两个与 context.write(word, one); 
* 的两个参数是对应的 
* **/ 
private Text word=new Text(); 
private IntWritable one=new IntWritable(1); 


@Override 
protected void map(LongWritable key, Text value,Context context) 
throws IOException, InterruptedException { 


String line=value.toString(); 
//处理记事本UTF-8的BOM问题 


            if (line.getBytes().length > 0) { 


                if ((int) line.charAt(0) == 65279) { 


                    line = line.substring(1); 


                } 


            } 


StringTokenizer st=new StringTokenizer(line); 
while(st.hasMoreTokens()){ 
word.set(st.nextToken());//设置单词 
context.write(word, one); 
















/*** 
* 由于在reduce中,需要向数据库里写入 
* 数据,所以跟combine,不能共用 
* ***/ 
private static class WCombine extends Reducer<Text, IntWritable, Text, IntWritable>{ 


@Override 
protected void reduce(Text text, Iterable<IntWritable> value,Context context) 
throws IOException, InterruptedException { 




int sum=0; 


for(IntWritable iw:value){ 
sum+=iw.get(); 
context.write(text, new IntWritable(sum)); 






















/** 
* Reduce类 
* **/ 
private static class WReduce extends Reducer<Text, IntWritable, WordRecoder, Text>{ 




@Override 
protected void reduce(Text key, Iterable<IntWritable> values,Context context) 
throws IOException, InterruptedException { 


  int sum=0; 


  for(IntWritable s:values){ 
  sum+=s.get(); 
  } 
  
  
  WordRecoder wr=new WordRecoder(); 
  wr.word=key.toString(); 
  wr.count=sum; 
  
  //写出到数据库里 
  context.write(wr, new Text()); 




























public static void main(String[] args)throws Exception { 




JobConf conf=new JobConf(WriteMapDB.class); 
//Configuration conf=new Configuration(); 
   // conf.set("mapred.job.tracker","192.168.75.130:9001"); 
//读取person中的数据字段 
  //conf.setJar("tt.jar"); 
// conf.setNumReduceTasks(1); 
//注意这行代码放在最前面,进行初始化,否则会报 
/**建立数据库连接**/ 
DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.211.36:3306/test?characterEncoding=utf-8", "root", "qin"); 
String fileds[]=new String[]{"word","count"}; 


Job job=new Job(conf, "writeDB"); 


System.out.println("运行模式:  "+conf.get("mapred.job.tracker")); 


/**设置输出表的的信息  第一个参数是job任务,第二个参数是表名,第三个参数字段项**/ 
DBOutputFormat.setOutput(job, "wordresult", fileds); 


/**设置DB的输入路径**/ 
job.setInputFormatClass(TextInputFormat.class); 
/**设置DB的输出路径**/ 
job.setOutputFormatClass(DBOutputFormat.class); 




/***设置Reduce的个数为1,可以得到全局统计的数字 
* 但,需要注意,在分布式环境下,最好不要设置为1,Reduce的个数 
* 正是Hadoop并发能力的体现 
* **/ 
// job.setNumReduceTasks(1); 




/**设置输出K路径**/ 
job.setOutputKeyClass(Text.class); 
/**设置输出V路径**/ 
job.setOutputValueClass(IntWritable.class); 




/**设置Map类**/ 
job.setMapperClass(WMap.class); 
/**设置Combiner类**/ 
job.setCombinerClass(WCombine.class); 
/**设置Reduce类**/ 
job.setReducerClass(WReduce.class); 






/**设置输入路径*/ 
FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.75.130:9000/root/input")); 






System.exit(job.waitForCompletion(true) ? 0 : 1);  








</pre> 
运行状态如下所示: 
<pre name="code" class="java">运行模式:  192.168.75.130:9001 
14/03/26 20:26:59 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 
14/03/26 20:27:01 INFO input.FileInputFormat: Total input paths to process : 2 
14/03/26 20:27:01 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable 
14/03/26 20:27:01 WARN snappy.LoadSnappy: Snappy native library not loaded 
14/03/26 20:27:01 INFO mapred.JobClient: Running job: job_201403262328_0006 
14/03/26 20:27:02 INFO mapred.JobClient:  map 0% reduce 0% 
14/03/26 20:27:10 INFO mapred.JobClient:  map 50% reduce 0% 
14/03/26 20:27:11 INFO mapred.JobClient:  map 100% reduce 0% 
14/03/26 20:27:18 INFO mapred.JobClient:  map 100% reduce 33% 
14/03/26 20:27:19 INFO mapred.JobClient:  map 100% reduce 100% 
14/03/26 20:27:20 INFO mapred.JobClient: Job complete: job_201403262328_0006 
14/03/26 20:27:20 INFO mapred.JobClient: Counters: 28 
14/03/26 20:27:20 INFO mapred.JobClient:   Job Counters 
14/03/26 20:27:20 INFO mapred.JobClient:     Launched reduce tasks=1 
14/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_MAPS=10345 
14/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all reduces waiting after reserving slots (ms)=0 
14/03/26 20:27:20 INFO mapred.JobClient:     Total time spent by all maps waiting after reserving slots (ms)=0 
14/03/26 20:27:20 INFO mapred.JobClient:     Launched map tasks=2 
14/03/26 20:27:20 INFO mapred.JobClient:     Data-local map tasks=2 
14/03/26 20:27:20 INFO mapred.JobClient:     SLOTS_MILLIS_REDUCES=8911 
14/03/26 20:27:20 INFO mapred.JobClient:   File Output Format Counters 
14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Written=0 
14/03/26 20:27:20 INFO mapred.JobClient:   FileSystemCounters 
14/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_READ=158 
14/03/26 20:27:20 INFO mapred.JobClient:     HDFS_BYTES_READ=325 
14/03/26 20:27:20 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=182065 
14/03/26 20:27:20 INFO mapred.JobClient:   File Input Format Counters 
14/03/26 20:27:20 INFO mapred.JobClient:     Bytes Read=107 
14/03/26 20:27:20 INFO mapred.JobClient:   Map-Reduce Framework 
14/03/26 20:27:20 INFO mapred.JobClient:     Map output materialized bytes=164 
14/03/26 20:27:20 INFO mapred.JobClient:     Map input records=6 
14/03/26 20:27:20 INFO mapred.JobClient:     Reduce shuffle bytes=164 
14/03/26 20:27:20 INFO mapred.JobClient:     Spilled Records=24 
14/03/26 20:27:20 INFO mapred.JobClient:     Map output bytes=185 
14/03/26 20:27:20 INFO mapred.JobClient:     Total committed heap usage (bytes)=336338944 
14/03/26 20:27:20 INFO mapred.JobClient:     CPU time spent (ms)=2850 
14/03/26 20:27:20 INFO mapred.JobClient:     Combine input records=20 
14/03/26 20:27:20 INFO mapred.JobClient:     SPLIT_RAW_BYTES=218 
14/03/26 20:27:20 INFO mapred.JobClient:     Reduce input records=12 
14/03/26 20:27:20 INFO mapred.JobClient:     Reduce input groups=8 
14/03/26 20:27:20 INFO mapred.JobClient:     Combine output records=12 
14/03/26 20:27:20 INFO mapred.JobClient:     Physical memory (bytes) snapshot=464982016 
14/03/26 20:27:20 INFO mapred.JobClient:     Reduce output records=8 
14/03/26 20:27:20 INFO mapred.JobClient:     Virtual memory (bytes) snapshot=2182836224 
14/03/26 20:27:20 INFO mapred.JobClient:     Map output records=20 
</pre> 
最后,我们就可以去数据库里,查看统计的信息了,截图如下: 


 
至此,我们就完成了使用MR来读写数据库了,注意测试前,先确保自己的hadoop集群,可以正常工作。 
 
分享到:
评论

相关推荐

    Hbase操作Hadoop分布式数据库java工程实例(含测试用例)

    在本Java工程实例中,我们将深入探讨如何使用HBase这一分布式数据库系统,它与Hadoop生态系统紧密集成。HBase是一个高度可扩展的NoSQL数据库,它以列族为存储结构,支持实时读写操作,特别适合大数据场景。下面将...

    Hadoop HBase数据库简介

    HBase 是基于 Apache Hadoop 的面向列的 NoSQL 数据库,是 Google 的 BigTable 的开源实现。HBase 是一个针对半结构化数据的开源的、多版本的、可伸缩的、高可靠的、高性能的、分布式的和面向列的动态模式数据库。 ...

    读写数据库数据的mr程序

    本程序的标题是"读写数据库数据的mr程序",这意味着我们将探讨如何利用MapReduce来从MySQL数据库读取数据,并将这些数据存储到HBASE这种分布式数据库中。这个过程通常被称为ETL(Extract, Transform, Load)操作,是...

    分布式数据库Hive笔记_HDFS_Hadoop_分布式数据库

    总的来说,学习这些内容可以帮助我们深入理解大数据处理的工作原理,掌握如何在Hadoop生态系统中使用Hive进行高效的数据分析。通过阅读和研究这些资料,你将能够熟练运用Hive进行大数据查询,同时了解Hadoop的运行...

    基于Hadoop云的数据库营销海量数据处理与挖掘的研究.docx

    例如,通过Hive(一个建立在Hadoop之上的数据仓库工具)可以方便地执行SQL查询操作,而HBase则提供了一种高性能的列式存储模式,适合实时读写和随机访问的场景。 ##### 2.3 Hadoop与传统数据库的比较 相比于传统的...

    Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展.rar

    Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展。Hbase中的数据是面向列的数据库,其中结构化数据存储在键值对中。Hbase用Java编写。Hbase的灵感来自Google Paper-“大表:结构化数据的分布式...

    Google +Hadoop使用编程

    不同于传统的关系型数据库,Bigtable使用一个简单的二维表模型来存储数据,其中每一行都由一个唯一的行键标识。Bigtable支持高并发的读写操作,能够应对大规模的数据存储需求。 **HBase** 基于HDFS构建,是对...

    hadoop/bin目录文件,含hadoop.dll + winutils.exe

    在Windows 7上,开发人员可以使用hadoop.dll和winutils.exe来访问HDFS,进行数据的读写操作。而Hive是基于Hadoop的数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供SQL查询功能,因此在Windows环境...

    hadoop.zip hadoop2.7.1安装包

    安装并配置好winutils后,Windows用户可以通过Hadoop的HDFS API访问HDFS,进行数据读写操作。 在大数据处理中,Hadoop的核心组件包括: 1. HDFS(Hadoop Distributed File System):分布式文件系统,能够将大规模...

    大数据安全-kerberos技术-hadoop安装包,hadoop版本:hadoop-3.3.4.tar.gz

    在Hadoop环境中,Kerberos可以应用于HDFS、YARN、HBase等组件,保护数据的读写操作、资源调度以及数据库访问。安装Hadoop-3.3.4时,需确保配置Kerberos以实现安全运行。首先,需要设置Kerberos的KDC服务器,并为每个...

    hadoop-2.6.0

    此外,Hadoop 2.6.0还包含许多其他组件,如HBase(一个分布式数据库)、Hive(一个数据仓库工具)、Pig(一种高级数据处理语言)和Sqoop(用于在Hadoop和传统数据库间导入导出数据的工具)。这些组件共同构建了一个...

    分布式文件系统Hadoop

    HBase是一个基于Hadoop的分布式数据库,提供实时读写和强一致性。该文档会讲解如何安装HBase,配置HBase与Hadoop的交互,以及如何使用HBase Shell进行基本操作。 "Hadoop的MapReduce执行过程介绍"涵盖了MapReduce...

    hadoop+hbase jar包

    例如,通过`addDependency`或者`classpath`指定jar路径,然后使用Hadoop的API读写HDFS,使用HBase的API操作数据库。同时,开发过程中需要注意版本兼容性,确保所使用的jar包与Hadoop和HBase集群的版本一致。 6. ...

    Hadoop数据迁移--从Hadoop向Oracle

    为了实现这种迁移,我们通常使用MapReduce编程模型,它允许Hadoop访问和写入Oracle数据库,并将数据从HDFS传输到Oracle中。 首先,要实现Hadoop到Oracle的数据迁移,需要配置相关的数据库信息类DBInfo,该类记录了...

    hadoop-2.5.2.tar.gz

    Hadoop生态系统还包括HBase、Hive、Pig、Oozie等工具,它们分别提供了NoSQL数据库服务、SQL查询接口、数据流处理和工作流调度等功能,共同构建了一个强大的大数据处理平台。 总结来说,“hadoop-2.5.2.tar.gz”是...

    HadoopAPI使用

    org.apache.hadoop.io 包定义了通用的 I/O API,用于读写各种数据对象,包括网络、数据库和文件等。org.apache.hadoop.ipc 包提供了网络服务端和客户端的工具,用于封装网络异步 I/O 操作。org.apache.hadoop.mapred...

    hadoop1.0.2 hbase0.94安装

    在IT行业中,分布式计算框架Hadoop和NoSQL数据库HBase是大数据处理的重要工具。本教程主要涵盖的是在较旧版本的Hadoop 1.0.2上安装并配置HBase 0.94,以及相关的MapReduce开发和Hadoop-Eclipse插件的编译。这些内容...

    java下hadoop开发使用jar包

    除了核心组件,Hadoop生态还包括许多其他模块,如HBase(分布式数据库)、Hive(数据仓库工具)、Pig(数据分析语言)等。如果项目中用到了这些模块,还需要额外引入对应的jar包。例如,对于HBase,你需要添加`hbase...

    prometheus grafana 基于开源监控apache Hadoop模板大全

    HBase是构建在Hadoop之上的分布式NoSQL数据库,适用于大规模实时读写操作。监控HBase的指标如RegionServer负载、MemStore大小、延迟时间等,可以帮助我们优化读写性能,预防热点问题,以及提前发现潜在的故障。 这...

Global site tag (gtag.js) - Google Analytics