这篇文章主要讲解了“hbase怎么在不同版本hdfs集群之间转移数据”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“hbase怎么在不同版本hdfs集群之间转移数据”吧！
十余年的监利网站建设经验，针对设计、前端、开发、售后、文案、推广等六对一服务，响应快，48小时及时工作处理。成都营销网站建设的优势是能够根据用户设备显示端的尺寸不同，自动调整监利建站的显示方式，使网站能够适用不同显示终端，在浏览器中调整网站的宽度，无论在任何一种浏览器上浏览网站，都能展现优雅布局与设计，从而大程度地提升浏览体验。创新互联建站从事“监利网站设计”,“监利网站推广”以来，每个客户项目都认真落实执行。
最简单的办法就是把src集群的数据导到本地，然后起另一个进程将本地数据传到des集群上去。
不过这有几个问题：
效率降低
占用本地磁盘空间
不能应付实时导数据需求
两个进程需要协调，复杂度增加
更好的办法是在同一个进程内一边读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];
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();
但是由于在生成HTable对象时，需要使用这个conf对象，而加载这个conf对象的代码本身是由默认的classloader加载的，也就是0.19.2的jar包。所以在以上代码最后一行所强制转换的Configuration对象仍然是0.19.2版本的。那怎么办呢？
琢磨了一会，发现如果要实现以上功能，必须将生成HTable对象，以及以后的所有hbase操作都使用这个新的classloader，因此这个新的classloader必须加载除了0.19.2的jar包外所有需要用到的jar包，然后把所有操作都封装进去。在外面用反射来调用。
这样的话，通常构造函数都不为空了，因此需要用到Constructor来构造一个自定义的构造函数
代码段如下：
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);
}
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);
}
总之，在同一个进程中加载多个classloader时一定要注意，classloader A所加载的对象是不能转换成classloader B的对象的，当然也不能使用。两个空间的相互调用只能用java的基本类型或是反射。
感谢各位的阅读，以上就是“hbase怎么在不同版本hdfs集群之间转移数据”的内容了，经过本文的学习后，相信大家对hbase怎么在不同版本hdfs集群之间转移数据这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！