HADOOP集群文件上傳下載

nevermind 發(fā)布于2019-08-14 12:05 / 2315人閱讀

摘要：對(duì)上的文件進(jìn)行上傳和下載是對(duì)集群的基本操作，在權(quán)威指南一書中，對(duì)文件的上傳和下載都有代碼的實(shí)例，但是對(duì)如何配置客戶端卻是沒(méi)有講得很清楚，經(jīng)過(guò)長(zhǎng)時(shí)間的搜索和調(diào)試，總結(jié)了一下，如何配置使用集群的方法，以及自己測(cè)試可用的對(duì)集群上的文件進(jìn)行操作的程

對(duì)HDFS上的文件進(jìn)行上傳和下載是對(duì)集群的基本操作，在《HADOOP權(quán)威指南》一書中，對(duì)文件的上傳和下載都有代碼的實(shí)例，但是對(duì)如何配置HADOOP客戶端卻是沒(méi)有講得很清楚，經(jīng)過(guò)長(zhǎng)時(shí)間的搜索和調(diào)試，總結(jié)了一下，如何配置使用集群的方法，以及自己測(cè)試可用的對(duì)集群上的文件進(jìn)行操作的程序。首先，需要配置對(duì)應(yīng)的環(huán)境變量:

hadoop_HOME="/home/work/tools/java/hadoop-client/hadoop"
for f in $hadoop_HOME/hadoop-*.jar; do
        hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done

