# 0. 免密钥配置( 已经在1配置了免密钥(对其他所有节点) ,需要在2也配置免密钥(对1即可) )
# 在node2中输入以下命令
ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa
cd /root/.ssh/
cat id_dsa.pub >> authorized_keys 
cat authorized_keys 
ssh node2
exit
scp id_dsa.pub node1:`pwd`/node2.pub

# 在node1中输入以下命令
cd /root/.ssh/

# 在node2中验证免密钥是否成功
[root@node2 .ssh]# ssh node1
Last login: Sun Oct 20 01:03:41 2019 from localhost
[root@node1 ~]# exit


# 1.在 /etc/hosts 文件中配置ip地址映射(对应四台主机,图1)
192.168.179.140 node1
192.168.179.141 node2
192.168.179.142 node3
192.168.179.143 node4


# 2. 验证主机名
hostname


# 3.本机免密钥配置, 生成密钥文件
ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa


# 4.在/root目录下,找到生成的密钥
ll -a #找到存放密钥和公钥的.ssh文件(图2)
cd ./ssh 

[root@node1 .ssh]# ll
total 12
-rw-------. 1 root root 668 Oct 19 17:28 id_dsa  #密钥
-rw-r--r--. 1 root root 600 Oct 19 17:28 id_dsa.pub #公钥
-rw-r--r--. 1 root root 794 Oct 19 17:23 known_hosts


# 5,读取/root目录下的公钥文件重定向并追加到/root目录下的authorized_keys文件中
cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

# 执行后,会生成authorized_keys文件
[root@node1 .ssh]# ll
total 16
-rw-r--r--. 1 root root 600 Oct 19 17:36 authorized_keys # 新文件
-rw-------. 1 root root 668 Oct 19 17:28 id_dsa
-rw-r--r--. 1 root root 600 Oct 19 17:28 id_dsa.pub
-rw-r--r--. 1 root root 794 Oct 19 17:23 known_hosts
# 通过对比发现两个文件内容一样
cat id_dsa.pub authorized_keys 


# 6.安装jdk,配置环境变量 ,重载该文件 
-------------------------------
export JAVA_HOME=/usr/java/jdk1.7.0_67
export HADOOP_PREFIX=/opt/chy/hadoop
-------------------------------
source /etc/profile


# 7.利用jps验证jdk是否安装以及环境变量是否配置成功
[root@node1 .ssh]# jps
2604 Jps


# 8.新建安装目录,上传hadoop2.6.5,并解压
tar -zxf hadoop-2.6.5.tar.gz 
mkdir /opt/chy


# 9,为/sbin 管理命令添加环境变量,令其在任何目录下都能够执行
pwd  #复制/sbin所在目录
#在 /etc/profile中添加如下,并重载
------------------------------------------------------------
export HADOOP_PREFIX=/opt/chy/hadoop
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_PREFIX/bin:$HADOOP_PREFIX/sbin
------------------------------------------------------------
.  /etc/profile


# 10.测试是否安装成功(图3)
ha + tab+tab
hdfs + tab+tab
start + tab+tab

# 1. JAVA_HOME的二次配置 ,复制JAVA_HOME的配置文件,都被注释, 需要放开(图4)
vi /opt/chy/hadoop/etc/hadoop/hapoop-env.sh
vi /opt/chy/hadoop/etc/hadoop/mapred-env.sh
vi /opt/chy/hadoop/etc/hadoop/yarn-env.sh


# 2.修改core-site.xml,设置了name在哪里启动, 以及执行本地存放数据的目录为/var/chy/hadoop/local
vi core-site.xml
------------注意主机名以及指定的目录信息---------------------
<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://node1:9000</value>
    </property>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>/var/chy/hadoop/local</value>
    </property>
</configuration>


# 3,配置hdfs-site.xml,修改second namenode在哪里启动
vi hdfs-site.xml
--------------注意主机名以及端口号,这里的端口号是图形化界面的端口号---------------------
<configuration>
    <property>
       <name>dfs.replication</name>
       <value>1</value>
    </property>
    <property>
        <name>dfs.namenode.secondary.http-address</name>
        <value>node1:50090</value>
    </property>
</configuration>


