读取HBase中的表到Mapper处理，输出至HDFS

你的名字 2023-06-23 12:51 40阅读 0赞

是maven环境，需要导入的包：

<dependency>
    				<groupId>org.apache.hadoop</groupId>
    				<artifactId>hadoop-common</artifactId>
    				<version>2.7.3</version>
    
    			</dependency>
    
    			
    			<dependency>
    				<groupId>org.apache.hadoop</groupId>
    				<artifactId>hadoop-hdfs</artifactId>
    				<version>2.7.3</version>
    			</dependency>
    
    			
    			
    
    			<dependency>
    				<groupId>junit</groupId>
    				<artifactId>junit</artifactId>
    				<version>4.10</version>
    			</dependency>
    
    
    
    			
    			<dependency>
    				<groupId>org.apache.mrunit</groupId>
    				<artifactId>mrunit</artifactId>
    				<version>0.9.0-incubating</version>
    				<classifier>hadoop2</classifier>
    				<scope>test</scope>
    			</dependency>
    
    
    
    
    
    			<dependency>
    				<groupId>jdk.tools</groupId>
    				<artifactId>jdk.tools</artifactId>
    				<version>1.8</version>
    				<scope>system</scope>
    				<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
    			</dependency>
    
    
    			
    			<dependency>
    				<groupId>org.apache.hbase</groupId>
    				<artifactId>hbase</artifactId>
    				<version>1.3.2</version>
    				<type>pom</type>
    			</dependency>
    
    			
    			<dependency>
    				<groupId>org.apache.hbase</groupId>
    				<artifactId>hbase-common</artifactId>
    				<version>1.3.2</version>
    			</dependency>
    			
    			<dependency>
    				<groupId>org.apache.hbase</groupId>
    				<artifactId>hbase-server</artifactId>
    				<version>1.3.2</version>
    			</dependency>
    
    
    			
    			<dependency>
    				<groupId>org.apache.hbase</groupId>
    				<artifactId>hbase-client</artifactId>
    				<version>1.3.2</version>
    			</dependency>
    
    			
    			<dependency>
    				<groupId>org.apache.zookeeper</groupId>
    				<artifactId>zookeeper</artifactId>
    				<version>3.4.6</version>
    				<type>pom</type>
    			</dependency>
    
    			
    			<dependency>
    				<groupId>org.glassfish.jersey.core</groupId>
    				<artifactId>jersey-client</artifactId>
    				<version>2.26</version>
    			</dependency>

1 前提是要Hbase和HDFS集群能跑起来  
目录结构：  
![在这里插入图片描述][2019121822552727.png]  
2 需要将该项目进行打包：jar包  
3 代码部分：

public class TableMapperDemo {
    	//继承TableMapper类
    	static class MyMapper extends TableMapper<Text, Text> {
    
    		
    		//map(Rowkey, 一条记录 value,context)
    		//注意：map参数是固定的
    		@Override
    		protected void map(ImmutableBytesWritable key, Result value,
    				Context context) throws IOException, InterruptedException {	
    			//得到所有单元格
    			List<Cell> cells=value.listCells();
    			//循环遍历单元格
    			for(Cell cell:cells){
    				
    				String outValue=String.format("RowKey:%s    Family:%s    Qualifier:%s    cellValue:%s  ", 
    						Bytes.toString(key.get()),
    						//将字节转化为String类型
    						Bytes.toString(CellUtil.cloneFamily(cell)),//列组
    						Bytes.toString(CellUtil.cloneQualifier(cell)),//列名字
    						Bytes.toString(CellUtil.cloneValue(cell))//值
    						);
    				//maper输出  单元格的key和outvalue
    				context.write(
    						new Text(CellUtil.getCellKeyAsString(cell)), //获取 单元格的key.不是行键,也不是列键
    						new Text(outValue));
    				System.out.println(new Text(CellUtil.getCellKeyAsString(cell)));
    			}
    		}
    	}
    	
    	public static void m1(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
    		//获取配置文件
    		Configuration conf = HBaseConfiguration.create();
    		//跨平台运行
    		conf.set("mapreduce.app-submission.cross-platform","true");
            //设置读取的表:在hbase中 的表
    		conf.set(TableInputFormat.INPUT_TABLE, "music4");	
    		conf.set(TableInputFormat.SCAN_COLUMNS, "info:name info:gender");
    		//conf.set(TableInputFormat.SCAN_ROW_START, "19_song9_2016-1-12");
    		//开启任务
            Job job = Job.getInstance(conf, "hbase-mapreduce-api");
            
            //MapReduce程序作业基本配置
            job.setJar("E:/BigDate/jar/HbaseMaven.jar");
            job.setInputFormatClass(TableInputFormat.class);
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
                   
           
          //解决三方jar文件依赖:3)使用TableMapReduceUtil
    		TableMapReduceUtil.addDependencyJars(job);//推荐使用   加入分布缓存中
    		//用于输出到HDFS中的路径
            Path output=new Path("/user/sheng/out/music1");
    		if(FileSystem.get(conf).exists(output))
    			FileSystem.get(conf).delete(output,true);
    		FileOutputFormat.setOutputPath(job, output);   
            job.waitForCompletion(true);
    		
    	}
    	
    	public static void m2(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
    		Configuration conf = HBaseConfiguration.create();
    		//解决三方jar文件依赖:1)使用-libjars
    		GenericOptionsParser gop= new GenericOptionsParser(conf, args);
    		System.out.println(conf.get("tmpjars"));     
            Job job = Job.getInstance(conf, "hbase-mapreduce-api");
            
            //MapReduce程序作业基本配置
            job.setJarByClass(TableMapperDemo.class); 
            
            Path output=new Path("a");
    		if(FileSystem.get(conf).exists(output))
    			FileSystem.get(conf).delete(output,true);
    		FileOutputFormat.setOutputPath(job, output);
            
    		
    		//使用hbase提供的工具类来设置job
    		Scan scan=new Scan();
    		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));
    		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"));
    		TableMapReduceUtil.initTableMapperJob("music", scan, MyMapper.class, Text.class, Text.class, job);
            
            job.waitForCompletion(true);
    		
    	}
    	
    	public static void main(String[] args) throws IOException,
    			ClassNotFoundException, InterruptedException {
    		m1(args);//使用TableMapper+TableInputFormat
    //		m2(args);//使用TableMapper+TableMapReduceUtil
    	
    	
    	}
    
    	
    }

[2019121822552727.png]: https://img-blog.csdnimg.cn/2019121822552727.png