windows 远程连接 Hadoop 2.7.2集群 开发环境搭建

1.环境准备

1.1 下载 hadoop安装包,hadoop.dll和winutils

因为hadoop 都是在linux环境开发,在windows 开发需要插件的支持

1. 下载与hadoop对应版本的hadoop.dll和winutils
2. 将hadoop.dll和winutils 添加到 hadoop安装目录下的bin目录下 eg:\Hadoop\hadoop-2.7.2\bin
3. hadoop.dll和winutils 下载地址：

• gitHub:https://github.com/cdarlint/winutils
• hadoop2.7.2(已经包含hadoop.dll和winutils )
  • 链接：https://pan.baidu.com/s/1u5Nv00BiodfTG9b87M3NPw
  • 提取码：03nl

1.2 配置环境变量

HADOOP_HOME=hadoop安装目录

将bin目录配置到path环境变量内 %HADOOP_HOME%

2.HDFS 客户端编写

2.1 pom.xml

    <properties>
        <hadoop.version>2.7.2</hadoop.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>jdk.tools</groupId>
            <artifactId>jdk.tools</artifactId>
            <version>1.8</version>
        </dependency>
    </dependencies>

2.2 log4j.properties

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

2.2 文件上传测试类

/**
     * 上传
     *
     * @throws IOException
     */
    @Test
    public void upload() throws IOException {

        // 1.创建配置文件读取对象
        Configuration conf = new Configuration();
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000/");
        // 2.创建 文件系统对象，相当于 hdfs 客户端
        FileSystem fs = FileSystem.get(conf);

        Path dest = new Path("hdfs://hadoop01:9000/test_work/test01.txt");
        FSDataOutputStream os = fs.create(dest);

        FileInputStream is = new FileInputStream("d:/test01.txt");

        IOUtils.copy(is, os);

        is.close();
        os.close();
    }

2.运行程序

可能报错：
org.apache.hadoop.security.AccessControlException: Permission denied: user=Admin, access=WRITE, inode="/test_work/test01.txt":qg:supergroup:drwxr-xr-x

客户端去操作HDFS时，是有一个用户身份的。默认情况下，HDFS客户端API会从JVM中获取一个参数来作为自己的用户身份(windows 用户环境变量)：-DHADOOP_USER_NAME=qg，qg为远程登录用户名称。

2.1 方法一: 指定用户

 FileSystem fs = FileSystem.get(new URI("hdfs://hadoop01:9000/"),conf,"qg");

2.1 方法二: 配置 jvm 参数

eclipse: RunAs–> Run configurations…

Idea:

3.一路踩坑

因为集群在linux上搭建,在windows上开发难免会有一些问题

可能遇到的问题:

• 是否下载hadoop.dl,winutils.exe
• 找不到tools.jar
• 需是否配置 hostname
• 端口没有开放
• 需要验证身份
• hdfs 系统uri 是否正确

2.1 下载 hadoop.dl,winutils.exe

java.io.FileNotFoundException: Could not locate Hadoop executable: E:\hadoop-2.8.0\bin\winutils.exe

Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

2.2 将tools.jar安装到maven 本地仓库

maven 在本地找不到tools.jar

mvn install:install-file -Dfile=D:\tools.jar    -DgroupId=jdk.tools -DartifactId=jdk.tools -Dversion=1.8 -Dpackaging=jar

mvn install:install-file
-Dfile=tools.jar所在路径,在jdk安装目录/lib目录下
-DgroupId=jdk.tools
-DartifactId=jdk.tools
-Dversion=1.8
-Dpackaging=jar

2.3 配置客户端用hostname方式访问集群

外域机器通信需要用外网ip，未配置hostname访问会访问异常

java 客户端:

Configuration conf=new Configuration();
conf.set("dfs.client.use.datanode.hostname", "true");

配置文件方式：hdfs-site.xml

<property>
    <name>dfs.client.use.datanode.hostname</name>
    <value>true</value>
    <description>only cofig in clients</description>
 </property>

2.4 端口开放

ERROR [org.apache.hadoop.util.Shell] - Failed to locate the winutils binary in the hadoop binary path