# 4.修改datanode在启动地方
vi slaves
---------内容---------------
node1


# 5. 格式化HDFS ,查看是否成功(图5)
hdfs namenode -format

# 查看本地数据所在目录
[root@node1 ]# cd /var/chy/hadoop/local/dfs/name/current/
[root@node1 current]# ll
total 16
-rw-r--r--. 1 root root 320 Oct 19 20:40 fsimage_0000000000000000000  #第一个初始的fsimage
-rw-r--r--. 1 root root  62 Oct 19 20:40 fsimage_0000000000000000000.md5
-rw-r--r--. 1 root root   2 Oct 19 20:40 seen_txid
-rw-r--r--. 1 root root 206 Oct 19 20:40 VERSION

# 可以看到VERSION中已经含有了NameNode中的Block等相关信息
[root@node1 current]# cat VERSION 
# Sat Oct 19 20:40:20 CST 2019
namespaceID=119524649
clusterID=CID-db364b61-d241-4d38-b370-3425de275ef3
cTime=0
storageType=NAME_NODE
blockpoolID=BP-635787903-192.168.179.140-1571488820328
layoutVersion=-60


# 6.启动管理脚本
start-dfs.sh 

# 查看日志文件
/opt/chy/hadoop/logs/hadoop-root-datanode-node1.out
# 通过jps查看启动是否成功
[root@node1 ~]# jps
1244 NameNode
1503 SecondaryNameNode
1331 DataNode
1947 Jps


# 7.在`/var/chy/hadoop/local/dfs/`目录下 ,我们可以看到有data和name文件,但是在真的集群中, 这两个文件时在不同主机上的
我们第一次初始化后可能又会因为误操作(格式化)导致namenode中的数据被删除而datanode中的数据没有被删除(因为datanode第一次从namenode中获得集群ID就不会改变了,格式化只格式namenode),
而导致二者ID不一致的情况,这样做最终的结果是会导致启动后DataNode消失(如图6,图7)
DataNode上的VERSION文件创建的时机: NameNode启动后, DataNode启动时与NameNode交互得来的 (一次交互, 永不更改 )

# 伪集群下访问如何目录,可以看到DN和NN存放的位置
[root@node1 ~]# cd /var/chy/hadoop/local/dfs/
[root@node1 dfs]# ll
total 12
drwx------. 3 root root 4096 Oct 19 21:41 data
drwxr-xr-x. 3 root root 4096 Oct 19 21:40 name
drwxr-xr-x. 2 root root 4096 Oct 19 21:41 namesecondary

#查看NameNode的VERSION信息
cd /var/chy/hadoop/local/dfs/name/current/
cat VERSION

#查看DataNode的VERSION信息
cd /var/chy/hadoop/local/dfs/date/current/
cat VERSION


# 8.访问图形化界面(图8)
service iptables stop
http://node1:50070/


# 9.操作HDFS ,我们可以通过输入 hdfs dfs后提示可知, hdfs dfs -绝大多数Linux命令

# eg: 
#新建目录
hdfs dfs -mkdir -p /user/root

# 测试文件上传(都可以通过图形化界面管理,图9)
hdfs dfs -put hadoop-2.6.5.tar.gz /user/root
# 并且,由图10可以看出文件(180MB)被为了两块 ,每块id不同

# 1.创建一个文本文件,内容是hello csdn 1-100000,10w行,并上传到HDFS(选择文本的原因是方便我们的查看文本内容)
for i in `seq 100000`;do echo "hello csdn $i" >> testhadoop.txt;done 


# 2.显示文件大小,换算成 M
[root@node1 ~]# ll -h 
-rw-r--r--. 1 root root 1.7M Oct 19 22:50 testhadoop.txt


# 3.设置块大小( 设置块大小图1 https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/hdfs-default.xml )
hdfs dfs -D dfs.blocksize=1048576 -put testhadoop.txt


# 4. 查看被切分的文件
cd /var/chy/hadoop/local/dfs/data/current/BP-206976221-192.168.179.140-1571489736295/current/finalized/subdir0/subdir0/

# 从测试中可以看到, HDFS严格按照字节去切分
[root@node1 subdir0]# vi blk_1073741827
[root@node1 subdir0]# vi blk_1073741828

