当前位置:首页 > 开发 > 开源软件 > 正文

Hadoop生成HFile直接入库HBase心得

发表于: 2013-12-18   作者:BlackWing   来源:转载   浏览次数:
摘要: 转载请标明出处:http://blackwing.iteye.com/blog/1991380 hbase自带了ImportTsv类,可以直接把tsv格式(官方教材显示,是\t分割各个字段的文本格式)生成HFile,并且使用另外一个类org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles直接把HFile移动到hbase对应的hdfs
转载请标明出处:http://blackwing.iteye.com/blog/1991380


hbase自带了ImportTsv类,可以直接把tsv格式(官方教材显示,是\t分割各个字段的文本格式)生成HFile,并且使用另外一个类org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles直接把HFile移动到hbase对应的hdfs目录。

PS:网上看到一个XD说,直接生成HFile并入库HBase效率不如先生成HFile,再通过LoadIncrementalHFiles移动文件到hbase目录高,这点没有验证,我的做法也是先生成,再move。

官方教材在此:
http://hbase.apache.org/book/ops_mgt.html#importtsv

但ImportTsv功能对我来说不适合,例如文件格式为:
topsid   uid   roler_num   typ        time
10      111111   255         0       1386553377000

ImportTsv导入的命令为:
bin/hbase org.apache.hadoop.hbase.mapreduce.ImportTsv -Dimporttsv.columns=HBASE_ROW_KEY,kq:topsid,kq:uid,kq:roler_num,kq:type -Dimporttsv.bulk.output=hdfs://storefile-outputdir <hdfs-data-inputdir>


它生成的表格式为:
row : 10 
cf  :  kq
qualifier: topsid
value: 10
.....

而我要求的格式是:
row : 10-111111-255
cf  :  kq
qualifier: 0
value: 1


所以还是自己写MR处理数据方便。
Mapper:
/*
 * adminOnOff.log 文件格式:
 * topsid   uid   roler_num   typ   time
 * */
public class HFileImportMapper2 extends
		Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue> {
	protected SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
	protected final String CF_KQ="kq";//考勤
	protected final int ONE=1;
	@Override
	protected void map(LongWritable key, Text value,Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		System.out.println("line : "+line);
		String[] datas = line.split("\\s+");
		// row格式为:yyyyMMdd-sid-uid-role_num-timestamp-typ
		String row = sdf.format(new Date(Long.parseLong(datas[4])))
				+ "-" + datas[0] + "-" + datas[1] + "-" + datas[2]
				+ "-" + datas[4] + "-" + datas[3];
		ImmutableBytesWritable rowkey = new ImmutableBytesWritable(
				Bytes.toBytes(row));
		KeyValue kv = new KeyValue(Bytes.toBytes(row),this.CF_KQ.getBytes(), datas[3].getBytes(),Bytes.toBytes(this.ONE));
		context.write(rowkey, kv);
		}
}


