上一节我们介绍了使用hbase提供的接口完成其与hdfs之间数据导入导出,但显然这种基于自带接口实现的操作是有局限的,即不能直接将hdfs中的数据导入到hbase。这一节我们将介绍一种更为通用的方法去完成从hdfs导入数据到hbase。
1、前提约束
- 已完成安装hdfs并启动
https://www.jianshu.com/p/b7ae3b51e559
假设hadoop的安装路径为:/root/hadoop-2.5.2,对应的ip为192.168.100.141,域名为hadoop1 - 已完成安装hbase并启动
https://www.jianshu.com/p/90d1713d55ce
假设hbase的安装路径为:/root/hbase-1.2.6,对应的ip为192.168.100.141 - win10本地安装hadoop
假设路径为 C:/hadoop2.7.2
2、操作步骤
- 在idea中创建maven项目,加入以下依赖
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-metastore</artifactId>
<version>2.3.4</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-jdbc</artifactId>
<version>0.14.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.47</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
<version>1.1.1</version>
</dependency>
- 在src/main/java文件夹下创建HdfsToHBase.java
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.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.LongWritable;
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) {
try {
System.setProperty("hadoop.home.dir", "C:/hadoop2.7.2");
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.100.141:2181");
conf.set(TableOutputFormat.OUTPUT_TABLE, "t_user");
Job job = Job.getInstance(conf, HdfsToHBase.class.getSimpleName());
TableMapReduceUtil.addDependencyJars(job);
job.setJarByClass(HdfsToHBase.class);
job.setMapperClass(HdfsToHBaseMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setReducerClass(HdfsToHBaseReducer.class);
FileInputFormat.addInputPath(job, new Path("hdfs://192.168.100.141:9000/t_user"));
job.setOutputFormatClass(TableOutputFormat.class);
job.waitForCompletion(true);
}
catch (Exception e)
{
e.printStackTrace();
}
}
public static class HdfsToHBaseMapper extends Mapper<LongWritable, Text, Text, Text> {
private Text outKey = new Text();
private Text outValue = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] splits = value.toString().split("\\s");
outKey.set(splits[0]);
outValue.set(splits[1] + "," + splits[2]);
context.write(outKey, outValue);
}
}
public static class HdfsToHBaseReducer extends TableReducer<Text, Text, NullWritable> {
@Override
protected void reduce(Text k2, Iterable<Text> v2s, Context context) throws IOException, InterruptedException {
Put put = new Put(k2.getBytes());
for (Text v2 : v2s) {
String[] splis = v2.toString().split(",");
if (splis[0] != null && !"NULL".equals(splis[0])) {
put.add("f1".getBytes(), "name".getBytes(), splis[0].getBytes());
}
if (splis[1] != null && !"NULL".equals(splis[1])) {
put.add("f1".getBytes(), "age".getBytes(), splis[1].getBytes());
}
}
context.write(NullWritable.get(), put);
}
}
}
- 修改C:\Windows\System32\drivers\etc\hosts,加入以下内容:
192.168.100.141 hadoop1 - 在linux的hdfs中上传以下内容到/t_user
1 ali 10
2 xiaoli 20
3 zhangli 30
4 ali 11
- 在linux的hbase中创建表格
/root/hbase-1.2.6/hbase shell
create 't_user','f1'
- 执行HdfsToHBase.java
- 在linux的hbase中查看t_user
/root/hbase-1.2.6/hbase shell
scan 't_user'
hbase中t_user的内容
以上就是使用mr自定义完成hdfs数据导入到hbase。
网友评论