# 1. 关闭NN,DN,SNN
stop-dfs.sh 


# 2.对其他结点(node2,node3,node4)安装jdk
# 可通过Xshell的全部会话功能同时对三个节点进行安装
rpm -ivh jdk-7u67-linux-x64.rpm
vi /etc/prvi /etc/profile
----------------------环境变量,注意修改成自己的-------------------------------
export JAVA_HOME=/usr/java/jdk1.7.0_67
export HADOOP_PREFIX=/opt/chy/hadoop
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_PREFIX/bin:$HADOOP_PREFIX/sbin
-----------------------------------------------------
source /etc/profile
jps


# 3.在/root下生成 .ssh文件(所有节点,可通过Xshell全部会话功能)
ssh localhost
exit
cd /root/.ssh


# 4.将node1中的公钥发送到其他节点
scp id_dsa.pub node2:`pwd`/node1.pub
scp id_dsa.pub node3:`pwd`/node1.pub
scp id_dsa.pub node4:`pwd`/node1.pub
# 将每个节点 中收到的公钥追加到 authorized_keys文件中(没有自动创建)
cat node1.pub >> authorized_keys
# 将node1.pub的内容发送到其他节点或者将node1的authorized_keys发送到其他节点
scp authorized_keys 其他节点:`pwd`


# 5.在 node1 配置HADOOP中的HDFS,配置完成后分发给其他节点
 cd /opt/chy/hadoop/etc/hadoop/
 
 # 修改NN启动配置
 vim core-site.xml
 -----------------------
 <configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://node1:9000</value>
    </property>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>/var/chy/hadoop/full</value>
    </property>
-----------------------

# 修改DN启动配置
vi slaves
-----------------------
node2
node3
node4
-----------------------

# 修改SNN启动配置(副本数,启动ip)
-----------------------
<configuration>
    <property>
       <name>dfs.replication</name>
       <value>2</value>
    </property>
    <property>
        <name>dfs.namenode.secondary.http-address</name>
        <value>node2:50090</value>
    </property>
</configuration>
-----------------------

注意 : 该步骤是在伪集群配置的基础上运行的 ,如果一开始没有搭建伪集群需要对JAVA_HOME进行二次配置 ,放开下面文件中关于java环境变量的配置, 指定成自己的环境变量配置
vi /opt/chy/hadoop/etc/hadoop/hadoop-env.sh
vi /opt/chy/hadoop/etc/hadoop/mapred-env.sh
vi /opt/chy/hadoop/etc/hadoop/yarn-env.sh


# 6. 将node1配置好的hdfs配置分发到其他节点
scp -r hadoop/ node2:`pwd`
scp -r hadoop/ node3:`pwd`
scp -r hadoop/ node4:`pwd`


# 7.启动hdfs
# 格式化(第一次启动必做, 图1)
hdfs dfs -format #之前没有配置伪集群
hdfs namenode -format #之前配置伪集群(dfs不会格式化namenode的信息)

# 查看namenode中是否创建了VERSION文件
cd /var/chy/hadoop/full/dfs/name/current/

# 启动hdfs服务(图2)
# 根据图示可知,node1启动了name,node2启动了SNN,node2,3,4启动了datanode,可以通过 jps 查看是否启动
start-dfs.sh

# 访问可视化界面(图3)
http://node1:50070

# 1.在node2中上传ZK , 配置完毕后 , 分发给加他节点(jdk支持)
mv zookeeper-3.4.6 /opt/zookeeper -f


# 2.为了能让ZK配置成在任意位置使用 ,需要配置ZK环境变量
# 配置环境变量
vi /etc/profile
-----------------------------------zk环境变量配置,注意ZK地址指定--------------------------
export ZOOKEEPER_HOME=/opt/zookeeper
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_PREFIX/bin:$HADOOP_PREFIX/sbin:$ZOOKEEPER_HOME/bin
-----------------------------------zk环境变量配置,注意ZK地址指定--------------------------

# 重新读取环境变量配置
. /etc/profile

# 测试配置是否成功 ,输入zk + tab+tab  查看是否会有下面的文件
[root@node2 bin]# zk
zkCleanup.sh  zkCli.cmd     zkCli.sh      zkEnv.cmd     zkEnv.sh      zkServer.cmd  zkServer.sh


