在離線任務場景中,MapReduce訪問HBASE數據,加快分析速度和擴展分析能力。
從hbase中讀取數據(result)node
public class ReadHBaseDataMR { private static final String ZK_KEY = "hbase.zookeeper.quorum"; private static final String ZK_VALUE = "hadoop01:2181,hadoop01:2182,hadoop03:2181"; private static Configuration conf; static { conf=HBaseConfiguration.create(); conf.set(ZK_KEY,ZK_VALUE); //由於是從hbase中讀取到本身的hdfs集羣中,因此這裏須要加載hdfs的配置文件 conf.addResource("core-site.xml"); conf.addResource("hdfs-site.xml"); } //job public static void main(String[] args) { Job job = null; try { //這裏使用hbase的 conf job = Job.getInstance(conf); job.setJarByClass(ReadHBaseDataMR.class); //全表掃描 Scan scans=new Scan(); String tableName="user_info"; //設置MapReduce與hbase的整合 TableMapReduceUtil.initTableMapperJob(tableName, scans, ReadHBaseDataMR_Mapper.class, Text.class, NullWritable.class, job, false); //設置ReducerTask 的個數爲0 job.setNumReduceTasks(0); //設置輸出搭配hdfs上的路徑 Path output=new Path("/output/hbase/hbaseToHDFS"); if(output.getFileSystem(conf).exists(output)) { output.getFileSystem(conf).delete(output, true); } FileOutputFormat.setOutputPath(job, output); //提交任務 boolean waitForCompletion = job.waitForCompletion(true); System.exit(waitForCompletion?0:1); } catch (Exception e) { e.printStackTrace(); } } //Mapper //使用TableMapper,去讀取hbase中的表的數據 private static class ReadHBaseDataMR_Mapper extends TableMapper<Text, NullWritable> { Text mk = new Text(); NullWritable kv = NullWritable.get(); @Override protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException { //默認的按照每個rowkey讀取 List<Cell> cells = value.listCells(); //這裏以四個座標肯定一行記錄,行鍵,列簇,列,時間戳 for(Cell cell:cells){ String row= Bytes.toString(CellUtil.cloneRow(cell)); //行鍵 String cf=Bytes.toString(CellUtil.cloneFamily(cell)); //列簇 String column=Bytes.toString(CellUtil.cloneQualifier(cell)); //列 String values=Bytes.toString(CellUtil.cloneValue(cell)); //值 long time=cell.getTimestamp(); //時間戳 mk.set(row+"\t"+cf+"\t"+column+"\t"+value+"\t"+time); context.write(mk,kv); } } } }
寫入數據到hbase中(put)mysql
public class HDFSToHbase { private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum"; private static final String ZK_CONNECT_VALUE = "hadoop02:2181,hadoop03:2181,hadoop01:2181"; private static Configuration conf; static { conf=HBaseConfiguration.create(); conf.set(ZK_CONNECT_KEY,ZK_CONNECT_VALUE); //由於是從hbase中讀取到本身的hdfs集羣中,因此這裏須要加載hdfs的配置文件 conf.addResource("core-site.xml"); conf.addResource("hdfs-site.xml"); } //job public static void main(String[] args) { try { Job job = Job.getInstance(conf); job.setJarByClass(HDFSToHbase.class); job.setMapperClass(MyMapper.class); //指定Map端的輸出 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); /** * 指定爲nulL的表示使用默認的 */ String tableName="student"; //整合MapReduce reducer 到hbase TableMapReduceUtil.initTableReducerJob(tableName,MyReducer.class, job,null, null, null, null, false ); //指定MapReducer的輸入路徑 Path input = new Path("/in/mingxing.txt"); FileInputFormat.addInputPath(job, input); //提交任務 boolean waitForCompletion = job.waitForCompletion(true); System.exit(waitForCompletion ? 0 : 1); } catch (Exception e) { e.printStackTrace(); } } //Mapper private static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> { NullWritable mv = NullWritable.get(); //map端不作任何操做,直接將讀取的數據輸出到reduce端 @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value, mv); } } //Reudcer,使用TableReducer的Reudcer /** * TableReducer<KEYIN, VALUEIN, KEYOUT> * KEYIN:mapper輸出的key * VALUEIN:mapper輸出的value * KEYOUT:reduce輸出的key * 默認的有第四個參數:Mutation,表示put/delete操做 */ private static class MyReducer extends TableReducer<Text, NullWritable, NullWritable>{ //列簇 String family[] = { "basicinfo","extrainfo"}; @Override protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { // zhangfenglun,M,20,13522334455,zfl@163.com,23521472 字段 for(NullWritable value:values){ String fields[]=key.toString().split(","); //以名稱做爲rowkey Put put=new Put(fields[0].getBytes()); put.addColumn(fields[0].getBytes(),"sex".getBytes(),fields[1].getBytes()); put.addColumn(fields[0].getBytes(),"age".getBytes(),fields[2].getBytes()); put.addColumn(fields[1].getBytes(),"phone".getBytes(),fields[3].getBytes()); put.addColumn(fields[1].getBytes(),"email".getBytes(),fields[4].getBytes()); put.addColumn(fields[1].getBytes(),"qq".getBytes(),fields[5].getBytes()); context.write(value, put); } } } }
#使用sqoop從MySQL導入HBASEsql
sqoop import \ --connect jdbc:mysql://hadoop01:3306/test \ #MySQL的入口 --username hadoop \ #登陸MySQL的用戶名 --password root \ #登陸MySQL的密碼 --table book \ #插入的到MySQL的表 --hbase-table book \ #HBASE的表名 --column-family info \ #HBASE表中的列簇 --hbase-row-key bid \ #mysql中的哪個列爲rowkey #ps:這裏因爲版本不兼容的問題,因此,這裏的HBASE中插入的表必須提早建立,而且不能使用:--hbase-create-table \,這個語句
原理:Hive與HBASE利用二者自己對外的API來實現整合,主要靠的是HBaseStorageHandler 進 行通訊,利用 HBaseStorageHandler,Hive 能夠獲取到 Hive 表對應的 HBase 表名,列簇以及 列,InputFormat 和 OutputFormat 類,建立和刪除 HBase 表等。
Hive 訪問 HBase 中表數據,實質上是經過 MapReduce 讀取 HBase 表數據,其實現是在 MR 中,使用 HiveHBaseTableInputFormat 完成對 HBase 表的切分,獲取 RecordReader 對象來讀 取數據。
對HBASE表的切分原則:一個region切分紅一個split,即表中有多少個region,MapReduce就有多少個map task。
讀取HBASE表數據都是經過scanner,對錶進行全表掃描,若是有過濾條件,則轉化爲filter,當過濾條件爲rowkey時,則轉化爲rowkey的過濾。
具體操做:apache
#指定 hbase 所使用的 zookeeper 集羣的地址:默認端口是 2181,能夠不寫: hive>set hbase.zookeeper.quorum=hadoop02:2181,hadoop03:2181,hadoop04:2181; #指定 hbase 在 zookeeper 中使用的根目錄 hive>set zookeeper.znode.parent=/hbase; #建立基於 HBase 表的 hive 表 hive>create external table mingxing(rowkey string, base_info map, extra_info map) row format delimited fields terminated by '\t' >stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' >with serdeproperties ("hbase.columns.mapping" = ":key,base_info:,extra_info:") >tblproperties("hbase.table.name"="mingxing","hbase.mapred.output.outputtable"="mingxing"); #ps:org.apache.hadoop.hive.hbase.HBaseStorageHandler:處理 hive 到 hbase 轉換關係的處理器 #ps:hbase.columns.mapping:定義 hbase 的列簇和列到 hive 的映射關係 #ps:hbase.table.name:hbase 表名
雖然hive整合了hbase,可是實際的數據仍是存儲在hbase上,hive相應的表目錄下對應的文件爲空,可是每次hbase中有數據添加時,hive在執行這張表查詢的時候,也會更新相應的字段。app