2. 程式人生 > >hbase海量資料匯入

hbase海量資料匯入

阿新 • • 發佈：2019-01-24
最近有個需求要對mysql的全量資料遷移到hbase,雖然hbase的設計非常利於高效的讀取，但是它的compaction實現對海量資料寫入造成非常大的影響，資料到一定量之後，就開始抽風。 
分析hbase的實現，不管其執行的機制，其最終儲存結構為分散式檔案系統中的hfile格式。 
剛好hbase的原始碼中提供一個HFileOutputFormat類，分析其原始碼可以看到： 
Java程式碼  收藏程式碼
  1. /** 
  2.  * Copyright 2009 The Apache Software Foundation 
  3.  * 
  4.  * Licensed to the Apache Software Foundation (ASF) under one
     
  5.  * or more contributor license agreements.  See the NOTICE file 
  6.  * distributed with this work for additional information 
  7.  * regarding copyright ownership.  The ASF licenses this file 
  8.  * to you under the Apache License, Version 2.0 (the 
  9.  * "License"); you may not use this file except in compliance
     
  10.  * with the License.  You may obtain a copy of the License at 
  11.  * 
  12.  *     http://www.apache.org/licenses/LICENSE-2.0 
  13.  * 
  14.  * Unless required by applicable law or agreed to in writing, software 
  15.  * distributed under the License is distributed on an "AS IS" BASIS, 
  16.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     
  17.  * See the License for the specific language governing permissions and 
  18.  * limitations under the License. 
  19.  */  
  20. package org.apache.hadoop.hbase.mapreduce;  
  21. import java.io.IOException;  
  22. import java.util.Map;  
  23. import java.util.TreeMap;  
  24. import org.apache.hadoop.conf.Configuration;  
  25. import org.apache.hadoop.fs.FileSystem;  
  26. import org.apache.hadoop.fs.Path;  
  27. import org.apache.hadoop.hbase.HConstants;  
  28. import org.apache.hadoop.hbase.KeyValue;  
  29. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;  
  30. import org.apache.hadoop.hbase.io.hfile.Compression;  
  31. import org.apache.hadoop.hbase.io.hfile.HFile;  
  32. import org.apache.hadoop.hbase.regionserver.StoreFile;  
  33. import org.apache.hadoop.hbase.util.Bytes;  
  34. import org.apache.hadoop.mapreduce.RecordWriter;  
  35. import org.apache.hadoop.mapreduce.TaskAttemptContext;  
  36. import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;  
  37. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
  38. import org.mortbay.log.Log;  
  39. /** 
  40.  * Writes HFiles. Passed KeyValues must arrive in order. 
  41.  * Currently, can only write files to a single column family at a 
  42.  * time.  Multiple column families requires coordinating keys cross family. 
  43.  * Writes current time as the sequence id for the file. Sets the major compacted 
  44.  * attribute on created hfiles. 
  45.  * @see KeyValueSortReducer 
  46.  */  
  47. public class HFileOutputFormat extends FileOutputFormat<ImmutableBytesWritable, KeyValue> {  
  48.   public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(TaskAttemptContext context)  
  49.   throws IOException, InterruptedException {  
  50.     // Get the path of the temporary output file   
  51.     final Path outputPath = FileOutputFormat.getOutputPath(context);  
  52.     final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath();  
  53.     Configuration conf = context.getConfiguration();  
  54.     final FileSystem fs = outputdir.getFileSystem(conf);  
  55.     // These configs. are from hbase-*.xml  
  56.     final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456);  
  57.     final int blocksize = conf.getInt("hfile.min.blocksize.size", 65536);  
  58.     // Invented config.  Add to hbase-*.xml if other than default compression.  
  59.     final String compression = conf.get("hfile.compression",  
  60.       Compression.Algorithm.NONE.getName());  
  61.     return new RecordWriter<ImmutableBytesWritable, KeyValue>() {  
  62.       // Map of families to writers and how much has been output on the writer.  
  63.       private final Map<byte [], WriterLength> writers =  
  64.         new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR);  
  65.       private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY;  
  66.       private final byte [] now = Bytes.toBytes(System.currentTimeMillis());  
  67.       public void write(ImmutableBytesWritable row, KeyValue kv)  
  68.       throws IOException {  
  69.         long length = kv.getLength();  
  70.         byte [] family = kv.getFamily();  
  71.         WriterLength wl = this.writers.get(family);  
  72.         if (wl == null || ((length + wl.written) >= maxsize) &&  
  73.             Bytes.compareTo(this.previousRow, 0, this.previousRow.length,  
  74.               kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) {  
  75.           // Get a new writer.  
  76.           Path basedir = new Path(outputdir, Bytes.toString(family));  
  77.           if (wl == null) {  
  78.             wl = new WriterLength();  
  79.             this.writers.put(family, wl);  
  80.             if (this.writers.size() > 1) throw new IOException("One family only");  
  81.             // If wl == null, first file in family.  Ensure family dir exits.  
  82.             if (!fs.exists(basedir)) fs.mkdirs(basedir);  
  83.           }  
  84.           wl.writer = getNewWriter(wl.writer, basedir);  
  85.           Log.info("Writer=" + wl.writer.getPath() +  
  86.             ((wl.written == 0)? "": ", wrote=" + wl.written));  
  87.           wl.written = 0;  
  88.         }  
  89.         kv.updateLatestStamp(this.now);  
  90.         wl.writer.append(kv);  
  91.         wl.written += length;  
  92.         // Copy the row so we know when a row transition.  
  93.         this.previousRow = kv.getRow();  
  94.       }  
  95.       /* Create a new HFile.Writer. Close current if there is one. 
  96.        * @param writer 
  97.        * @param familydir 
  98.        * @return A new HFile.Writer. 
  99.        * @throws IOException 
  100.        */  
  101.       private HFile.Writer getNewWriter(final HFile.Writer writer,  
  102.           final Path familydir)  
  103.       throws IOException {  
  104.         close(writer);  
  105.         return new HFile.Writer(fs,  StoreFile.getUniqueFile(fs, familydir),  
  106.           blocksize, compression, KeyValue.KEY_COMPARATOR);  
  107.       }  
  108.       private void close(final HFile.Writer w) throws IOException {  
  109.         if (w != null) {  
  110.           StoreFile.appendMetadata(w, System.currentTimeMillis(), true);  
  111.           w.close();  
  112.         }  
  113.       }  
  114.       public void close(TaskAttemptContext c)  
  115.       throws IOException, InterruptedException {  
  116.         for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) {  
  117.           close(e.getValue().writer);  
  118.         }  
  119.       }  
  120.     };  
  121.   }  
  122.   /* 
  123.    * Data structure to hold a Writer and amount of data written on it.  
  124.    */  
  125.   static class WriterLength {  
  126.     long written = 0;  
  127.     HFile.Writer writer = null;  
  128.   }  
  129. }  


