當前位置：首頁 > 人文社科 > 生活经验 >内容正文

生活经验

2021年大数据HBase（十五）：HBase的Bulk Load批量加载操作

發(fā)布時間：2023/11/28 生活经验 23 豆豆

生活随笔收集整理的這篇文章主要介紹了 2021年大数据HBase（十五）：HBase的Bulk Load批量加载操作小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

全網(wǎng)最詳細的大數(shù)據(jù)HBase文章系列，強烈建議收藏加關(guān)注！?

新文章都已經(jīng)列出歷史文章目錄，幫助大家回顧前面的知識重點。

系列歷史文章

HBase的Bulk Load批量加載操作

一、Bulk Load 基本介紹

二、需求說明

三、準備工作

1、在hbase中創(chuàng)建名稱空間, 并創(chuàng)建hbase的表

2、創(chuàng)建 maven項目加載相關(guān)的pom 依賴

3、在項目中創(chuàng)建包和添加配置文件log4j.properties

4、將轉(zhuǎn)換csv數(shù)據(jù)上傳到HDFS中: 數(shù)據(jù)在資料中

四、將CSV數(shù)據(jù)轉(zhuǎn)換為HFile文件格式數(shù)據(jù)

map 程序的代碼

驅(qū)動類的代碼?

五、將Hfile文件格式數(shù)據(jù)加載HBase中

語法說明

案例

系列歷史文章

2021年大數(shù)據(jù)HBase（十七）：HBase的360度全面調(diào)優(yōu)

2021年大數(shù)據(jù)HBase（十六）：HBase的協(xié)處理器(Coprocessor)

2021年大數(shù)據(jù)HBase（十五）：HBase的Bulk Load批量加載操作

2021年大數(shù)據(jù)HBase（十四）：HBase的原理及其相關(guān)的工作機制

2021年大數(shù)據(jù)HBase（十三）：HBase讀取和存儲數(shù)據(jù)的流程

2021年大數(shù)據(jù)HBase（十二）：Apache Phoenix 二級索引

2021年大數(shù)據(jù)HBase（十一）：Apache Phoenix的視圖操作

2021年大數(shù)據(jù)HBase（十）：Apache Phoenix的基本入門操作

2021年大數(shù)據(jù)HBase（九）：Apache Phoenix的安裝

2021年大數(shù)據(jù)HBase（八）：Apache Phoenix的基本介紹

2021年大數(shù)據(jù)HBase（七）：Hbase的架構(gòu)！【建議收藏】

2021年大數(shù)據(jù)HBase（六）：HBase的高可用！【建議收藏】

2021年大數(shù)據(jù)HBase（五）：HBase的相關(guān)操作-JavaAPI方式！【建議收藏】

2021年大數(shù)據(jù)HBase（四）：HBase的相關(guān)操作-客戶端命令式！【建議收藏】

2021年大數(shù)據(jù)HBase（三）：HBase數(shù)據(jù)模型

2021年大數(shù)據(jù)HBase（二）：HBase集群安裝操作

2021年大數(shù)據(jù)HBase（一）：HBase基本簡介

HBase的Bulk Load批量加載操作

一、Bulk Load 基本介紹

????????很多時候，我們需要將外部的數(shù)據(jù)導入到HBase集群中，例如：將一些歷史的數(shù)據(jù)導入到HBase做備份。我們之前已經(jīng)學習了HBase的Java API，通過put方式可以將數(shù)據(jù)寫入到HBase中，我們也學習過通過MapReduce編寫代碼將HDFS中的數(shù)據(jù)導入到HBase。但這些方式都是基于HBase的原生API方式進行操作的。這些方式有一個共同點，就是需要與HBase連接，然后進行操作。HBase服務器要維護、管理這些連接，以及接受來自客戶端的操作，會給HBase的存儲、計算、網(wǎng)絡資源造成較大消耗。此時，在需要將海量數(shù)據(jù)寫入到HBase時，通過Bulk load（大容量加載）的方式，會變得更高效。可以這么說，進行大量數(shù)據(jù)操作，Bulk load是必不可少的。

? ? ? ?我們知道，HBase的數(shù)據(jù)最終是需要持久化到HDFS。HDFS是一個文件系統(tǒng)，那么數(shù)據(jù)可定是以一定的格式存儲到里面的。例如：Hive我們可以以ORC、Parquet等方式存儲。而HBase也有自己的數(shù)據(jù)格式，那就是HFile。Bulk Load就是直接將數(shù)據(jù)寫入到StoreFile（HFile）中，從而繞開與HBase的交互，HFile生成后，直接一次性建立與HBase的關(guān)聯(lián)即可。使用BulkLoad，繞過了Write to WAL，Write to MemStore及Flush to disk的過程

原有的數(shù)據(jù)寫入操作大致流轉(zhuǎn)流程:

正常寫入數(shù)據(jù)的流程: 數(shù)據(jù)寫入到Hlog --> MemStore --> StoreFile --> Hfile?

如果以及有一批數(shù)據(jù), 需要寫入到Hbase中某個表中, 傳統(tǒng)做法, 按照上述流程, 一步步將數(shù)據(jù)最終寫入Hfile中, 此時整個region集群會經(jīng)歷大量的寫入請求操作,HBase集群需要調(diào)度大量資源來滿足本次的數(shù)據(jù)寫入工作,如果這個時候, 又出現(xiàn)大量的讀取數(shù)據(jù)請求也去訪問這個表, 會發(fā)生什么問題呢? 讀取性能有可能回受到影響甚至出現(xiàn)卡頓現(xiàn)象

思考如何解決呢?
?? ?對一批數(shù)據(jù), 提前按照HBase的Hfile文件格式存儲好, 然后將Hfile文件格式數(shù)據(jù)直接放置到Hbase對應數(shù)據(jù)目錄下, 讓Hbase直接加載, 此時不需要Hbase提供大量的寫入資源, 即可完成全部數(shù)據(jù)寫入操作
?
總結(jié):?
?? ?第一個步驟: 將數(shù)據(jù)文件轉(zhuǎn)換為HFile文件格式 ? -- MapReduce
?? ?第二個步驟: 將Hfile文件格式數(shù)據(jù)加載到Hbase中

二、需求說明

需求: 需要將每一天的銀行轉(zhuǎn)賬記錄的數(shù)據(jù) 存儲到HBase中 , 數(shù)據(jù)量比較的龐大

數(shù)據(jù)所在位置: HDFS中,
數(shù)據(jù)格式為: CSV 分割符號為逗號

三、準備工作

1、在hbase中創(chuàng)建名稱空間, 并創(chuàng)建hbase的表

# 創(chuàng)建名稱空間: 
create_namespace 'IT_BANK'
# 在名稱空間下, 創(chuàng)建目標表: 
create 'IT_BANK:TRANSFER_RECORD' ,{NAME=>'C1',COMPRESSION=>'GZ'},{NUMREGIONS=>6,SPLITALGO=>'HexStringSplit'}

2、創(chuàng)建 maven項目加載相關(guān)的pom 依賴

說明: 如果將此全部導入到本項目中, 出現(xiàn)全部爆紅錯誤, 可以將此內(nèi)容放置到父工程的pom依賴中

此時有可能導致其他某個項目爆紅(不用管), 只需要保證當前自己的項目沒有任何問題即可

<repositories><repository><id>aliyun</id><url>http://maven.aliyun.com/nexus/content/groups/public/</url><releases><enabled>true</enabled></releases><snapshots><enabled>false</enabled><updatePolicy>never</updatePolicy></snapshots></repository></repositories><dependencies><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>2.1.0</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-mapreduce</artifactId><version>2.1.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-jobclient</artifactId><version>2.7.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>2.7.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-core</artifactId><version>2.7.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-auth</artifactId><version>2.7.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>2.7.5</version></dependency><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.6</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><target>1.8</target><source>1.8</source></configuration></plugin></plugins></build>

3、在項目中創(chuàng)建包和添加配置文件log4j.properties

4、將轉(zhuǎn)換csv數(shù)據(jù)上傳到HDFS中: 數(shù)據(jù)在資料中

hdfs dfs -mkdir -p /bulkLoad/outputhdfs dfs -put bank_record.csv /bulkLoad/output

四、將CSV數(shù)據(jù)轉(zhuǎn)換為HFile文件格式數(shù)據(jù)

map 程序的代碼

package com.it.bulkload.mr;import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class BulkLoadMapper extends Mapper<LongWritable,Text,ImmutableBytesWritable,Put> {private  ImmutableBytesWritable k2 = new ImmutableBytesWritable();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//1. 獲取一行數(shù)據(jù)String line = value.toString();if(line != null && !"".equalsIgnoreCase(line)){//2. 對數(shù)據(jù)進行切割操作String[] fields = line.split(",");//3. 封裝k2 和 v2的數(shù)據(jù)//封裝 k2String rowkeyStr = fields[0];k2.set(rowkeyStr.getBytes());// 封裝v2Put v2 = new Put(rowkeyStr.getBytes());v2.addColumn("C1".getBytes(),"code".getBytes(),fields[1].getBytes());v2.addColumn("C1".getBytes(),"rec_account".getBytes(),fields[2].getBytes());v2.addColumn("C1".getBytes(),"rec_bank_name".getBytes(),fields[3].getBytes());v2.addColumn("C1".getBytes(),"rec_name".getBytes(),fields[4].getBytes());v2.addColumn("C1".getBytes(),"pay_account".getBytes(),fields[5].getBytes());v2.addColumn("C1".getBytes(),"pay_name".getBytes(),fields[6].getBytes());v2.addColumn("C1".getBytes(),"pay_comments".getBytes(),fields[7].getBytes());v2.addColumn("C1".getBytes(),"pay_channel".getBytes(),fields[8].getBytes());v2.addColumn("C1".getBytes(),"pay_way".getBytes(),fields[9].getBytes());v2.addColumn("C1".getBytes(),"status".getBytes(),fields[10].getBytes());v2.addColumn("C1".getBytes(),"timestamp".getBytes(),fields[11].getBytes());v2.addColumn("C1".getBytes(),"money".getBytes(),fields[12].getBytes());//4. 輸出context.write(k2,v2);}}
}

