java结合HADOOP集群文件上传下载
对HDFS上的文件进行上传和下载是对集群的基本操作，在《HADOOP权威指南》一书中，对文件的上传和下载都有代码的实例，但是对如何配置HADOOP客户端却是没有讲得很清楚，经过长时间的搜索和调试，总结了一下，如何配置使用集群的方法，以及自己测试可用的对集群上的文件进行操作的程序。首先，需要配置对应的环境变量:
hadoop_HOME="/home/work/tools/java/hadoop-client/hadoop"
forfin$hadoop_HOME/hadoop-*.jar;do
hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done
forfin$hadoop_HOME/lib/*.jar;do
hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done
hadoopvfs_HOME="/home/work/tools/java/hadoop-client/hadoop-vfs"
forfin$hadoopvfs_HOME/lib/*.jar;do
hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done
exportLD_LIBRARY_PATH=$LD_LIBRARY_PATH:/home/work/tools/java/hadoop-client/hadoop/lib/native/Linux-amd64-64/
其中LD_LIBRARY_PATH是在调用时需要用到的库的路径，hadoop_CLASSPATH则是我们hadoop客户端里各种jar包
有一点需要注意的是最好不要使用HADOOP_HOME这个变量，这个是一个系统使用的环境变量，最好不要和它冲突
编译类的方法：
javac-classpath$CLASSPATH:$hadoop_CLASSPATHHDFSUtil.java
运行的方法：
java-classpath$CLASSPATH:$hadoop_CLASSPATHHDFSUtil
但是在实际的使用过程中，会报NoPermission之类的错误，或者你能保证代码没有问题的情况下，在运行的时候也会报一些奇奇怪怪的错误
那么问题来了，这是什么鬼？
答案：这是因为没有配置对应集群的配置文件
因为在《HADOOP权威指南》一书中，弱化了配置的东西，所以在具体使用集群的时候就会出现问题，如何解决呢，这样子：
this.conf=newConfiguration(false);
conf.addResource("./hadoop-site.xml");
conf.addResource("./hadoop-default.xml");
conf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());conf.set("fs.file.impl",org.apache.hadoop.fs.LocalFileSystem.class.getName());
为什么会这样，书上只是很简单的：
this.conf=newConfiguration();
那是因为默认你的集群在本地，所以不需要做配置，但是在实际使用的过程中，各个集群的配置是不同的，所以我们要引入集群的配置
这是非常重要的一点，因为实际使用的过程中我们都是使用的HADOOP的客户端，而且是已经搭好环境的集群，所以我们需要做好本地的配置
hadoop-site.xml和hadoop-default.xml这两个文件在所使用的客户端的conf目录下，在addResource的时候指定好目录就行了
将以上所提到的配置，全部配完之后，这个程序才能真正运行起来，所以配置是非常重要的一环。
以下是对应的工具的代码，有兴趣的看一下吧，使用的是文件流的方式来搞的，这样子也可以打通FTP和HDFS之间文件的互传：
importjava.io.BufferedInputStream;
importjava.io.FileInputStream;
importjava.io.FileNotFoundException;
importjava.io.FileOutputStream;
importjava.io.IOException;
importjava.io.InputStream;
importjava.io.OutputStream;
importjava.net.URI;
importjava.net.URL;
importjava.io.*;
importorg.apache.hadoop.conf.Configuration;
importorg.apache.hadoop.fs.FSDataInputStream;
importorg.apache.hadoop.fs.FileSystem;
importorg.apache.hadoop.fs.Path;
importorg.apache.hadoop.io.IOUtils;
importorg.apache.hadoop.util.Progressable;
publicclassHDFSUtil{
privateStringhdfs_node="";
privateStringhdfs_path="";
privateStringfile_path="";
privateStringhadoop_site="";
privateStringhadoop_default="";
privateConfigurationconf=null;
publicHDFSUtil(Stringhdfs_node){
this.hdfs_node=hdfs_node;
}
publicStringgetHdfsNode(){
returnthis.hdfs_node;
}
publicvoidsetHdfsPath(Stringhdfs_path){
this.hdfs_path=hdfs_path;
}
publicStringgetHdfsPath(){
returnthis.hdfs_path;
}
publicvoidsetFilePath(Stringfile_path){
this.file_path=file_path;
}
publicStringgetFilePath(){
returnthis.file_path;
}
publicvoidsetHadoopSite(Stringhadoop_site){
this.hadoop_site=hadoop_site;
}
publicStringgetHadoopSite(){
returnthis.hadoop_site;
}
publicvoidsetHadoopDefault(Stringhadoop_default){
this.hadoop_default=hadoop_default;
}
publicStringgetHadoopDefault(){
returnthis.hadoop_default;
}
publicintsetConfigure(booleanflag){
if(flag==false){
if(this.getHadoopSite()==""||this.getHadoopDefault()==""){
return-1;
}
else{
this.conf=newConfiguration(false);
conf.addResource(this.getHadoopDefault());
conf.addResource(this.getHadoopSite());
conf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
conf.set("fs.file.impl",org.apache.hadoop.fs.LocalFileSystem.class.getName());
return0;
}
}
this.conf=newConfiguration();
return0;
}
publicConfigurationgetConfigure(){
returnthis.conf;
}
publicintupLoad(StringlocalName,StringremoteName)throwsFileNotFoundException,IOException{
InputStreaminStream=null;
FileSystemfs=null;
try{
inStream=newBufferedInputStream(newFileInputStream(localName));
fs=FileSystem.get(URI.create(this.hdfs_node),this.conf);
OutputStreamoutStream=fs.create(newPath(remoteName),newProgressable(){
publicvoidprogress(){
System.out.print('.');
}
});
IOUtils.copyBytes(inStream,outStream,4096,true);
inStream.close();
return0;
}catch(IOExceptione){
inStream.close();
e.printStackTrace();
return-1;
}
}
publicintupLoad(InputStreaminStream,StringremoteName)throwsFileNotFoundException,IOException{
FileSystemfs=null;
try{
fs=FileSystem.get(URI.create(this.hdfs_node),this.conf);
OutputStreamoutStream=fs.create(newPath(remoteName),newProgressable(){
publicvoidprogress(){
System.out.print('.');
}
});
IOUtils.copyBytes(inStream,outStream,4096,true);
inStream.close();
return0;
}catch(IOExceptione){
inStream.close();
e.printStackTrace();
return-1;
}
}
publicintdonwLoad(StringremoteName,StringlocalName,intlines)throwsFileNotFoundException,IOException{
FileOutputStreamfos=null;
InputStreamReaderisr=null;
BufferedReaderbr=null;
Stringstr=null;
OutputStreamWriterosw=null;
BufferedWriterbuffw=null;
PrintWriterpw=null;
FileSystemfs=null;
InputStreaminStream=null;
try{
fs=FileSystem.get(URI.create(this.hdfs_node+remoteName),this.conf);
inStream=fs.open(newPath(this.hdfs_node+remoteName));
fos=newFileOutputStream(localName);
osw=newOutputStreamWriter(fos,"UTF-8");
buffw=newBufferedWriter(osw);
pw=newPrintWriter(buffw);
isr=newInputStreamReader(inStream,"UTF-8");
br=newBufferedReader(isr);
while((str=br.readLine())!=null&&lines>0){
lines--;
pw.println(str);
}
}catch(IOExceptione){
thrownewIOException("Couldn'twrite.",e);
}finally{
pw.close();
buffw.close();
osw.close();
fos.close();
inStream.close()
}
return0;
}
//maintotest
publicstaticvoidmain(String[]args){
Stringhdfspath=null;
Stringlocalname=null;
Stringhdfsnode=null;
intlines=0;
if(args.length==4){
hdfsnode=args[0];
hdfspath=args[1];
localname=args[2];
lines=Integer.parseInt(args[3]);
}
else{
hdfsnode="hdfs://nj01-nanling-hdfs.dmop.baidu.com:54310";
hdfspath="/app/ps/spider/wdmqa/wangweilong/test/HDFSUtil.java";
localname="/home/work/workspace/project/dhc2-0/dhc/base/ftp/papapa";
lines=5;
}
HDFSUtilhdfsutil=newHDFSUtil(hdfsnode);
hdfsutil.setFilePath(hdfsutil.getHdfsNode()+hdfspath);
hdfsutil.setHadoopSite("./hadoop-site.xml");
hdfsutil.setHadoopDefault("./hadoop-default.xml");
hdfsutil.setConfigure(false);
try{
hdfsutil.donwLoad(hdfspath,localname,lines);
}catch(IOExceptione){
e.printStackTrace();
}
}
如果想要了解FTP上文件的下载，请参考这篇文章：
ftp下载工具
如果想要打通FTP和HDFS文件互传，只要创建一个类，调用这两篇文章中的工具的接口就可以搞定，自己写的代码，实测有效。
以上就是本文的全部内容了，希望能够对大家熟练掌握java有所帮助。
请您花一点时间将文章分享给您的朋友或者留下评论。我们将会由衷感谢您的支持！