可以看到，它的工作流程就是首先根據你的配置檔案初始化，然後寫成hfile的格式。 
這裡我做了個偷懶的demo: 
Java程式碼  收藏程式碼
  1. HFileOutputFormat hf = new HFileOutputFormat();  
  2.         HBaseConfiguration conf = new HBaseConfiguration();  
  3.         conf.addResource(new Path("/home/performance/softs/hadoop/conf/core-site.xml"));  
  4.         conf.set("mapred.output.dir", "/tmp");  
  5.         conf.set("hfile.compression", Compression.Algorithm.LZO.getName());  
  6.         TaskAttemptContext context = new TaskAttemptContext(conf, new TaskAttemptID());  
  7.         RecordWriter writer = hf.getRecordWriter(context);  
  8.         KeyValue kv = new KeyValue(Bytes.toBytes("1111111111111"), Bytes.toBytes("offer:action"),  
  9.                                    System.currentTimeMillis(), Bytes.toBytes("test"));  
  10.         KeyValue kv1 = new KeyValue(Bytes.toBytes("1111111111111"), Bytes.toBytes("offer:id"),  
  11.                                     System.currentTimeMillis(), Bytes.toBytes("123"));  
  12.         KeyValue kv3 = new KeyValue(Bytes.toBytes("1111111111112"), Bytes.toBytes("offer:action"),  
  13.                                     System.currentTimeMillis(), Bytes.toBytes("test"));  
  14.         KeyValue kv4 = new KeyValue(Bytes.toBytes("1111111111112"), Bytes.toBytes("offer:id"),  
  15.                                     System.currentTimeMillis(), Bytes.toBytes("123"));  
  16.         writer.write(null, kv);  
  17.         writer.write(null, kv1);  
  18.         writer.write(null, kv3);  
  19.         writer.write(null, kv4);  
  20.         writer.close(context);  