for f in $hadoop_HOME/lib/*.jar; do
        hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done

hadoopvfs_HOME="/home/work/tools/java/hadoop-client/hadoop-vfs"
for f in $hadoopvfs_HOME/lib/*.jar; do
        hadoop_CLASSPATH=${hadoop_CLASSPATH}:$f
done

export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/home/work/tools/java/hadoop-client/hadoop/lib/native/Linux-amd64-64/

其中LD_LIBRARY_PATH是在調(diào)用時(shí)需要用到的庫(kù)的路徑，hadoop_CLASSPATH則是我們hadoop客戶端里各種jar包
有一點(diǎn)需要注意的是最好不要使用HADOOP_HOME這個(gè)變量，這個(gè)是一個(gè)系統(tǒng)使用的環(huán)境變量，最好不要和它沖突
編譯類的方法：

javac -classpath $CLASSPATH:$hadoop_CLASSPATH HDFSUtil.java

運(yùn)行的方法：

java -classpath $CLASSPATH:$hadoop_CLASSPATH HDFSUtil

但是在實(shí)際的使用過(guò)程中，會(huì)報(bào)No Permission之類的錯(cuò)誤，或者你能保證代碼沒(méi)有問(wèn)題的情況下，在運(yùn)行的時(shí)候也會(huì)報(bào)一些奇奇怪怪的錯(cuò)誤
那么問(wèn)題來(lái)了，這是什么鬼？
答案：這是因?yàn)闆](méi)有配置對(duì)應(yīng)集群的配置文件
因?yàn)樵凇禜ADOOP權(quán)威指南》一書中，弱化了配置的東西，所以在具體使用集群的時(shí)候就會(huì)出現(xiàn)問(wèn)題，如何解決呢，這樣子：

this.conf = new Configuration(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());

為什么會(huì)這樣，書上只是很簡(jiǎn)單的：

this.conf = new Configuration();

那是因?yàn)槟J(rèn)你的集群在本地，所以不需要做配置，但是在實(shí)際使用的過(guò)程中，各個(gè)集群的配置是不同的，所以我們要引入集群的配置
這是非常重要的一點(diǎn)，因?yàn)閷?shí)際使用的過(guò)程中我們都是使用的HADOOP的客戶端，而且是已經(jīng)搭好環(huán)境的集群，所以我們需要做好本地的配置
hadoop-site.xml和hadoop-default.xml這兩個(gè)文件在所使用的客戶端的conf目錄下，在addResource的時(shí)候指定好目錄就行了

將以上所提到的配置，全部配完之后，這個(gè)程序才能真正運(yùn)行起來(lái)，所以配置是非常重要的一環(huán)。

以下是對(duì)應(yīng)的工具的代碼，有興趣的看一下吧，使用的是文件流的方式來(lái)搞的，這樣子也可以打通FTP和HDFS之間文件的互傳：

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URL;
import java.io.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

public class HDFSUtil {
    private String hdfs_node = "";
    private String hdfs_path = "";
    private String file_path = "";
    private String hadoop_site = "";
    private String hadoop_default = "";
    private Configuration conf = null;

    public HDFSUtil(String hdfs_node) {
        this.hdfs_node = hdfs_node;
    }

    public String getHdfsNode() {
        return this.hdfs_node;
    }

    public void setHdfsPath(String hdfs_path){
        this.hdfs_path = hdfs_path;
    }

    public String getHdfsPath(){
        return this.hdfs_path;
    }

    public void setFilePath(String file_path){
        this.file_path = file_path;
    }

    public String getFilePath(){
        return this.file_path;
    }

    public void setHadoopSite(String hadoop_site){
        this.hadoop_site = hadoop_site;
    }

    public String getHadoopSite(){
        return this.hadoop_site;
    }

    public void setHadoopDefault(String hadoop_default){
        this.hadoop_default = hadoop_default;
    }

    public String getHadoopDefault(){
        return this.hadoop_default;
    }

    public int setConfigure(boolean flag) {
        if (flag == false){
            if (this.getHadoopSite() == "" || this.getHadoopDefault() == ""){
                return -1;
            }
            else {
                this.conf = new Configuration(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());
                return 0;
            }
        }
        this.conf = new Configuration();
        return 0;
    }

    public Configuration getConfigure() {
        return this.conf;
    }

    public int upLoad(String localName, String remoteName) throws FileNotFoundException, IOException {
        InputStream inStream = null;
        FileSystem fs = null;
        try{
            inStream = new BufferedInputStream(new FileInputStream(localName));
            fs = FileSystem.get(URI.create(this.hdfs_node), this.conf);
            OutputStream outStream = fs.create(new Path(remoteName) ,new Progressable() {
                public void progress(){
                    System.out.print(".");
                }
            });

            IOUtils.copyBytes(inStream, outStream, 4096, true);
            inStream.close();
            return 0;
        } catch (IOException e){
            inStream.close();
            e.printStackTrace();
            return -1;
        }
    }

    public int upLoad(InputStream inStream, String remoteName) throws FileNotFoundException, IOException {
        FileSystem fs = null;
        try{
            fs = FileSystem.get(URI.create(this.hdfs_node), this.conf);
            OutputStream outStream = fs.create(new Path(remoteName) ,new Progressable() {
                public void progress(){
                    System.out.print(".");
                }
            });

            IOUtils.copyBytes(inStream, outStream, 4096, true);
            inStream.close();
            return 0;
        } catch (IOException e){
            inStream.close();
            e.printStackTrace();
            return -1;
        }
    }

    public int donwLoad(String remoteName, String localName, int lines) throws FileNotFoundException, IOException {
        FileOutputStream fos = null;
        InputStreamReader isr = null;
        BufferedReader br = null;
        String str = null;
        OutputStreamWriter osw = null;
        BufferedWriter buffw = null;
        PrintWriter pw = null;
        FileSystem fs = null;
        InputStream inStream = null;
        try {
            fs = FileSystem.get(URI.create(this.hdfs_node + remoteName), this.conf);
            inStream = fs.open(new Path(this.hdfs_node + remoteName));
            fos = new FileOutputStream(localName);
            osw = new OutputStreamWriter(fos, "UTF-8");
            buffw = new BufferedWriter(osw);
            pw = new PrintWriter(buffw);
            isr = new InputStreamReader(inStream, "UTF-8");
            br = new BufferedReader(isr);
            while((str = br.readLine()) != null && lines > 0){
                lines--;
                pw.println(str);
            }
        } catch (IOException e){
            throw new IOException("Couldn"t write.", e);
        } finally {
            pw.close();
            buffw.close();
            osw.close();
            fos.close();
            inStream.close()
        }
        return 0;
    }

    //main to test
    public static void main(String[] args){
        String hdfspath = null;
        String localname = null;
        String hdfsnode = null;
        int lines = 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;
        }
        HDFSUtil hdfsutil = new HDFSUtil(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 (IOException e){
            e.printStackTrace();
        }
    }

如果想要了解FTP上文件的下載，請(qǐng)參考這篇文章：
ftp下載工具

如果想要打通FTP和HDFS文件互傳，只要?jiǎng)?chuàng)建一個(gè)類，調(diào)用這兩篇文章中的工具的接口就可以搞定，自己寫的代碼，實(shí)測(cè)有效。