/test01.txt._COPYING_ could only be replicated to 0 nodes instead of minReplication (=1). There are 2 datanode(s) running and 2 node(s) are excluded in this operation.

…
windows 主机访问集群,是通过ip+端口访问

• 关闭防火墙
• 开放hadoop 所需要的端口

组件	节点	默认端口	配置	用途说明
HDFS	DataNode	50010	dfs.datanode.address	datanode服务端口，用于数据传输
HDFS	DataNode	50075	dfs.datanode.http.address	http服务的端口
DFS	DataNode	50475	dfs.datanode.https.address	https服务的端口
HDFS	DataNode	50020	dfs.datanode.ipc.address	ipc服务的端口
HDFS	NameNode	50070	dfs.namenode.http-address	http服务的端口
HDFS	NameNode	50470	dfs.namenode.https-address	https服务的端口
HDFS	NameNode	8020	fs.defaultFS	接收Client连接的RPC端口，用于获取文件系统metadata信息。
HDFS	journalnode	8485	dfs.journalnode.rpc-address	RPC服务
HDFS	journalnode	8480	dfs.journalnode.http-address	HTTP服务
HDFS	ZKFC	8019	dfs.ha.zkfc.port	ZooKeeper FailoverController，用于NN HA
YARN	ResourceManager	8032	yarn.resourcemanager.address	RM的applications manager(ASM)端口
YARN	ResourceManager	8030	yarn.resourcemanager.scheduler.address	scheduler组件的IPC端口
YARN	ResourceManager	8031	yarn.resourcemanager.resource-tracker.address	IPC
YARN	ResourceManager	8033	yarn.resourcemanager.admin.address	IPC
YARN	ResourceManager	8088	yarn.resourcemanager.webapp.address	http服务端口
YARN	NodeManager	8040	yarn.nodemanager.localizer.address	localizer IPC
YARN	NodeManager	8042	yarn.nodemanager.webapp.address	http服务端口
YARN	NodeManager	8041	yarn.nodemanager.address	NM中container manager的端口
YARN	JobHistory Server	10020	mapreduce.jobhistory.address	IPC
YARN	JobHistory Server	19888	mapreduce.jobhistory.webapp.address	http服务端口
HBase	Master	60000	hbase.master.port	IPC
HBase	Master	60010	hbase.master.info.port	http服务端口
HBase	RegionServer	60020	hbase.regionserver.port	IPC
HBase	RegionServer	60030	hbase.regionserver.info.port	http服务端口
HBase	HQuorumPeer	2181	hbase.zookeeper.property.clientPort	HBase-managed ZK mode，使用独立的ZooKeeper集群则不会启用该端口。
HBase	HQuorumPeer	2888	hbase.zookeeper.peerport	HBase-managed ZK mode，使用独立的ZooKeeper集群则不会启用该端口。
HBase	HQuorumPeer	3888	hbase.zookeeper.leaderport	HBase-managed ZK mode，使用独立的ZooKeeper集群则不会启用该端口。
Hive	Metastore	9083	/etc/default/hive-metastore中export PORT=来更新默认端口
Hive	HiveServer	10000	/etc/hive/conf/hive-env.sh中export HIVE_SERVER2_THRIFT_PORT=来更新默认端口
ZooKeeper	Server	2181	/etc/zookeeper/conf/zoo.cfg中clientPort=	对客户端提供服务的端口
ZooKeeper	Server	2888	/etc/zookeeper/conf/zoo.cfg中server.x=[hostname]:nnnnn[:nnnnn]，标蓝部分	follower用来连接到leader，只在leader上监听该端口。
ZooKeeper	Server	3888	/etc/zookeeper/conf/zoo.cfg中server.x=[hostname]:nnnnn[:nnnnn]，标蓝部分	用于leader选举的。只在electionAlg是1,2或3(默认)时需要。

2.5 完整测试代码

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

import java.net.URI;

public class HdfsClientTest {

