hadoop hbase 项目

た入场券 2022-03-26 14:00 250阅读 0赞

**1，bulkloading 批量导入工具**

**1.1，环境准备**

1）新建文本数据  music1.txt，music2.txt

music1.txt：

编号（rowkey）    歌名（song）   歌手（singer）   歌手性别（gender）     节奏 （ryghme）      客户端（terminal）

1\_song1\_2016      song1                 singer1                  man                               slow       ios

2\_song1\_2017      song2                 singer2                  man                               slow       pc

2）新建表 musicrecored 表（播放记录表）

列簇只有一个 info

3）新建表 namelist 表（歌曲播放统计表）

列簇只有一个 details

**1.2，hbase 集成到 hadoop 中（集群每一个节点都需要配置）**

1）将 hbase 核心配置文件 hbase-site.xml 添加到 hadoop主目录/etc/hadoop/下

2）将 hbase 的库（hbase主目录下lib目录）配置到 hadoop 的环境（hadoop-env.sh）中

export HADOOP\_CLASSPATH=$HADOOP\_CLASSPATH:/home/hduser/hbase-1.2.0/lib/\*

3）利用新建的 musicrecored 表测试集成是否成功

在 hbase 主目录/lib/中有 hbase-server\*\*\*.jar 内有 hbase 提供的工具类 rowcounter 主要用于测试利用 mapreduce 统计 hbase 表中记录总数，命令如下：

**$>hadoop  jar  hbase主目录/lib/hbase-server\*.jar   rowcounter   musicrecord (表名)**

**       1.3，概念**

**             **bulkloading 批量导入分成两部完成的：

第一部分将需要导入的文本转化为 HFILE（分布式文件）格式的文件；

第二部分将 HFILE 文件关联到 hbase 的表

**       1.4，导入数据过程**

**             **1）hdfs 上文本数据生成 HFILE 数据文件（importtsv）

语法： $>hadoop  jar  hbase-server\*.jar  importtsv  -D  importtsv.bulk.output=HFILE位置  -D  importtsv.columns=HBASE\_ROW\_KEY,info:song,info:singer,info:gender,info:ryghme,info:terminal  表名  hdfs输入文件路径

样例：$>hadoop  jar  hbase-server\*.jar  importtsv  -D  importtsv.bulk.output=/output/m1  -D  importtsv.columns=HBASE\_ROW\_KEY,info:song,info:singer,info:gender,info:ryghme,info:terminal  musicrecord  /input/music.txt

2）将生成的 HFILE 文件与 hbase 库表关联（completebulkload）

语法： $>hadoop  jar  hbase-server\*.jar  completebulkload  HFILE位置  表名

样例：$>hadoop  jar  hbase-server\*.jar  completebulkload  /output/m1/  musicrecord

**2，java 使用 hbase 完成**