執行然之後，會在hdfs的/tmp目錄下生成一份檔案。注意批量寫資料的時候一定要保證key的有序性
這個時候，hbase自己提供的一個基於jruby的loadtable.rb指令碼就可以發揮作用了。 
它的格式是loadtable.rb 你希望的表明 hdfs路徑： 
hbase org.jruby.Main loadtable.rb offer hdfs://user/root/importoffer/_temporary/_attempt__0000_r_000000_0/ 
執行完之後: 
執行./hbase shell 
>list 
就會顯示剛才匯入的offer表了。 

相關推薦

hbase海量資料匯入

最近有個需求要對mysql的全量資料遷移到hbase,雖然hbase的設計非常利於高效的讀取，但是它的compaction實現對海量資料寫入造成非常大的影響，資料到一定量之後，就開始抽風。 分析hbase的實現，不管其執行的機制，其最終儲存結構為分散式檔案系統中的hfile

hbase海量資料的全量匯入方法

最近有個需求要對mysql的全量資料遷移到hbase,雖然hbase的設計非常利於高效的讀取，但是它的compaction實現對海量資料寫入造成非常大的影響，資料到一定量之後，就開始抽風。 分析hbase的實現，不管其執行的機制，其最終儲存結構為分散式檔案系統中的hfile格

資料匯入終章：如何將HBase的資料匯入HDFS？

我們的最終目標是將資料匯入Hadoop，在之前的章節中，我們介紹瞭如何將傳統關係資料庫的資料匯入Hadoop，本節涉及到了HBase。HBase是一種實時分散式資料儲存系統，通常位於與Hadoop叢集相同的硬體上，或者與Hadoop叢集緊密相連，能夠直接在MapReduce中使用HBase資料，或將

【資料庫】load data infile上億條的海量資料匯入mysql的那些事

因為做股票金融的，每天產生的資料量是很大的，一個月幾十億的交易記錄，也常有出現，特別是今年大跌之前大漲那會。 作為程式設計師，問題來了，有時需要將一些並不是特別符合規範的csv檔案匯入資料庫中，而且每個檔案有十幾萬行，而這樣的檔案幾萬個，於是幾十億的記錄如何匯入資料庫呢？

Mysql海量資料--匯入匯出、備份恢復方法

使用phpmyadmin確實很方便。但是當我們需要匯出幾百兆甚至幾個G的資料庫時，phpmyadmin就會很卡。甚至如果你是想匯入大量資料，則phpmyadmin完全無法完成。因為phpmyadmin是基於PHP網頁的

Hbase通過BulkLoad的方式快速匯入海量資料

摘要 載入資料到HBase的方式有多種，通過HBase API匯入或命令列匯入或使用第三方(如sqoop)來匯入或使用MR來批量匯入（耗費磁碟I/O，容易在匯入的過程使節點宕機），但是這些方式不是慢就是在匯入的過程的佔用Region資料導致效率低下，今天要講的就是利用H

