美文网首页
05 hdfs的数据导出到hbase

05 hdfs的数据导出到hbase

作者: 张力的程序园 | 来源:发表于2020-06-17 19:38 被阅读0次

    上一节我们介绍了使用hbase提供的接口完成其与hdfs之间数据导入导出,但显然这种基于自带接口实现的操作是有局限的,即不能直接将hdfs中的数据导入到hbase。这一节我们将介绍一种更为通用的方法去完成从hdfs导入数据到hbase。

    1、前提约束

    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。

    相关文章

      网友评论

          本文标题:05 hdfs的数据导出到hbase

          本文链接:https://www.haomeiwen.com/subject/awldxktx.html