HBase和HDFS数据互导程序

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
云原生网关 MSE Higress,422元/月
简介: 下面说说JAVA API 提供的这些类的功能和他们之间有什么样的联系。1.HBaseConfiguration关系:org.apache.hadoop.hbase.HBaseConfiguration作用:通过此类可以对HBase进行配置用法实例: Configuration config = HBaseConfiguration.create();说明: HBaseConf




下面说说JAVA API 提供的这些类的功能和他们之间有什么样的联系。


1.HBaseConfiguration

关系:org.apache.hadoop.hbase.HBaseConfiguration

作用:通过此类可以对HBase进行配置

用法实例: Configuration config = HBaseConfiguration.create();

说明: HBaseConfiguration.create() 默认会从classpath 中查找 hbase-site.xml 中的配置信息,初始化 Configuration

2.HBaseAdmin 类

关系:org.apache.hadoop.hbase.client.HBaseAdmin

作用:提供接口关系HBase 数据库中的表信息

用法:HBaseAdmin admin = new HBaseAdmin(config);

3.Descriptor类

关系:org.apache.hadoop.hbase.HTableDescriptor

作用:HTableDescriptor 类包含了表的名字以及表的列族信息

用法:HTableDescriptor htd =new HTableDescriptor(tablename);

             构造一个表描述符指定TableName对象。

             Htd.addFamily(new HColumnDescriptor(“myFamily”));

             将列家族给定的描述符

4.HTable

关系:org.apache.hadoop.hbase.client.HTable

作用:HTable HBase 的表通信

用法:HTable tab = new HTable(config,Bytes.toBytes(tablename));

           ResultScanner sc = tab.getScanner(Bytes.toBytes(“familyName”));

说明:获取表内列族 familyNme 的所有数据。

5.Put

关系:org.apache.hadoop.hbase.client.Put

作用:获取单个行的数据

用法:HTable table = new HTable(config,Bytes.toBytes(tablename));

           Put put = new Put(row);

           p.add(family,qualifier,value);

说明:向表 tablename 添加 “family,qualifier,value”指定的值。

6.Get

关系:org.apache.hadoop.hbase.client.Get

作用:获取单个行的数据

用法:HTable table = new HTable(config,Bytes.toBytes(tablename));

           Get get = new Get(Bytes.toBytes(row));

           Result result = table.get(get);

说明:获取 tablename 表中 row 行的对应数据

7.ResultScanner

关系:Interface

作用:获取值的接口

用法:ResultScanner scanner = table.getScanner(Bytes.toBytes(family));

           For(Result rowResult : scanner){

                   Bytes[] str = rowResult.getValue(family,column);

}

说明:循环获取行中列值。



例1 HBase之读取HDFS数据写入HBase

package org.hadoop.hbase;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
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.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.GenericOptionsParser;
public class WordCountHbaseWriter {
 public static class WordCountHbaseMapper extends
   Mapper<Object, Text, Text, IntWritable> {
  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();
  public void map(Object key, Text value, Context context)
    throws IOException, InterruptedException {
   StringTokenizer itr = new StringTokenizer(value.toString());
   while (itr.hasMoreTokens()) {
    word.set(itr.nextToken());
    context.write(word, one);// 输出<key,value>为<word,one>
   }
  }
 }
 public static class WordCountHbaseReducer extends
   TableReducer<Text, IntWritable, ImmutableBytesWritable> {
  public void reduce(Text key, Iterable<IntWritable> values,
    Context context) throws IOException, InterruptedException {
   int sum = 0;
   for (IntWritable val : values) {// 遍历求和
    sum += val.get();
   }
   Put put = new Put(key.getBytes());//put实例化,每一个词存一行
   //列族为content,列修饰符为count,列值为数目
   put.add(Bytes.toBytes("content"), Bytes.toBytes("count"), Bytes.toBytes(String.valueOf(sum)));
   context.write(new ImmutableBytesWritable(key.getBytes()), put);// 输出求和后的<key,value>
  }
 }
 