利用Sqoop將MySQL海量測試資料匯入HDFS和HBase

宣告：作者原創，轉載註明出處。 一、安裝Sqoop 1、下載sqoop,解壓、資料夾重新命名 wget http://mirror.bit.edu.cn/apache/sqoop/1.4.6/sqoop-1.4.6.bin_

MySQL-8.0.12資料庫的解除安裝，安裝和海量sql資料匯入

     最近學習mySQL，作為一名新手，遇到了三個問題。      第一：如何解除安裝乾淨原來的mySQL;      第二：如何下載並正確安裝mySQL; &nb

面對海量資料儲存，如何保證HBase叢集的高效以及穩定

內容來源：2018 年 09 月 15 日，平安科技資料平臺部大資料高階工程師鄧傑在“中國HBase技術社群第五屆MeetUp ——HBase應用與發展”進行《HBase應用與實踐》的演講分享。IT 大咖說作為獨家視訊合作方，經主辦方和講者審閱授權釋出。 閱讀

hbase資料匯入匯出

hbase資料匯入 將本地檔案(test.csv)上傳到hdfs的根目錄下,然後匯入資料到hbase 1.本地寫一個檔案進行測試，檔名為test.csv,內容如下： 2.將檔案上傳到Hadoop 3.檢視是否上傳成功（檔案存在，表示成功） 4.進入hbase s

Mysql 資料匯入 Hbase

目錄 一、前言 一、前言 在大資料專案中需要做資料遷移時，我們第一時間總會想到sqoop。sqoop是apache 旗下一款“Hadoop 和關係資料庫伺服器之間傳送資料”的工具，

hbase資料匯入hdfs中之（使用MapReduce程式設計統計hbase庫中的mingxing表中男女數量）

資料 zhangfenglun,M,20,13522334455,[email protected],23521472 chenfei,M,20,13684634455,[email protected],84545472 liyuchen,M,20,1352233425

flume將資料匯入到hbase中

安裝flume： [[email protected] ~]$ tar -zxvf flume-ng-1.6.0-cdh5.5.2.tar.gz 修改 flume-env.sh 配置檔案,主要是JAVA_HOME變數設定 [[email protect

mysql中的資料匯入到hbase中，並關聯phoenix

1.在hbase上建立表： hbase>create 'ES','f1' 2.phoenix上建立表： jdbc:phoenix:es01> create table ES(ids varchar primary key ,"f1"."class_name" va

將sqlserver的資料匯入hbase中

將sqlserver的資料匯入hbase中 1.解壓sqoop-sqlserver-1.0.tar.gz，並改名（可以不改）          tar  -zxvf  sqoop- sql

HBASE資料匯入HIVE

package com.hx.data.hbase import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName} import org

Elasticsearch+Hbase實現海量資料秒回查詢

---------------------------------------------------------------------------------------------[版權申明：本文系作者原創，轉載請註明出處] 文章出處：http://blog.csdn.

HBase資料匯入（一）importtsv工具匯入文字檔案到Hbase

剛安裝好Hbase，如果不知道怎麼安裝，請參見我這篇博文，首先想到的就是能夠匯入大量資料，然後查詢玩玩。 怎麼匯入呢，瞭解到可以從文字檔案匯入，那就先測試一下吧，在這之前先要配置一下Hadoop。 配置步驟： 1 首先要修改Hadoop的配置檔案hadoop-env.s

使用mapreduce 將hdfs中的資料匯入到到hbase 中

package hbase; import java.text.SimpleDateFormat; import java.util.Date; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase

flume將資料匯入hbase

1 將hbase的lib目錄下jar拷貝到flume的lib目錄下；2 在hbase中建立儲存資料的表hbase(main):002:0> create 'test_idoall_org','uid','name'3 建立flume配置檔案 vi.confa1.sour