# 3,修改ZK的配置文件
-------------------------------------------------------------------------
# 修改数据存放目录
dataDir=/var/chy/hadoop/zk
# 集群配置
server.1=192.168.179.141:2888:3888
server.2=192.168.179.142:2888:3888
server.3=192.168.179.143:2888:3888
-------------------------------------------------------------------------
# 修改配置文件名称(ZK运行时默认读取zoo.cfg)
mv zoo_sample.cfg zoo.cfg

# 数据存放新建目录 ,创建ZK集群的唯一标识文件
mkdir -p /var/chy/hadoop/zk
echo 1 > /var/chy/hadoop/zk/myid


# 4.分发文件到其他节点 ,并修改ZK集群的唯一标识文件
scp -r zookeeper/ node3:`pwd`
scp -r zookeeper/ node4:`pwd`
# node3
mkdir -p /var/chy/hadoop/zk
echo 2 > /var/chy/hadoop/zk/myid
# node4
mkdir -p /var/chy/hadoop/zk
echo 3 > /var/chy/hadoop/zk/myid


# 5.修改node3,node4的ZK环境变量配置 ,
export ZOOKEEPER_HOME=/opt/zookeeper
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_PREFIX/bin:$HADOOP_PREFIX/sbin:$ZOOKEEPER_HOME/bin

# 重载, 并测试
[root@node4 chy]# . /etc/profile
[root@node4 chy]# zk
zkCleanup.sh  zkCli.cmd     zkCli.sh      zkEnv.cmd     zkEnv.sh      zkServer.cmd  zkServer.sh


#  9.启动node2,3,4的zookeeper ,查看哪个是领导者哪个是追随者
 zkServer.sh start
 zkServer.sh status

# 1. 基于伪集群 / 集群 安装后
cd /opt/chy/hadoop/etc/

# 备份hadoop到hadoop-full, hadoop-full是我们之前配置的hdfs集群
cp -rf hadoop/ hadoop-full

# 在hadoop目录下而不是hadoop-full
# 2.修改hdfs-site.xml ,修改集群逻辑名称到物理的映射 ,代码块在下方
# 3.修改core-site.xml,指定集群名称, 存放数据的地址, 集群ip, 代码块在下方
# 4. 修改slave配置
[root@node2 hadoop-full]# vim slaves 
--------------------------------------------------
node2
node3
node4


# 5.分发这些文件到其他节点
scp core-site.xml hdfs-site.xml slaves node2:`pwd`
scp core-site.xml hdfs-site.xml slaves node3:`pwd`
scp core-site.xml hdfs-site.xml slaves node4:`pwd`
 

# 6.要进行初始化, 首先应该启动JN(node1,node2,node3)
hadoop-daemon.sh start journalnode
# node1
hdfs namenode -format   #图1

# node1启动namenode(主)
hadoop-daemon.sh start namenode
# node2启动备用namenode
hdfs namenode -bootstrapStandby

# node1格式化ZK(图2 ,可以看到初始化ZK后会在集群中创建/hadoop-ha/mycluster节点,且目录下为空)
hdfs zkfc -formatZK


# 7.启动HDFS服务(图3, 图4)
start-dfs.sh 

# 在node2,3,4中使用jps可以看到QuorumPeerMain进程,这个就是代表ZK启动进程

# 访问主NN和备NN的图形化界面 
# 前提: 配置本地hosts地址映射,将备NN所在IP映射成node2
http://node1:50070/
http://node2:50070/

# 第二次启动只需要
1. 启动zk
2. start-dfs.sh

    <property>
       <name>dfs.replication</name>
       <value>2</value>
    </property>
<property>
    <name>dfs.nameservices</name>
    <value>mycluster</value>
</property>
<property>
    <name>dfs.ha.namenodes.mycluster</name>
    <value>nn1,nn2</value>
</property>
<property>
    <name>dfs.namenode.rpc-address.mycluster.nn1</name>
    <value>node1:8020</value>
</property>
<property>
    <name>dfs.namenode.rpc-address.mycluster.nn2</name>
    <value>node2:8020</value>