    FileSystem fs = null;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        /**
         *  获取配置对象：
         *
         *    1.读取 classpath: 下的 xxx-site.xml 将配置封装进 configuration 对象中
         *    2.手动设置属性值 key-value
         *       手动设置优先级高,会覆盖掉 从配置文件中读取的值
         */
        Configuration conf = new Configuration();

        //外域机器通信需要用外网ip，未配置hostname访问会访问异常
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000/");

        // 2.创建 文件系统对象，相当于 hdfs 客户端
        //FileSystem fs = FileSystem.get(conf);
        fs = FileSystem.get(new URI("hdfs://hadoop01:9000/"),conf,"qg");
    }


    /**
     * 上传
     *    较为底层的方法
     * @throws IOException
     */
    @Test
    public void upload() throws IOException, URISyntaxException, InterruptedException {

        Path dest = new Path("hdfs://hadoop01:9000/test_work/test02.txt");
        FSDataOutputStream os = fs.create(dest);

        FileInputStream is = new FileInputStream("d:/test01.txt");

        IOUtils.copy(is, os);

        is.close();
        os.close();

    }

    /**
     *  使用hdfs封装好的方法
     * @throws IOException
     */
    @Test
    public void upload02() throws IOException {

        // 源文件   目标文件
        fs.copyFromLocalFile(new Path("d:/test01.txt"), new Path("hdfs://hadoop01:9000/test_work/test03.txt"));
    }

    /**
     * 下载:
     *   较为底层的写法
     */
    @Test
    public void download() throws IOException {

        // 获取hdfs  输入流
        Path src = new Path("hdfs://hadoop01:9000/test_work/test02.txt");
        FSDataInputStream is = fs.open(src);

        // 获取本地文件系统输出流
        FileOutputStream os = new FileOutputStream("d:/test02.txt");

        // 向本地文件写入数据
        IOUtils.copy(is, os);

        // 关闭资源
        is.close();;
        os.close();
    }

    /**
     *  使用hdfs 封装好的方法
     * @throws IOException
     */
    @Test
    public void download02() throws IOException {
        fs.copyToLocalFile(new Path("hdfs://hadoop01:9000/test_work/test02.txt"), new Path("d:/test03.txt"));
    }

    /**
     *   创建文件夹
     * @throws IOException
     */
    @Test
    public void mkdir() throws IOException {
        // 可以创建多级目录  /: hdfs://hadoop01:9000
        fs.mkdirs(new Path("/aa/bb/cc"));
    }

    /**
     * 删除文件
     */
    @Test
    public void rmdir() throws IOException {
        // delete(文件路径,是否递归删除文件下的数据(true|false))
        fs.delete(new Path("/aa"), true);
    }

    /**
     *  查看文件信息   fs.listFiles
     *     ----只看文件，忽略目录，能把目录深层次的文件列出
     */
    @Test
    public void catFile() throws IOException {

        // 获取 hdfs 文件系统下的所有文件(忽略目录，包括深层次目录的文件列出)
        RemoteIterator<LocatedFileStatus> locatedFileStatus = fs.listFiles(new Path("/"), true);

        // 迭代
        while(locatedFileStatus.hasNext()){
            // 获取文件对象
            LocatedFileStatus file = locatedFileStatus.next();

            Path filePath = file.getPath();  //获取文件全路径
            String filePathName = filePath.getName(); //获取文件名

            System.out.println(filePathName);
        }
    }


    /**
     *  fs.listStatus
     *  查看文件信息(只查看第一层,深层次需自己递归遍历所有目录)
     */
    @Test
    public void catFile02() throws IOException {
        //
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            iterationAllFile(fileStatus);
        }
    }

    // 递归遍历目录
    private void iterationAllFile(FileStatus fileStatus) throws IOException {
        // 判断是不是 文件夹
        if(fileStatus.isDirectory()){
            System.out.println("我是文件夹:" +  fileStatus.getPath());
            FileStatus[] fileStatuses = fs.listStatus(new Path(String.valueOf(fileStatus.getPath())));
            for (FileStatus status : fileStatuses) {
                iterationAllFile(status);
            }
        }else{
            System.out.println("我是文件:" + fileStatus.getPath().getName());
        }
    }

    
}