job:
public class GenHFile2 {
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		conf.addResource("myConf.xml");
		String input = conf.get("input");
		String output = conf.get("output");
		String tableName = conf.get("source_table");
		System.out.println("table : "+tableName);
		HTable table;
		try {
			//运行前,删除已存在的中间输出目录
			try {
				FileSystem fs = FileSystem.get(URI.create(output), conf);
				fs.delete(new Path(output),true);
				fs.close();
			} catch (IOException e1) {
				e1.printStackTrace();
			}
			
			table = new HTable(conf,tableName.getBytes());
			Job job = new Job(conf);
			job.setJobName("Generate HFile");
			
			job.setJarByClass(HFileImportMapper2.class);
			job.setInputFormatClass(TextInputFormat.class);
			job.setMapperClass(HFileImportMapper2.class);
			FileInputFormat.setInputPaths(job, input);
			
//job.setReducerClass(KeyValueSortReducer.class);
//job.setMapOutputKeyClass(ImmutableBytesWritable.class);
//job.setMapOutputValueClass(KeyValue.class);
			job.getConfiguration().set("mapred.mapoutput.key.class", "org.apache.hadoop.hbase.io.ImmutableBytesWritable");
			job.getConfiguration().set("mapred.mapoutput.value.class", "org.apache.hadoop.hbase.KeyValue");
			
//job.setOutputFormatClass(HFileOutputFormat.class);
FileOutputFormat.setOutputPath(job, new Path(output));
	//job.setPartitionerClass(SimpleTotalOrderPartitioner.class);
HFileOutputFormat.configureIncrementalLoad(job,table);
			try {
				job.waitForCompletion(true);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}


生成的HFile文件在hdfs的/output目录下,已经根据cf名称建好文件目录:
hdfs://namenode/output/kq/601c5029fb264dc8869a635043c24560

其中:
HFileOutputFormat.configureIncrementalLoad(job,table);

根据其源码知道,会自动为job设置好以下参数:
  public static void configureIncrementalLoad(Job job, HTable table)
  throws IOException {
    Configuration conf = job.getConfiguration();

    job.setOutputKeyClass(ImmutableBytesWritable.class);
    job.setOutputValueClass(KeyValue.class);
    job.setOutputFormatClass(HFileOutputFormat.class);

    // Based on the configured map output class, set the correct reducer to properly
    // sort the incoming values.
    // TODO it would be nice to pick one or the other of these formats.
    if (KeyValue.class.equals(job.getMapOutputValueClass())) {
      job.setReducerClass(KeyValueSortReducer.class);
    } else if (Put.class.equals(job.getMapOutputValueClass())) {
      job.setReducerClass(PutSortReducer.class);
    } else if (Text.class.equals(job.getMapOutputValueClass())) {
      job.setReducerClass(TextSortReducer.class);
    } else {
      LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
    }

    conf.setStrings("io.serializations", conf.get("io.serializations"),
        MutationSerialization.class.getName(), ResultSerialization.class.getName(),
        KeyValueSerialization.class.getName());

    // Use table's region boundaries for TOP split points.
    LOG.info("Looking up current regions for table " + Bytes.toString(table.getTableName()));
    List<ImmutableBytesWritable> startKeys = getRegionStartKeys(table);
    LOG.info("Configuring " + startKeys.size() + " reduce partitions " +
        "to match current region count");
    job.setNumReduceTasks(startKeys.size());

    configurePartitioner(job, startKeys);
    // Set compression algorithms based on column families
    configureCompression(table, conf);
    configureBloomType(table, conf);
    configureBlockSize(table, conf);

    TableMapReduceUtil.addDependencyJars(job);
    TableMapReduceUtil.initCredentials(job);
    LOG.info("Incremental table " + Bytes.toString(table.getTableName()) + " output configured.");
  }


HFileOutputFormat只支持写单个column family,如果有多个cf,则需要写多个job来实现了。

Hadoop生成HFile直接入库HBase心得

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
HFile的整体结构图如下: 整个HFile分四部分: 1.可以被迭代器扫描的部分,如数据块 2.不被迭代器扫描
HFile的整体结构图如下: 整个HFile分四部分: 1.可以被迭代器扫描的部分,如数据块 2.不被迭代器扫描
what is every db holds its storage level,either memory or fs.this is similar to,hbase,it has
HBase put数据时会先将数据写入内存,其内存结构是一个ConcurrentSkipListMap,其Comparator是KVCom
写入数据: public class TestWrit { private static Configuration cfg = new Configuration(); pri
写入数据: public class TestWrit { private static Configuration cfg = new Configuration(); pri
1.HFile详解 HFile文件分为以下六大部分 序号 名称 描述 1 数据块 由多个block(块)组成,每个块的格
1.HFile详解 HFile文件分为以下六大部分 序号 名称 描述 1 数据块 由多个block(块)组成,每个块的格
1. 首先看看HBase中存储的文件内容 执行如下命令添加测试数据: create 'table3', 'colfam1', { SPL
1. 首先看看HBase中存储的文件内容 执行如下命令添加测试数据: create 'table3', 'colfam1', { SPL
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号