hbase--在不一樣版本hdfs集羣之間轉移數據

時間 2019-11-24

標籤 hbase 不一樣版本 hdfs 集羣之間轉移數據欄目 Hadoop 简体版

原文原文鏈接

不少人會有這樣一個需求：將一個hdfs集羣上的數據寫入另外一個hdfs集羣所在的hbase數據庫。一般狀況下兩個hdfs集羣的版本差距並不大，這樣的程序會很容易寫。但有時會跨大版本。好比做者所在的廠子，數據都在基於hadoop0.19.2版本修改的hdfs集羣上，要將這樣的數據導入版本爲0.20.2+的hdfs集羣，就不能使用同一個hadoop jar包來完成了。如何實現呢？

最簡單的辦法就是把src集羣的數據導到本地，而後起另外一個進程將本地數據傳到des集羣上去。
不過這有幾個問題： java

效率下降
佔用本地磁盤空間
不能應付實時導數據需求
兩個進程須要協調，複雜度增長

更好的辦法是在同一個進程內一邊讀src數據，一邊寫des集羣。不過這至關於在同一個進程空間內加載兩個版本的hadoop jar包，這就須要在程序中使用兩個classloader來實現。
如下代碼能夠實現classloader加載自定義的jar包，並生成須要的Configuration對象：數據庫

Java代碼微信

URL[] jarUrls = new URL[1];
jarUrls[0]=new File(des_jar_path).toURI().toURL();
ClassLoader jarloader = new URLClassLoader(jarUrls, null);
Class Proxy = Class.forName("yourclass", true, jarloader);
Configuration conf = (Configuration)Proxy.newInstance();

URL[] jarUrls = new URL[1]; ide

jarUrls[0]=new File(des_jar_path).toURI().toURL(); 函數

ClassLoader jarloader = new URLClassLoader(jarUrls, null); oop

Class Proxy = Class.forName("yourclass", true, jarloader); spa

Configuration conf = (Configuration)Proxy.newInstance(); code

    可是因爲在生成HTable對象時，須要使用這個conf對象，而加載這個conf對象的代碼自己是由默認的classloader加載的，也就是0.19.2的jar包。因此在以上代碼最後一行所強制轉換的Configuration對象仍然是0.19.2版本的。那怎麼辦呢？
    琢磨了一會，發現若是要實現以上功能，必須將生成HTable對象，以及之後的全部hbase操做都使用這個新的classloader，所以這個新的classloader必須加載除了0.19.2的jar包外全部須要用到的jar包，而後把全部操做都封裝進去。在外面用反射來調用。
    這樣的話，一般構造函數都不爲空了，所以須要用到Constructor來構造一個自定義的構造函數
    代碼段以下： orm

Java代碼對象

main.java
void init(){
ClassLoader jarloader = generateJarLoader();
Class Proxy = Class.forName("test.writer.hbasewriter.HBaseProxy", true, jarloader);
Constructor con = Proxy.getConstructor(new Class[]{String.class, String.class, boolean.class});
Boolean autoflush = param.getBoolValue(ParamsKey.HbaseWriter.autoFlush, true);
proxy = con.newInstance(new Object[]{path, tablename, autoflush});
}
void put(){
...
while((line = getLine()) != null) {
proxy.getClass().getMethod("generatePut",String.class).invoke(proxy, line.getField(rowkey));
Method addPut = proxy.getClass().getMethod("addPut",
new Class[]{String.class, String.class, String.class});
addPut.invoke(proxy, new Object[]{field, column, encode});
proxy.getClass().getMethod("putLine").invoke(proxy);
}
}
ClassLoader generateJarLoader() throws IOException {
String libPath = System.getProperty("java.ext.dirs");
FileFilter filter = new FileFilter() {
@Override
public boolean accept(File pathname) {
if(pathname.getName().startsWith("hadoop-0.19.2"))
return false;
else
return pathname.getName().endsWith(".jar");
}
};
File[] jars = new File(libPath).listFiles(filter);
URL[] jarUrls = new URL[jars.length+1];
int k = 0;
for (int i = 0; i < jars.length; i++) {
jarUrls[k++] = jars[i].toURI().toURL();
}
jarUrls[k] = new File("hadoop-0.20.205.jar")
ClassLoader jarloader = new URLClassLoader(jarUrls, null);
return jarloader;
}

main.java

void init(){

ClassLoader jarloader = generateJarLoader();

Class Proxy = Class.forName("test.writer.hbasewriter.HBaseProxy", true, jarloader);

Constructor con = Proxy.getConstructor(new Class[]{String.class, String.class, boolean.class});

Boolean autoflush = param.getBoolValue(ParamsKey.HbaseWriter.autoFlush, true);

proxy = con.newInstance(new Object[]{path, tablename, autoflush});

}

void put(){

...

while((line = getLine()) != null) {

proxy.getClass().getMethod("generatePut",String.class).invoke(proxy, line.getField(rowkey));

Method addPut = proxy.getClass().getMethod("addPut",

new Class[]{String.class, String.class, String.class});

addPut.invoke(proxy, new Object[]{field, column, encode});

proxy.getClass().getMethod("putLine").invoke(proxy);

}

ClassLoader generateJarLoader() throws IOException {

String libPath = System.getProperty("java.ext.dirs");

FileFilter filter = new FileFilter() {

@Override

public boolean accept(File pathname) {

if(pathname.getName().startsWith("hadoop-0.19.2"))

return false;

else

return pathname.getName().endsWith(".jar");

}

};

File[] jars = new File(libPath).listFiles(filter);

URL[] jarUrls = new URL[jars.length+1];

int k = 0;

for (int i = 0; i < jars.length; i++) {

jarUrls[k++] = jars[i].toURI().toURL();

}

jarUrls[k] = new File("hadoop-0.20.205.jar")

ClassLoader jarloader = new URLClassLoader(jarUrls, null);

return jarloader;

}

Java代碼

HBaseProxy.java
public HBaseProxy(String hbase_conf, String tableName, boolean autoflush)
throws IOException{
Configuration conf = new Configuration();
conf.addResource(new Path(hbase_conf));
config = new Configuration(conf);
htable = new HTable(config, tableName);
admin = new HBaseAdmin(config);
htable.setAutoFlush(autoflush);
}
public void addPut(String field, String column, String encode) throws IOException {
try {
p.add(column.split(":")[0].getBytes(), column.split(":")[1].getBytes(),
field.getBytes(encode));
} catch (UnsupportedEncodingException e) {
p.add(column.split(":")[0].getBytes(), column.split(":")[1].getBytes(),
field.getBytes());
}
}
public void generatePut(String rowkey){
p = new Put(rowkey.getBytes());
}
public void putLine() throws IOException{
htable.put(p);
}