</property>
<property>
    <name>dfs.namenode.http-address.mycluster.nn1</name>
    <value>node1:50070</value>
</property>
<property>
    <name>dfs.namenode.http-address.mycluster.nn2</name>
    <value>node2:50070</value>
</property>
<property>
  <name>dfs.namenode.shared.edits.dir</name>
  <value>qjournal://node1:8485;node2:8485;node3:8485/mycluster</value>
</property>

<property>
  <name>dfs.journalnode.edits.dir</name>
  <value>/var/chy/hadoop/ha/jn</value>
</property>

<property>
  <name>dfs.client.failover.proxy.provider.mycluster</name>
  <value>
      org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
    </value>
</property>
<property>
  <name>dfs.ha.fencing.methods</name>
  <value>sshfence</value>
</property>
<property>
  <name>dfs.ha.fencing.ssh.private-key-files</name>
  <value>/root/.ssh/id_dsa</value>
</property>

<property>
   <name>dfs.ha.automatic-failover.enabled</name>
   <value>true</value>
 </property>

<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://mycluster</value>
    </property>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>/var/chy/hadoop/ha</value>
    </property>
    <property>
       <name>ha.zookeeper.quorum</name>
       <value>node2:2181,node3:2181,node4:2181</value>
   </property>
</configuration>

# 1.在Xshell右下方选择全部会话功能, 安装ntp服务
yum -y install ntpdate

# 2.启动ntp服务(-u指定的是网络上的ntp服务器,可以自行搜索)
ntpdate -u ntp.api.bz

# 3.查看当前日期
date

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class TestHDFS {
	// 为配置文件创建对象
	Configuration conf = null;
	// 客户端对象
	FileSystem fs = null;
	
	@Before
	public void before() throws Exception{
		// true 代表读配置文件
		conf = new Configuration(true);
		// 通过配置文件获取对应的对象HDFS,FTP...
		fs = FileSystem.get(conf);
	}
	
	@After
	public void after() throws Exception{
		fs.close(); //关闭资源
	}
	
	// 创建目录
	@Test
	public void mkdir() throws Exception{
		// 给定一个目录，若该目录存在，则直接删除再创建
		// 若该目录不存在，直接创建
		Path f = new Path("/test1");
		if(fs.exists(f)){
			fs.delete(f, true);
		}
		fs.mkdirs(f);	
	}
	
	// 上传文件
	@Test
	public void upload() throws Exception{
		
		InputStream is = 
				new BufferedInputStream(
            new FileInputStream(new File("E:\testhadoop.txt")));//上传文件原地址

		Path p = new Path("/user/root/test-h.txt");//上传文件目的地址
		FSDataOutputStream out = fs.create(p);
		
		IOUtils.copyBytes(is, out, conf, true);
		
	}
	
	// 下载文件
	@Test 
	public void download() throws Exception{
		Path path = new Path("/user/root/test-h.txt");//下载文件源地址
		InputStream in = fs.open(path);
		// 输入流
		FSDataInputStream fsin = new FSDataInputStream(in);
		// 输出流
		OutputStream ops = 
			new BufferedOutputStream(
            	new FileOutputStream(new File("E:\download-test-h.txt")));//下载文件目的地址
		// 流对接
		IOUtils.copyBytes(fsin, ops, conf);
		
	}
	
	
	// 块
	@Test 
	public void testBlockLocation() throws Exception{
		Path path = new Path("/user/root/testhadoop.txt");
		FileStatus fStatus = fs.getFileStatus(path);
		//位置新 ,包括坐在块起始字节,结束字节,所在块地址
		BlockLocation[] bkls = 
            fs.getFileBlockLocations(fStatus , 0, fStatus.getLen());
		
		for (BlockLocation bkl : bkls) {
			System.out.println(bkl);
		}
		
		FSDataInputStream in = fs.open(path);
		System.out.println("----------下面是第一个Block--------------");
		//输出文件中的字节
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());

		System.out.println((char)in.readByte());
		System.out.println("----------下面是第二个Block--------------");
		//跳到第1048576个字节后,这里指的也就是第二个块中的文件
		in.seek(1048576);
		
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
		System.out.println((char)in.readByte());
	}

}