 public static void main(String[] args){
  String tablename = "wordcount";
  Configuration conf = HBaseConfiguration.create();
    conf.set("hbase.zookeeper.quorum", "192.168.1.139");
    conf.set("hbase.zookeeper.property.clientPort", "2191");
  HBaseAdmin admin = null;
  try {
   admin = new HBaseAdmin(conf);
   if(admin.tableExists(tablename)){
    System.out.println("table exists!recreating.......");
    admin.disableTable(tablename);
    admin.deleteTable(tablename);
   }
   HTableDescriptor htd = new HTableDescriptor(tablename);
   HColumnDescriptor tcd = new HColumnDescriptor("content");
   htd.addFamily(tcd);//创建列族
   admin.createTable(htd);//创建表
   String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
      if (otherArgs.length != 1) {
        System.err.println("Usage: WordCountHbaseWriter <in>");
        System.exit(2);
      }
      Job job = new Job(conf, "WordCountHbaseWriter");
  job.setNumReduceTasks(2);
      job.setJarByClass(WordCountHbaseWriter.class);
   //使用WordCountHbaseMapper类完成Map过程;
      job.setMapperClass(WordCountHbaseMapper.class);
      TableMapReduceUtil.initTableReducerJob(tablename, WordCountHbaseReducer.class, job);
      //设置任务数据的输入路径;
      FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
   //设置了Map过程的输出类型,其中设置key的输出类型为Text;
      job.setOutputKeyClass(Text.class);
   //设置了Map过程的输出类型,其中设置value的输出类型为IntWritable;
      job.setOutputValueClass(IntWritable.class);
   //调用job.waitForCompletion(true) 执行任务,执行成功后退出;
      System.exit(job.waitForCompletion(true) ? 0 : 1);
  } catch (Exception e) {
   e.printStackTrace();
  } finally{
   if(admin!=null)
    try {
     admin.close();
    } catch (IOException e) {
     e.printStackTrace();
    }
  }
  
 }
}


例2 HBase之读取HBase数据写入HDFS

package org.hadoop.hbase;
import java.io.IOException;
import java.util.Map.Entry;
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.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class WordCountHbaseReader {
 
 public static class WordCountHbaseReaderMapper extends 
    TableMapper<Text,Text>{
    @Override
    protected void map(ImmutableBytesWritable key,Result value,Context context)
            throws IOException, InterruptedException {
        StringBuffer sb = new StringBuffer("");
        for(Entry<byte[],byte[]> entry:value.getFamilyMap("content".getBytes()).entrySet()){
            String str =  new String(entry.getValue());
            //将字节数组转换为String类型
            if(str != null){
                sb.append(new String(entry.getKey()));
                sb.append(":");
                sb.append(str);
            }
            context.write(new Text(key.get()), new Text(new String(sb)));
        }
    }
}
 public static class WordCountHbaseReaderReduce extends Reducer<Text,Text,Text,Text>{
     private Text result = new Text();
     @Override
     protected void reduce(Text key, Iterable<Text> values,Context context)
             throws IOException, InterruptedException {
         for(Text val:values){
             result.set(val);
             context.write(key, result);
         }
     }
 }
 
 public static void main(String[] args) throws Exception {
     String tablename = "wordcount";
     Configuration conf = HBaseConfiguration.create();
     conf.set("hbase.zookeeper.quorum", "192.168.1.139");
     conf.set("hbase.zookeeper.property.clientPort", "2191");
     
     String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
     if (otherArgs.length != 1) {
       System.err.println("Usage: WordCountHbaseReader <out>");
       System.exit(2);
     }
     Job job = new Job(conf, "WordCountHbaseReader");
     job.setJarByClass(WordCountHbaseReader.class);
     //设置任务数据的输出路径;
     FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
     job.setReducerClass(WordCountHbaseReaderReduce.class);
     Scan scan = new Scan();
     TableMapReduceUtil.initTableMapperJob(tablename,scan,WordCountHbaseReaderMapper.class, Text.class, Text.class, job);
     //调用job.waitForCompletion(true) 执行任务,执行成功后退出;
     System.exit(job.waitForCompletion(true) ? 0 : 1);

 }
}