驅(qū)動類的代碼?

package com.it.bulkload.mr;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;public class BulkLoadDriver {public static void main(String[] args) throws Exception {//1. 獲取job對象//Configuration conf = new Configuration();Configuration conf = HBaseConfiguration.create();conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");Job job = Job.getInstance(conf, "BulkLoadDriver");//2. 配置集群運行的必備項job.setJarByClass(BulkLoadDriver.class);//3. 配置 MR的天龍八部//3.1: 指定輸入類, 及其輸入的路徑job.setInputFormatClass(TextInputFormat.class);TextInputFormat.addInputPath(job,new Path("hdfs://node1:8020/bulkLoad/input/bank_record.csv"));//3.2: 指定map類 及其輸出的k2和v2的類型job.setMapperClass(BulkLoadMapper.class);job.setMapOutputKeyClass(ImmutableBytesWritable.class);job.setMapOutputValueClass(Put.class);//3.3 : 指定 shuffle操作:  分區(qū) 排序 規(guī)約 分組  默認即可//3.7: 指定reduce類, 及其輸出 k3 和 v3的類型job.setNumReduceTasks(0);job.setOutputKeyClass(ImmutableBytesWritable.class);job.setOutputValueClass(Put.class);//3.8: 設置輸出類, 及其輸出的路徑: HFile文件格式job.setOutputFormatClass(HFileOutputFormat2.class);HFileOutputFormat2.setOutputPath(job,new Path("hdfs://node1:8020/bulkLoad/output"));Connection conn = ConnectionFactory.createConnection(conf);Table table = conn.getTable(TableName.valueOf("IT_BANK:TRANSFER_RECORD"));HFileOutputFormat2.configureIncrementalLoad(job,table,conn.getRegionLocator(TableName.valueOf("IT_BANK:TRANSFER_RECORD")));//4. 提交任務boolean flag = job.waitForCompletion(true);//5. 退出程序System.exit(flag ? 0 : 1);}}

五、將Hfile文件格式數(shù)據(jù)加載HBase中

語法說明

hbase org.apache.hadoop.hbase.tool.LoadIncrementalHFiles  數(shù)據(jù)路徑 Hbase表名

案例

hbase org.apache.hadoop.hbase.tool.LoadIncrementalHFiles  hdfs://node1:8020/bulkLoad/output/  IT_BANK:TRANSFER_RECORD

📢博客主頁：https://lansonli.blog.csdn.net
📢歡迎點贊 👍 收藏 ?留言 📝 如有錯誤敬請指正！
📢本文由 Lansonli 原創(chuàng)，首發(fā)于 CSDN博客🙉
📢大數(shù)據(jù)系列文章會每天更新，停下休息的時候不要忘了別人還在奔跑，希望大家抓緊時間學習，全力奔赴更美好的生活?

總結(jié)

以上是生活随笔為你收集整理的2021年大数据HBase（十五）：HBase的Bulk Load批量加载操作的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。

上一篇： 2021年大数据HBase（十四）：HB
下一篇： 2021年大数据HBase（十六）：HB

生活经验

2021年大数据HBase（十五）：HBase的Bulk Load批量加载操作

系列歷史文章

HBase的Bulk Load批量加載操作

一、Bulk Load 基本介紹

二、需求說明

三、準備工作

1、在hbase中創(chuàng)建名稱空間, 并創(chuàng)建hbase的表

2、創(chuàng)建 maven項目 加載相關(guān)的pom 依賴

3、在項目中創(chuàng)建包 和 添加配置文件log4j.properties

4、將轉(zhuǎn)換csv數(shù)據(jù)上傳到HDFS中: 數(shù)據(jù)在資料中

四、將CSV數(shù)據(jù)轉(zhuǎn)換為HFile文件格式數(shù)據(jù)

map 程序的代碼

驅(qū)動類的代碼?

五、將Hfile文件格式數(shù)據(jù)加載HBase中

語法說明

案例

總結(jié)

2、創(chuàng)建 maven項目加載相關(guān)的pom 依賴

3、在項目中創(chuàng)建包和添加配置文件log4j.properties