数据在HDFS和HBASE之间互相传递的过程

　　对于简单的结构化数据，我们在HDFS和HBASE上的传递可能只需要用框架即可完成，但是对于复杂的数据传输，特别是实际工作中，数据的收集整理并非简单的结构，因此，我们需要对数据重新整理，并进行发送。这个过程就是依赖MapReduce，通过底层对数据的拆分和重组，达到我们要传输的结构要求。

下面我们开始进行一个简单的小测试：

从HDFS 到HBASE

首先，我们在虚拟机本地创建一个临时文件demo，简单的结构化数据为name，age，sex。值可以适当添加几行即可。然后通过命令

hdfs dfs -put demo

此时已经将文件存放到了HDFS上，然后在HBASE中，创建我们需要存储数据的表，创建命令：

create 'tb_test','info'

hbase 中表建立为表名tb_test,列族名info。

接下来我们编写对应的代码：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import java.io.IOException;public class HdfstoHbase
{public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf=HBaseConfiguration.create();Job job=Job.getInstance(conf, "hdfs-to-hbase");job.setJarByClass(HdfstoHbase.class);job.setMapperClass(HdfsMap.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);TableMapReduceUtil.initTableReducerJob("tb_test",HbaseReduce.class,job );FileInputFormat.addInputPath(job,new Path("/user/bda/demo"));System.out.println(job.waitForCompletion(true)?0:1);}public static class HdfsMap extends Mapper<Object,Text,Text,NullWritable>{@Overrideprotected void map(Object key, Text value, Context context) throws IOException, InterruptedException {context.write(value,NullWritable.get() );}}public static class HbaseReduce extends TableReducer<Text,NullWritable,ImmutableBytesWritable>{private final byte[] cf="info".getBytes();private final byte[] name="name".getBytes();private final byte[] age="age".getBytes();private final byte[] sex="sex".getBytes();@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {String[] ss=key.toString().split(",");Put put=new Put(key.toString().getBytes());put.addColumn(cf,name ,ss[0].getBytes() );put.addColumn(cf,age ,ss[1].getBytes() );put.addColumn(cf,sex ,ss[2].getBytes() );context.write(null,put);}}
}

典型的MapReduce结构框架。然后打包，如果你打的是瘦包，那么，执行jar文件的过程你需要添加相应的依赖，或者是脚本提前写好执行，下面直接贴出脚本执行命令：

HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase classpath`:`${HBASE_HOME}/bin/hbase mapredcp` hadoop jar 打包的jar文件 文件的类名

执行后，在HBASE中，tb_test表即可查看到传入的数据。

HBASE到HDFS

同样的操作，这里只粘贴代码部分，其余操作类似。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class HbasetoHdfs {public static void main(String[] args) throws Exception {Configuration cfg = HBaseConfiguration.create();Job job = Job.getInstance(cfg,"hbase-to-hdfs");job.setJarByClass(HbasetoHdfs.class);Scan scan = new Scan();scan.setCaching(50);        // 1 is the default in Scan, which will be bad for MapReduce jobsscan.setCacheBlocks(false);TableMapReduceUtil.initTableMapperJob("tb_test",scan,HbaseMap.class,Text.class,NullWritable.class,job);job.setReducerClass(HdfsReduce.class);FileOutputFormat.setOutputPath(job,new Path("/user/bda/dsj3"));System.out.println(job.waitForCompletion(true)?0:1);}public static class HbaseMap extends TableMapper<Text,NullWritable> {private final byte[] cf="info".getBytes();private final byte[] name="name".getBytes();private final byte[] age="age".getBytes();private final byte[] sex="sex".getBytes();@Overrideprotected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {String line="";String n = Bytes.toString(value.getValue(cf,name));String a = Bytes.toString(value.getValue(cf,age));String s = Bytes.toString(value.getValue(cf,sex));line = n+","+a+","+s;context.write(new Text(line),NullWritable.get());}}public static class HdfsReduce extends Reducer<Text,NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {context.write(key,NullWritable.get());}}}

转载于:https://www.cnblogs.com/qianshuixianyu/p/9871897.html

数据在HDFS和HBASE之间互相传递的过程相关推荐

利用Sqoop将MySQL海量测试数据导入HDFS和HBase
声明:作者原创,转载注明出处. 作者:帅气陈吃苹果一.安装Sqoop 1.下载sqoop,解压.文件夹重命名 wget http://mirror.bit.edu.cn/apache/sqoop/1 ...
Thinking in BigData（八）大数据Hadoop核心架构HDFS+MapReduce+Hbase+Hive内部机理详解
纯干货:Hadoop核心架构HDFS+MapReduce+Hbase+Hive内部机理详解. 通过这一阶段的调研总结,从内部机理的角度详细分析,HDFS.MapReduce.Hbase.H ...
大数据-NoSQL数据库：HBase【基于Zookeeper/HDFS的分布式、高并发、可扩展的NoSQL数据库；支持“十亿行”×“百万列”海量数据的实时随机读写；仅支持单表查询；不擅长数据分析】
HBase适合场景:单表超千万,上亿,且高并发! HBase不适合场景:主要需求是数据分析,比如做报表.数据量规模不大,对实时性要求高! HBase的查询工具有很多,如:Hive.Tez.Impala ...
android fragment传递数据,Android 两个Fragment之间传递数据实例详解
Android 两个Fragment之间如何传递数据 FragmentA启动FragmentB,做一些选择操作后,返回FragmentA,需要把FragmentB里面选择的数据传回来.有什么办法? F ...
BigData：大数据开发的简介、核心知识(linux基础+Java/Python编程语言+Hadoop{HDFS、HBase、Hive}+Docker)、经典场景应用之详细攻略
BigData:大数据开发的简介.核心知识(linux基础+Java/Python编程语言+Hadoop{HDFS.HBase.Hive}+Docker).经典场景应用之详细攻略 BigData:大数 ...
vue中数据之间的传递用propos
子集父集以上是子父集互相传递数据. 传参数的时候是以对象的形式展示的,封装在propos中的. 如果默认值是string或者number类型,那么可以用default表示. 如果默认值是objec ...
java mapreduce 读hbase数据写入hdfs 含maven依赖
mapreduce 读hbase数据写入hdfs java代码如下 import com.google.common.collect.Lists; import java.io.FileInputS ...
Vue 组件之间传值(传递数据)的5种方法 (七)
文章目录一.父传子(父组件向子组件传递数据) 二.子传父(子组件向父组件中传递数据) 三.兄弟组件之间的传值四.总线传值五.通过Vuex这个仓库,进行数据交互(后面会讲) 一.父传子(父组件向子 ...
hbase的备份恢复1,Expor过程，Import过程，统计hbase表行数；hbase备份恢复方式2：使用hdfs备份hbase数据，基于hbase数据进行恢复
25.28 hbase数据备份和恢复以下过程针对线上没有启动新旧集群的情况,所以使用hbase export / import的方式进行数据的备份和迁移 25.28.1 Export过程首先,先进 ...

数据在HDFS和HBASE之间互相传递的过程

数据在HDFS和HBASE之间互相传递的过程相关推荐

最新文章

热门文章