程序中用到hadoop的相关JAR包(如下图)及hbase所有jar包

wKiom1VSw0bjkI7QAABrwzSlqdo801.jpg

如果上面的API还不能满足你的要求,可以到下面这个网站里面Hbase全部API介绍

http://www.yiibai.com/hbase/

 

本文出自 “点滴积累” 博客,请务必保留此出处http://tianxingzhe.blog.51cto.com/3390077/1650856

相关实践学习
lindorm多模间数据无缝流转
展现了Lindorm多模融合能力——用kafka API写入,无缝流转在各引擎内进行数据存储和计算的实验。
云数据库HBase版使用教程
&nbsp; 相关的阿里云产品:云数据库 HBase 版 面向大数据领域的一站式NoSQL服务,100%兼容开源HBase并深度扩展,支持海量数据下的实时存储、高并发吞吐、轻SQL分析、全文检索、时序时空查询等能力,是风控、推荐、广告、物联网、车联网、Feeds流、数据大屏等场景首选数据库,是为淘宝、支付宝、菜鸟等众多阿里核心业务提供关键支撑的数据库。 了解产品详情:&nbsp;https://cn.aliyun.com/product/hbase &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
相关文章
|
2月前
|
存储 分布式数据库 数据库
Hbase学习二:Hbase数据特点和架构特点
Hbase学习二:Hbase数据特点和架构特点
47 0
|
21天前
|
SQL 存储 分布式计算
HDFS数据(跨集群)迁移
HDFS数据(跨集群)迁移
|
2月前
|
分布式计算 Hadoop
|
1月前
|
存储 分布式计算 分布式数据库
《HBase MapReduce之旅:我的学习笔记与心得》——跟随我的步伐,一同探索HBase世界,揭开MapReduce的神秘面纱,分享那些挑战与收获,让你在数据的海洋里畅游无阻!
【8月更文挑战第17天】HBase是Apache顶级项目,作为Bigtable的开源版,它是一个非关系型、分布式数据库,具备高可扩展性和性能。结合HDFS存储和MapReduce计算框架,以及Zookeeper协同服务,HBase支持海量数据高效管理。MapReduce通过将任务拆解并在集群上并行执行,极大提升处理速度。学习HBase MapReduce涉及理解其数据模型、编程模型及应用实践,虽然充满挑战,但收获颇丰,对职业发展大有裨益。
29 0
|
2月前
|
缓存 监控 Shell
使用 HBase Shell 进行数据的实时监控和备份
使用 HBase Shell 进行数据的实时监控和备份
|
2月前
|
Shell 分布式数据库 Hbase
使用 HBase Shell 进行数据的批量导入和导出
使用 HBase Shell 进行数据的批量导入和导出
330 6
|
2月前
|
分布式计算 Hadoop 关系型数据库
实时计算 Flink版操作报错合集之Hadoop在将文件写入HDFS时,无法在所有指定的数据节点上进行复制,该如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2月前
|
存储 Java 分布式数据库
HBase构建图片视频数据的统一存储检索
HBase构建图片视频数据的统一存储检索
|
3月前
|
存储 分布式计算 Hadoop
Hadoop的HDFS数据均衡
【6月更文挑战第13天】
143 3
|
3月前
|
Java 大数据 API
【大数据】HDFS、HBase操作教程(含指令和JAVA API)
【大数据】HDFS、HBase操作教程(含指令和JAVA API)
108 0
【大数据】HDFS、HBase操作教程(含指令和JAVA API)

热门文章

最新文章