使用Hadoop JAVA API 远程连接HDFS

0x00 前言

关于hadoop的搭建，可以参考之前的一篇文章Centos7搭建Hadoop伪分布式集群详细步骤
本篇主要讲解如何使用Hadoop java api远程连接hadoop分布式文件系统（HDFS）

0x01 本地操作HDFS

在远程连接HDFS之前，先本地操作一波，便于对HDFS有个更深层的认识。

首先应该对HDFS的概念有个认识，他就是一个文件系统，只不过是适应于hadoop的分布式文件系统，既然是文件系统，其就可以执行跟普通文件系统一样的操作，例如文件操作，目录操作等。
下面演示几个简单的例子。

列出目录文件
hadoop fs -ls \
创建文件
hadoop fs -touchz /hello

更多命令参考hadoop HDFS常用文件操作命令

0x02 IDEA远程连接HDFS

准备环境

首先在windows环境下安装hadoop并配置环境变量。
windows下配置java jdk。
下载winutils：https://github.com/steveloughran/winutils
按照自己hadoop版本选择hadoop.dll winutils.exe放到hadoop bin目录下

建立工程

在IDEA下新建Maven工程

创建过程中右下角会出现import提醒，点击import。

创建完成后下载对应版本的hadoop pom.xml文件
这里我的为

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>3.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-core</artifactId>
        <version>3.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>3.2.0</version>
    </dependency>
</dependencies>

将之前配置的core-site.xml hdfs-site.xml log4j.properties复制到resources目录

目录结构如下

测试

新建class

package hadoop;

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

public class hadoop{
    public static void main(String[] args){
        try{
            String filename = "hdfs://192.168.111.141:9000/user/input/boogle.txt";
            Configuration configuration = new Configuration();
            FileSystem fileSystem = FileSystem.get(configuration);
            if(fileSystem.exists(new Path(filename)))
            {
                System.out.println("文件存在");
            }
            else{
                System.out.println("文件不存在");
            }
            Path inFile = new Path("/user/input/boogle.txt");
            FSDataOutputStream os = fileSystem.create(inFile);
            os.writeUTF("Chinese Hadoop Community");
            os.flush();
            os.close();
            FSDataInputStream is = fileSystem.open(inFile);
            IOUtils.copyBytes(is,System.out,1024,true);
        }catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}

按照自己环境修改一下上面的代码。

之前本地搭建的时候连接hdfs用的localhost，此处远程连接需换成ip。
如我的core-site.xml

查看运行结果

详细代码

常用hdfs java api已封装好，直接在main函数中调用即可。

package hadoop;
        import java.io.IOException;
        import java.net.URI;
        import java.net.URISyntaxException;

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

        import org.apache.hadoop.fs.FileStatus;




public class Hdfs {
    static Configuration conf = new Configuration();

    static FileSystem hdfs;
    public static void init(){

    try{
        hdfs = FileSystem.get(conf);

    } catch(
    IOException e)

    {
        e.printStackTrace();
    }

}
    public static void main(String[] args){


        try {

             listFiles("/user");
             createDir("/user/input");
             createFile("/user/input/test1.txt","HelloWoRLd ,It is 2019");
             createFile("/user/input/test2.txt","hello boogle ,Your Site Is :zhengbao,wang");
             listFiles("/user/input");

        }catch (IOException e){
            e.printStackTrace();
        }



    }
    public static void createDir(String dir) throws IOException {
        init();
        //System.out.println("newdir \t" + conf.get("fs.defaultFS") + dir);
        Path path = new Path(dir);

        hdfs.mkdirs(path);
        System.out.println("newdir \t" + conf.get("fs.defaultFS") + dir);
    }

    //copy from local file to HDFS file
    public static void copyFile(String localSrc,String hdfsDst) throws IOException {
        init();
        Path src = new Path(localSrc);
        Path dst = new Path(hdfsDst);
        hdfs.copyFromLocalFile(src, dst);

        //list all the files in thecurrent direction
        FileStatus files[] = hdfs.listStatus(dst);
        System.out.println("Uploadto \t" + conf.get("fs.defaultFS") + hdfsDst);
        for (FileStatus file : files) {
            System.out.println(file.getPath());
        }
    }
    //create a new file
    public static  void createFile(String fileName,String fileContent) throws IOException {
        init();
        Path dst = new Path(fileName);
        byte[] bytes =fileContent.getBytes();
        FSDataOutputStream output =hdfs.create(dst);
        output.write(bytes);
        System.out.println("newfile \t" + conf.get("fs.defaultFS") + fileName);
    }

    //list all files
    public static void listFiles(String dirName)throws IOException {
        init();
        Path f = new Path(dirName);
        FileStatus[] status =hdfs.listStatus(f);
        System.out.println(dirName +" has all files:");
        for (int i = 0; i<status.length; i++) {
            System.out.println(status[i].getPath().toString());
        }
    }
    public static void deleteFile(String fileName)throws IOException {
        init();
        Path f = new Path(fileName);
        boolean isExists =hdfs.exists(f);
        if (isExists) {      //if exists, delete
            boolean isDel =hdfs.delete(f,true);
            System.out.println(fileName+ "  delete? \t" + isDel);
        } else {
            System.out.println(fileName+ "  exist? \t" + isExists);
        }
    }

}

0x03 问题解决

因为之前hadoop环境已经搭好，容易出现的问题主要是配置IDEA配置问题和网络连接问题。

关于配置没有详细讲，但是要按照每一个步骤自己操作为，并无难度，勿丢步骤即可。
关于网络连接，主要出现的问题在于防火墙禁止连接9000端口可通过修改hosts或者修改防火墙配置