专栏首页数据之美Hadoop Mapper 阶段将数据直接从 HDFS 导入 Hbase

Hadoop Mapper 阶段将数据直接从 HDFS 导入 Hbase

数据源格式如下:

20130512	1	-1	-1	13802	1	2013-05-12 07:26:22	
20130512	1	-1	-1	13802	1	2013-05-12 11:18:24

我们期待的结果是数据直接从 hdfs 读取后 写入 hbase,没有 reduce 阶段,

代码如下:

package WebsiteAnalysis;

import java.io.IOException;

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 Map2Hdfs {
	public static final String NAME = "ImportFromFile";

	public enum Counters {
		LINES
	}

	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 {
			String column = context.getConfiguration().get("conf.column");
			byte[][] colkey = KeyValue.parseColumn(Bytes.toBytes(column));
			family = colkey[0];
			if (colkey.length > 1) {
				qualifier = colkey[1];
			}
		}

		@Override
		public void map(LongWritable offset, Text line, Context context) throws IOException {
			try {
				String[] lineArr = line.toString().split("\t");
				Put put = new Put(Bytes.toBytes(offset + ""));
				put.add(family, Bytes.toBytes("time"), Bytes.toBytes(lineArr[lineArr.length - 1]));
				context.write(new ImmutableBytesWritable(Bytes.toBytes(offset + "")), put);
				context.getCounter(Counters.LINES).increment(1);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		conf.set("conf.column", "cf");
		String inputPath = "/dsap/middata/lj/ooxx/pv";
		Job job = new Job(conf, "TestMap2Hdfs");

		job.setJarByClass(Map2Hdfs.class);
		job.setMapperClass(ImportMapper.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "TestMap2Hdfs");
		job.setOutputKeyClass(ImmutableBytesWritable.class);
		job.setOutputValueClass(Writable.class);
		job.setNumReduceTasks(0);
		FileInputFormat.addInputPath(job, new Path(inputPath + "/" + otherArgs[0]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

REF:

http://stackoverflow.com/questions/11061854/hadoop-writing-to-hbase-directly-from-the-mapper

http://blog.sina.com.cn/s/blog_62a9902f0101904h.html  新建表的方式写入

hbase-hdfs MapReduce 数据读写总结

http://blog.pureisle.net/archives/1938.html  hbase hdfs MR 读写的几种情况总结

http://blog.csdn.net/kirayuan/article/details/7001278  hbase表拷贝样例代码

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • BloomFilter 简介及在 Hadoop reduce side join 中的应用

    1、BloomFilter能解决什么问题?      以少量的内存空间判断一个元素是否属于这个集合, 代价是有一定的错误率 2、工作原理 ...

    用户1177713
  • Pig、Hive、MapReduce 解决分组 Top K 问题

    问题: 有如下数据文件 city.txt (id, city, value) cat city.txt  1 wh 500 2 bj 600 3 wh 1...

    用户1177713
  • 自定义 hadoop MapReduce InputFormat 切分输入文件

    在上一篇中,我们实现了按 cookieId 和 time 进行二次排序,现在又有新问题:假如我需要按 cookieId 和 cookieId&time 的组合...

    用户1177713
  • HBase快速入门系列(7) | 官方HBase-MapReduce与自定义

    不温卜火
  • HBase 2.0 协处理器实现 ES 数据同步

    在正式进行讲述实现之前,我觉得有必要说一下出发点。团队期初数据都是基于 HBase+Phoenix 这样架构进行持久化。随着业务的复杂性增加,对部分表的查询效率...

    迹_Jason
  • 大数据应用之HBase数据插入性能优化实测教程

    大家在使用HBase的过程中,总是面临性能优化的问题,本文从HBase客户端参数设置的角度,研究HBase客户端数据批量插入性能优化的问题。事实胜于雄辩,数据比...

    数据饕餮
  • Mapreduce和HBase新版本整合之WordCount计数案例

    先计数单词数量存到hdfs文件上,这个是以前的就做过的 package com.my.myhnase.mapreduce; import java.io.IO...

    汤高
  • HBase整合MapReduce之建立HBase索引

    HBase索引主要用于提高Hbase中表数据的访问速度,有效的避免了全表扫描,HBase中的表根据行健被分成了多个Regions,通常一个region的一行都会...

    汤高
  • HBase Java API 03:HBase与MapReduce整合

    编写MapReduce程序,把"student"表中"info"列族下的"name"那一列抽取出来,存入新HBase表"student_extract"中,要求...

    CoderJed
  • HBase新的客户端接口

    最近学习接触HBase的东西,看了《Habase in Action》,但里面关于HBase接口都是过时的接口,以下为HBase新的客户端接口:

    chaplinthink

扫码关注云+社区

领取腾讯云代金券