package cn.kgc1803;
    
    import java.io.IOException;
    import java.util.StringTokenizer;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.hbase.Cell;
    import org.apache.hadoop.hbase.CellUtil;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.client.Result;
    import org.apache.hadoop.hbase.client.Scan;
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
    import org.apache.hadoop.hbase.mapreduce.TableMapper;
    import org.apache.hadoop.hbase.mapreduce.TableReducer;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class TopMusic {
        /**
        * TableMapper 输入键是行键，类型是 Immutable
         *                  输入值是查询结果集，类型键 Result
         *  统计歌曲名   播放次数
         */
    	public static class TopmusicMapper extends TableMapper<Text, IntWritable>{
    		//每一行进入一次 map 方法
    		public void map(ImmutableBytesWritable rowKey,Result rs,Context context) throws IOException, InterruptedException{
    			//输出结果  歌曲名
    			Text keyOut = null;
    			IntWritable valueOut = new IntWritable(1);
    			for(Cell c:rs.listCells()){
    				//从 cell 中获取行键
    				keyOut = new Text(Bytes.toString(CellUtil.cloneValue(c)));
    				context.write(keyOut, valueOut);
    			}
    		}
    	}
    	
    	//reduce（输出到 txt )
    	public static class TopMusicReducel extends Reducer<Text, IntWritable, Text, IntWritable>{
    		public void reduce(Text keyIn,Iterable<IntWritable> valuesIn,Context context) throws IOException, InterruptedException{
    			Text keyOut = keyIn;
    			IntWritable valueOut = new IntWritable();
    			int sum = 0;
    			for(IntWritable val:valuesIn){
    				sum+=val.get();
    			}
    			valueOut.set(sum);
    			context.write(keyOut, valueOut);
    		}
    	}
    	
    	//读 file
    	public static class TopMusicMapper2 extends Mapper<Object, Text, Text, IntWritable>{
    		public void map(Object keyIn,Text valueIn,Context context) throws IOException, InterruptedException{
    			Text keyOut = null;
    			IntWritable valueOut = null;
    			StringTokenizer token = new StringTokenizer(valueIn.toString());
    			keyOut = new Text(token.nextToken());
    			valueOut = new IntWritable(Integer.parseInt(token.nextToken()));
    			context.write(keyOut, valueOut);
    		}
    	}
    	
    	/**
    	 * reduce 写到数据库
    	 * TableReducer 三个泛型（1：输入键类型，2：输入值类型，3：数据的行键也就是输出键类型）
    	 * TableReducer 继承列 Reducer，并且限制列第四个参数，只能是 muna 类型（PUT，DELETE）的
    	 */
    	public static class TopMusicReduce2 extends TableReducer<Text, IntWritable, Text>{
    		public void reduce(Text keyIn,Iterable<IntWritable> valuesIn,Context context) throws IOException, InterruptedException{
    			for(IntWritable num:valuesIn){
    				Put p = new Put(Bytes.toBytes(keyIn.toString()));
    				p.addColumn(Bytes.toBytes("details"), Bytes.toBytes("ranks"), Bytes.toBytes(num.get()));
    				context.write(keyIn, p);
    			}
    		}
    	}
    	
    	//累加
    	public static class TopMusicReduce3 extends TableReducer<Text, IntWritable, Text>{
    		public void reduce(Text keyIn,Iterable<IntWritable> valuesIn,Context context) throws IOException, InterruptedException{
    			int sum = 0;
    			for(IntWritable num:valuesIn){
    				Put p = new Put(Bytes.toBytes(keyIn.toString()));
    				p.addColumn(Bytes.toBytes("details"), Bytes.toBytes("ranks"), Bytes.toBytes(num.get()));
    				context.write(keyIn, p);
    			}
    		}
    	}
    	
    	//读取 hbase 数据 写到 file 中
    	public static void readHbaseToFile() throws IOException, ClassNotFoundException, InterruptedException{
    		// 创建 job
    		Configuration conf = HBaseConfiguration.create();
    		Job job = Job.getInstance(conf, "topMusicJob");
    		job.setJarByClass(TopMusic.class);
    		job.setMapperClass(TopmusicMapper.class);
    		job.setReducerClass(TopMusicReducel.class);
    
    		// 最终输入的数据类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    
    		// 输出的位置
    		Path p = new Path("/output/topMusic1");
    		FileOutputFormat.setOutputPath(job, p);
    
    		// 输入的配置
    		Scan sc = new Scan();
    		// 筛选数据
    		sc.addColumn(Bytes.toBytes("info"), Bytes.toBytes("song"));
    		TableMapReduceUtil.initTableMapperJob("musicrecord", sc, TopmusicMapper.class, Text.class, IntWritable.class,
    				job, false);
    
    		System.out.println(job.waitForCompletion(true) ? 0 : 1);
    	}
    	
    	//读取 file 写到 hbase 数据库中
    	public static void readFileToHbase() throws IOException, ClassNotFoundException, InterruptedException{
    		Configuration conf = HBaseConfiguration.create();
    		Job job = Job.getInstance(conf,"topMusicJob");
    		job.setJarByClass(TopMusic.class);
    		//设置 mapper 和 reduce 静态处理类
    		job.setMapperClass(TopMusicMapper2.class);
    		
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		//设置原数据文件
    		Path p = new Path("/output/topMusic1");
    		FileInputFormat.addInputPath(job, p);
    		//设置输出到数据库中
    		TableMapReduceUtil.initTableReducerJob("namelist", TopMusicReduce2.class, job,null,null,null,null,false);
    		//启动
    		System.out.println(job.waitForCompletion(true)?0:1);
    	}
    	
    	//从数据库 musicrecord 表读数据 统计完后 写到 namelist details 中去
    	public static void readHbaseToHbase() throws IOException, ClassNotFoundException, InterruptedException{
    		Configuration conf = HBaseConfiguration.create();
    		Job job = Job.getInstance(conf,"topMusicJob");
    		job.setJarByClass(TopMusic.class);
    		// 输入的配置
    		Scan sc = new Scan();
    		// 筛选数据
    		sc.addColumn(Bytes.toBytes("info"), Bytes.toBytes("song"));
    		TableMapReduceUtil.initTableMapperJob("musicrecord", sc, TopmusicMapper.class, Text.class, IntWritable.class,job, false);
    		// 设置输出到数据库中
    		TableMapReduceUtil.initTableReducerJob("namelist", TopMusicReduce3.class, job, null, null, null, null, false);
    		// 启动
    		System.out.println(job.waitForCompletion(true) ? 0 : 1);
    	}
    	
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
    		readHbaseToFile();
    		readFileToHbase();
    	}
    }