用eclipse编程实现对hbase中表的数据的增加
- 用eclipse编程实现对hbase中表的数据的增加
先启动hadoop，之后启动hbase
在其中添加一个study的表，查看study的内容
显示无内容
创建一个新的java project，命名为hbase_study，创建一个com.study.hbase的包，在里面添加了EduAppend的类
程序如下
package com.study.hbase;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.util.Bytes;
public class EduAppend {
public static void main(String[] args) throws IOException {
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.139.134");
conf.set("hbase.zookeeper.property.clientPort", "2181");
HConnection conn = HConnectionManager.createConnection(conf);
HTableInterface table = conn.getTable("study");
Append append = new Append(Bytes.toBytes("r1"));
append.add(Bytes.toBytes("d"),Bytes.toBytes("a"),Bytes.toBytes("a"));
table.append(append);
table.close();
conn.close();
}
}再次查看study表中的内容
数据成功插入
追加数据，修改代码为
package com.study.hbase;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.util.Bytes;
public class EduAppend {
public static void main(String[] args) throws IOException {
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.139.134");
conf.set("hbase.zookeeper.property.clientPort", "2181");
HConnection conn = HConnectionManager.createConnection(conf);
HTableInterface table = conn.getTable("study");
Append append = new Append(Bytes.toBytes("r1"));
append.add(Bytes.toBytes("d"),Bytes.toBytes("a"),Bytes.toBytes("bcd"));
table.append(append);
table.close();
conn.close();
}
}查看study表中数据，变为
数据成功追加
- 内置过滤器的使用
HBase为筛选数据提供了一组过滤器，通过这个过滤器可以在HBase中数据的多个维度（行、列、数据版本）上进行对数据的筛选操作，也就是说过滤器最终能够筛选的数据能够细化到具体的一个存储单元格上（由行键、列名、时间戳定位）。通常来说，通过行键、值来筛选数据的应用场景较多。需要说明的是，过滤器会极大地影响查询效率。所以，在数据量较大的数据表中，应尽量避免使用过滤器。
下面介绍一些常用的HBase内置过滤器的用法：
1、RowFilter：筛选出匹配的所有的行。使用BinaryComparator可以筛选出具有某个行键的行，或者通过改变比较运算符来筛选出符合某一条件的多条数据。
2、PrefixFilter：筛选出具有特定前缀的行键的数据。这个过滤器所实现的功能其实也可以由RowFilter结合RegexComparator来实现。
3、KeyOnlyFilter：这个过滤器唯一的功能就是只返回每行的行键，值全部为空，这对于只关注于行键的应用场景来说非常合适，这样忽略掉其值就可以减少传递到客户端的数据量，能起到一定的优化作用。
4、RandomRowFilter：按照一定的几率（<=0会过滤掉所有的行，>=1会包含所有的行）来返回随机的结果集，对于同样的数据集，多次使用同一个RandomRowFilter会返回不同的结果集，对于需要随机抽取一部分数据的应用场景，可以使用此过滤器。
5、InclusiveStopFilter：扫描的时候，我们可以设置一个开始行键和一个终止行键，默认情况下，这个行键的返回是前闭后开区间，即包含起始行，但不包含终止行。如果我们想要同时包含起始行和终止行，那么可以使用此过滤器。
6、FirstKeyOnlyFilter：如果想要返回的结果集中只包含第一列的数据，那么这个过滤器能够满足要求。它在找到每行的第一列之后会停止扫描，从而使扫描的性能也得到了一定的提升。
7、ColumnPrefixFilter：它按照列名的前缀来筛选单元格，如果我们想要对返回的列的前缀加以限制的话，可以使用这个过滤器。
8、ValueFilter：按照具体的值来筛选单元格的过滤器，这会把一行中值不能满足的单元格过滤掉。
9、ColumnCountGetFilter：这个过滤器在遇到一行的列数超过我们所设置的限制值的时候，结束扫描操作。
10、SingleColumnValueFilter：用一列的值决定这一行的数据是否被过滤，可对它的对象调用setFilterIfMissing方法，默认的参数是false。
11、SingleColumnValueExcludeFilter：这个过滤器与第10种过滤器唯一的区别就是，作为筛选条件的列，其行不会包含在返回的结果中。
12、SkipFilter：这是一种附加过滤器，其与ValueFilter结合使用，如果发现一行中的某一列不符合条件，那么整行就会被过滤掉。
13、WhileMatchFilter：使用这个过滤器，当遇到不符合设定条件的数据的时候，整个扫描结束。
14. FilterList：可以用于综合使用多个过滤器。其有两种关系： Operator.MUST_PASS_ONE表示关系AND，Operator.MUST_PASS_ALL表示关系OR，并且FilterList可以嵌套使用，使得我们能够表达更多的需求。
- HBase与Mapreduce集成
HBase与mapreduce的集成也就是使mapreduce作业以HBase表作为输入，或者作为输出，也或者作为mapreduce作业之间共享数据的介质。
读取存储在hdfs上的txt文本数据，简单地以json字符串的形式 存储 到HBase表中。
代码如下
import java.io.IOException;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
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 HdfsToHBase
{
private static final Log LOG = LogFactory.getLog(HdfsToHBase.class);
public static final String NAME = "ImportFromFile";
public enum Counters { LINES }
/**
* Map类
*
*/
static class ImportMapper
extends Mapper<LongWritable, Text, ImmutableBytesWritable, Writable>
{
private byte[] family = null;
private byte[] qualifier = null;
@Override
protected void setup(Context context) throws IOException, InterruptedException
{
//获取通过Configuration传过来的列名
String columns = context.getConfiguration().get("conf.column");
//解析出列族和列的名称
byte[][] columnsBytes = KeyValue.parseColumn(Bytes.toBytes(columns));
family = columnsBytes[0];
qualifier = columnsBytes[1];
LOG.info("family:"+family.toString()+"qualifiers:"+qualifier);
}
@Override
public void map(LongWritable offset, Text line, Context context) throws IOException
{
try
{
String lineStr = line.toString();
byte[] rowkey = DigestUtils.md5(lineStr);
//构造Put对象
Put put = new Put(rowkey);
put.add(family, qualifier, Bytes.toBytes(lineStr));
//发射Put对象
context.write(new ImmutableBytesWritable(rowkey), put);
context.getCounter(Counters.LINES).increment(1);
}catch(Exception e)
{
e.printStackTrace();
}
}
}
/**
* 将命令行参数解析为HBase的CommandLine对象
* @param args
* @return
* @throws ParseException
*/
private static CommandLine parseArgs(String[] args) throws ParseException
{
Options options = new Options();
Option o = new Option("t", "table", true, "table to import into (must exist)");
o.setArgName("table-name");
o.setRequired(true);
options.addOption(o);
o = new Option("c", "column", true, "column to store row data into (must exist)");
o.setArgName("family:qualifier");
o.setRequired(true);
options.addOption(o);
o = new Option("i", "input", true, "the directory or file to read from");
o.setArgName("path-in-HDFS");
o.setRequired(true);
options.addOption(o);
CommandLineParser parser = new PosixParser();
CommandLine cmd = null;
try
{
cmd = parser.parse(options, args);
} catch (Exception e) {
System.err.println("ERROR: " + e.getMessage() + "\n");
HelpFormatter formatter = new HelpFormatter();
formatter.printHelp(NAME + " ", options, true);
System.exit(-1);
}
return cmd;
}
/**
* 主函数
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception
{
//将输入参数解析为CommandLine对象
Configuration conf = HBaseConfiguration.create();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
CommandLine cmd = parseArgs(otherArgs);
//取出各项参数
String tableName = cmd.getOptionValue("t");
String inputFileName = cmd.getOptionValue("i");
String columnName = cmd.getOptionValue("c");
conf.set("conf.column", columnName);
Job job = new Job(conf, "Import from file " + inputFileName + " into table " + tableName);
job.setJarByClass(HdfsToHBase.class);
//设置map和reduce类
job.setMapperClass(ImportMapper.class);
job.setNumReduceTasks(0);
//设置map阶段输出的键值对类型
job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(Writable.class);
//设置job输入输出格式
job.setOutputFormatClass(TableOutputFormat.class);
job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName);
//设置输入输出路径
FileInputFormat.addInputPath(job, new Path(inputFileName));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
在伪分布式模式和完全分布式模式下的HBase是架构在HDFS之上的，因此完全可以将MapReduce编程框架和HBase结合起来使用。也就是说，将HBase作为底层存储结构，MapReduce调用HBase进行特殊的处理，这样能够充分结合HBase分布式大型数据库和MapReduce并行计算的优点。
HBase实现了TableInputFormatBase类，该类提供了对表数据的大部分操作，其子类TableInputFormat则提供了完整的实现，用于处理表数据并生成键值对。TableInputFormat类将数据表按照Region分割成split，即有多少个Regions就有多个splits，然后将Region按行键分成<key,value>对，key值对应与行键，value值为该行所包含的数据。
HBase实现了MapReduce计算框架对应的TableMapper类和TableReducer类。其中，TableMapper类并没有具体的功能，只是将输入的<key,value>对的类型分别限定为Result和ImmutableBytesWritable。IdentityTableMapper类和IdentityTableReducer类则是上述两个类的具体实现，其和Mapper类和Reducer类一样，只是简单地将<key,value>对输出到下一个阶段。
HBase实现的TableOutputFormat将输出的<key,value>对写到指定的HBase表中，该类不会对WAL（Write-Ahead Log）进行操作，即如果服务器发生故障将面临丢失数据的风险。可以使用MultipleTableOutputFormat类解决这个问题，该类可以对是否写入WAL进行设置。
- hbase-bulkload学习
HBase本身提供了很多种数据导入的方式，通常有两种常用方式：
1、使用HBase提供的TableOutputFormat，原理是通过一个Mapreduce作业将数据导入HBase
2、另一种方式就是使用HBase原生Client API
这两种方式由于需要频繁的与数据所存储的RegionServer通信，一次性入库大量数据时，特别占用资源，所以都不是最有效的。了解过HBase底层原理的应该都知道，HBase在HDFS中是以HFile文件结构存储的，一个比较高效便捷的方法就是使用 “Bulk Loading”方法直接生成HFile，即HBase提供的HFileOutputFormat类。
Bulk Load处理由两个主要步骤组成
1、准备数据文件
Bulk Load的第一步，会运行一个Mapreduce作业，其中使用到了HFileOutputFormat输出HBase数据文件：StoreFile。HFileOutputFormat的作用在于使得输出的HFile文件可以适应单个region，使用TotalOrderPartitioner类将map输出结果分区到各个不同的key区间中，每个key区间都对应着HBase表的region。
2、导入HBase表
第二步使用completebulkload工具将第一步的结果文件依次交给负责文件对应region的RegionServer，并将文件move到region在HDFS上的存储目录中，一旦完成，将数据开放给clients。
如果在bulk load准备导入或在准备导入与完成导入的临界点上发现region的边界已经改变，completebulkload工具会自动split数据文件到新的边界上，但是这个过程并不是最佳实践，所以用户在使用时需要最小化准备导入与导入集群间的延时，特别是当其他client在同时使用其他工具向同一张表导入数据。
生成HFile程序说明：
1、最终输出结果，无论是map还是reduce，输出部分key和value的类型必须是： < ImmutableBytesWritable, KeyValue>或者< ImmutableBytesWritable, Put>。
2、最终输出部分，Value类型是KeyValue 或Put，对应的Sorter分别是KeyValueSortReducer或PutSortReducer。
3、MR例子中job.setOutputFormatClass(HFileOutputFormat.class); HFileOutputFormat只适合一次对单列族组织成HFile文件。
4、MR例子中HFileOutputFormat.configureIncrementalLoad(job, table);自动对job进行配置。SimpleTotalOrderPartitioner是需要先对key进行整体排序，然后划分到每个reduce中，保证每一个reducer中的的key最小最大值区间范围，是不会有交集的。因为入库到HBase的时候，作为一个整体的Region，key是绝对有序的。
5、MR例子中最后生成HFile存储在HDFS上，输出路径下的子目录是各个列族。如果对HFile进行入库HBase，相当于move HFile到HBase的Region中，HFile子目录的列族内容没有了。
示例：
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FsShell;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
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.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
public class BulkLoadJob {
static Logger logger = LoggerFactory.getLogger(BulkLoadJob.class);
public static class BulkLoadMap extends
Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] valueStrSplit = value.toString().split("\t");
String hkey = valueStrSplit[0];
String family = valueStrSplit[1].split(":")[0];
String column = valueStrSplit[1].split(":")[1];
String hvalue = valueStrSplit[2];
final byte[] rowKey = Bytes.toBytes(hkey);
final ImmutableBytesWritable HKey = new ImmutableBytesWritable(rowKey);
Put HPut = new Put(rowKey);
byte[] cell = Bytes.toBytes(hvalue);
HPut.add(Bytes.toBytes(family), Bytes.toBytes(column), cell);
context.write(HKey, HPut);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
String inputPath = args[0];
String outputPath = args[1];
HTable hTable = null;
try {
Job job = Job.getInstance(conf, "ExampleRead");
job.setJarByClass(BulkLoadJob.class);
job.setMapperClass(BulkLoadJob.BulkLoadMap.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);
// speculation
job.setSpeculativeExecution(false);
job.setReduceSpeculativeExecution(false);
// in/out format
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(HFileOutputFormat2.class);
FileInputFormat.setInputPaths(job, inputPath);
FileOutputFormat.setOutputPath(job, new Path(outputPath));
hTable = new HTable(conf, args[2]);
HFileOutputFormat2.configureIncrementalLoad(job, hTable);
if (job.waitForCompletion(true)) {
FsShell shell = new FsShell(conf);
try {
shell.run(new String[]{"-chmod", "-R", "777", args[1]});
} catch (Exception e) {
logger.error("Couldnt change the file permissions ", e);
throw new IOException(e);
}
//加载到hbase表
LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
loader.doBulkLoad(new Path(outputPath), hTable);
} else {
logger.error("loading failed.");
System.exit(1);
}
} catch (IllegalArgumentException e) {
e.printStackTrace();
} finally {
if (hTable != null) {
hTable.close();
}
}
}
}