大数据入门（Hadoop简介、Hadoop伪分布式安装、HDFS相关概念及shell操作和Java API操作）

tar -zxvf hadoop*tar.gz
为了解决警告,可以直接下载源码自己编译.但是这个过程比较复杂.
配置：
   		1、hadoop-env.sh
			export JAVA_HOME=/opt/modules/jdk1.6.0_45
		2、yarn-env.sh
			export JAVA_HOME=/opt/modules/jdk1.6.0_45
		3、mapred-env.sh
			export JAVA_HOME=/opt/modules/jdk1.6.0_45
		4、core-site.xml
			<property>		 
				<name>fs.default.name</name>
				<value>hdfs://hadoop-yarn.dragon.org:8020</value>
			</property>
			
			<property>
				<name>hadoop.tmp.dir</name>
				<value>/opt/modules/hadoop-2.2.0/data/tmp</value>
			</property>
		5、hdfs-site.xml
			<property>		
				<name>dfs.replication</name>
				<value>1</value>
			</property>
		6、yarn-site.xml
			<property>
				<name>yarn.nodemanager.aux-services</name>
				<value>mapreduce_shuffle</value>
			</property>
		7、mapred-site.xml
			<property>	 	        		
				<name>mapreduce.framework.name</name>
				<value>yarn</value>
			</property>

修改hostname 
core-site.xml里面保持一致 hostname  ...
     	DataNode NameNode SecendardNameNode
	 	    * NameNode 格式化
				bin/hdfs namenode -format 产生一个集群的ID
			* 启动NameNode
				sbin/hadoop-daemon.sh start namenode
			* 启动DataNode
				sbin/hadoop-daemon.sh start datanode
			* 启动SecondaryNameNode
				sbin/hadoop-daemon.sh start secondarynamenode

jps
查看java进程 namenode secondarynamenode datanode

测试：
http://hadoop-yarn.zhengqing.com:50070
http://hadoop-yarn.zhengqing.com:50090

 [-appendToFile <localsrc> ... <dst>]
        [-cat [-ignoreCrc] <src> ...]
        [-checksum <src> ...]
        [-chgrp [-R] GROUP PATH...]
        [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
        [-chown [-R] [OWNER][:[GROUP]] PATH...]
        [-copyFromLocal [-f] [-p] <localsrc> ... <dst>]
        [-copyToLocal [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-count [-q] <path> ...]
        [-cp [-f] [-p] <src> ... <dst>]
        [-createSnapshot <snapshotDir> [<snapshotName>]]
        [-deleteSnapshot <snapshotDir> <snapshotName>]
        [-df [-h] [<path> ...]]
        [-du [-s] [-h] <path> ...]
        [-expunge]
        [-get [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-getfacl [-R] <path>]
        [-getmerge [-nl] <src> <localdst>]
        [-help [cmd ...]]
        [-ls [-d] [-h] [-R] [<path> ...]]
        [-mkdir [-p] <path> ...]
        [-moveFromLocal <localsrc> ... <dst>]
        [-moveToLocal <src> <localdst>]
        [-mv <src> ... <dst>]
        [-put [-f] [-p] <localsrc> ... <dst>]
        [-renameSnapshot <snapshotDir> <oldName> <newName>]
        [-rm [-f] [-r|-R] [-skipTrash] <src> ...]
        [-rmdir [--ignore-fail-on-non-empty] <dir> ...]
        [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
        [-setrep [-R] [-w] <rep> <path> ...]
        [-stat [format] <path> ...]
        [-tail [-f] <file>]
        [-test -[defsz] <path>]
        [-text [-ignoreCrc] <src> ...]
        [-touchz <path> ...]
        [-usage [cmd ...]]

-help             
功能：输出这个命令参数手册

-ls                  
功能：显示目录信息
示例： hadoop fs -ls hdfs://hadoop-server01:9000/
备注：这些参数中，所有的hdfs路径都可以简写
-->hadoop fs -ls /   等同于上一条命令的效果

-mkdir              
功能：在hdfs上创建目录
示例：hadoop fs  -mkdir  -p  /aaa/bbb/cc/dd

-moveFromLocal            
功能：从本地剪切粘贴到hdfs
示例：hadoop  fs  - moveFromLocal  /home/hadoop/a.txt  /aaa/bbb/cc/dd
-moveToLocal              
功能：从hdfs剪切粘贴到本地
示例：hadoop  fs  - moveToLocal   /aaa/bbb/cc/dd  /home/hadoop/a.txt 

--appendToFile  
功能：追加一个文件到已经存在的文件末尾
示例：hadoop  fs  -appendToFile  ./hello.txt  hdfs://hadoop-server01:9000/hello.txt
可以简写为：
Hadoop  fs  -appendToFile  ./hello.txt  /hello.txt

-cat  
功能：显示文件内容  
示例：hadoop fs -cat  /hello.txt

-tail                 
功能：显示一个文件的末尾
示例：hadoop  fs  -tail  /weblog/access_log.1
-text                  
功能：以字符形式打印一个文件的内容
示例：hadoop  fs  -text  /weblog/access_log.1

-chgrp 
-chmod
-chown
功能：linux文件系统中的用法一样，对文件所属权限
示例：
hadoop  fs  -chmod  666  /hello.txt
hadoop  fs  -chown  someuser:somegrp   /hello.txt

-copyFromLocal    
功能：从本地文件系统中拷贝文件到hdfs路径去
示例：hadoop  fs  -copyFromLocal  ./jdk.tar.gz  /aaa/
-copyToLocal      
功能：从hdfs拷贝到本地
示例：hadoop fs -copyToLocal /aaa/jdk.tar.gz

-cp              
功能：从hdfs的一个路径拷贝hdfs的另一个路径
示例： hadoop  fs  -cp  /aaa/jdk.tar.gz  /bbb/jdk.tar.gz.2

-mv                     
功能：在hdfs目录中移动文件
示例： hadoop  fs  -mv  /aaa/jdk.tar.gz  /

-get              
功能：等同于copyToLocal，就是从hdfs下载文件到本地
示例：hadoop fs -get  /aaa/jdk.tar.gz
-         
功能：合并下载多个文件
示例：比getmerge    如hdfs的目录 /aaa/下有多个文件:log.1, log.2,log.3,...
hadoop fs -getmerge /aaa/log.* ./log.sum

-put                
功能：等同于copyFromLocal
示例：hadoop  fs  -put  /aaa/jdk.tar.gz  /bbb/jdk.tar.gz.2

-rm                
功能：删除文件或文件夹
示例：hadoop fs -rm -r /aaa/bbb/

-rmdir                 
功能：删除空目录
示例：hadoop  fs  -rmdir   /aaa/bbb/ccc

-df               
功能：统计文件系统的可用空间信息
示例：hadoop  fs  -df  -h  /

-du 
功能：统计文件夹的大小信息
示例：
hadoop  fs  -du  -s  -h /aaa/*

-count         
功能：统计一个指定目录下的文件节点数量
示例：hadoop fs -count /aaa/

-setrep                
功能：设置hdfs中文件的副本数量
示例：hadoop fs -setrep 3 /aaa/jdk.tar.gz

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

public static void main(String[] args) throws IOException {
        //创建配置对象并设置hdfs访问路径
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://hadoop-yarn.zhengqing.com:8020");

        //伪造身份,必须放到获取文件系统前面
        System.setProperty("HADOOP_USER_NAME","root");
        //通过问卷系统进行访问
        FileSystem fs = FileSystem.get(conf);
        //本地拷贝到hdfs  Permission denied: user=Administrator, access=WRITE, inode="/data01":root:supergroup:drwxr-xr-x
        //fs.copyFromLocalFile(new Path( "G://dblog.sql"),new Path("/data01"));

        //拷贝到本地
        fs.copyToLocalFile(new Path("/data01/dblog.sql"),new Path("c://"));
        //递归列表根路径下面所有的文件
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/"), true);
        while (iterator.hasNext()) {
            LocatedFileStatus next = iterator.next();
            System.out.println(next.getPath());
        }
        fs.close();
    }

Configuration conf = new Configuration()
FileSystem fs = FileSystem.get(conf)

public class HdfsClient {
	FileSystem fs = null;
	@Before
	public void init() throws Exception {
		// 构造一个配置参数对象，设置一个参数：我们要访问的hdfs的URI
		// 从而FileSystem.get()方法就知道应该是去构造一个访问hdfs文件系统的客户端，以及hdfs的访问地址
		// new Configuration();的时候，它就会去加载jar包中的hdfs-default.xml
		// 然后再加载classpath下的hdfs-site.xml
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", "hdfs://hdp-node01:9000");
		/**
		 * 参数优先级： 1、客户端代码中设置的值 2、classpath下的用户自定义配置文件 3、然后是服务器的默认配置
		 */
		conf.set("dfs.replication", "3");

		// 获取一个hdfs的访问客户端，根据参数，这个实例应该是DistributedFileSystem的实例
		// fs = FileSystem.get(conf);

		// 如果这样去获取，那conf里面就可以不要配"fs.defaultFS"参数，而且，这个客户端的身份标识已经是hadoop用户
		fs = FileSystem.get(new URI("hdfs://hdp-node01:9000"), conf, "hadoop");
	}

	/**
	 * 往hdfs上传文件
	 * @throws Exception
	 */
	@Test
	public void testAddFileToHdfs() throws Exception {
		// 要上传的文件所在的本地路径
		Path src = new Path("g:/redis-recommend.zip");
		// 要上传到hdfs的目标路径
		Path dst = new Path("/aaa");
		fs.copyFromLocalFile(src, dst);
		fs.close();
	}

	/**
	 * 从hdfs中复制文件到本地文件系统
	 * @throws IOException
	 * @throws IllegalArgumentException
	 */
	@Test
	public void testDownloadFileToLocal() throws IllegalArgumentException, IOException {
		fs.copyToLocalFile(new Path("/jdk-7u65-linux-i586.tar.gz"), new Path("d:/"));
		fs.close();
	}

	@Test
	public void testMkdirAndDeleteAndRename() throws IllegalArgumentException, IOException {
		// 创建目录
		fs.mkdirs(new Path("/a1/b1/c1"));
		// 删除文件夹 ，如果是非空文件夹，参数2必须给值true
		fs.delete(new Path("/aaa"), true);
		// 重命名文件或文件夹
		fs.rename(new Path("/a1"), new Path("/a2"));
	}

	/**
	 * 查看目录信息，只显示文件
	 * @throws IOException
	 * @throws IllegalArgumentException
	 * @throws FileNotFoundException
	 */
	@Test
	public void testListFiles() throws FileNotFoundException, IllegalArgumentException, IOException {
		// 思考：为什么返回迭代器，而不是List之类的容器
		RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
		while (listFiles.hasNext()) {
			LocatedFileStatus fileStatus = listFiles.next();
			System.out.println(fileStatus.getPath().getName());
			System.out.println(fileStatus.getBlockSize());
			System.out.println(fileStatus.getPermission());
			System.out.println(fileStatus.getLen());
			BlockLocation[] blockLocations = fileStatus.getBlockLocations();
			for (BlockLocation bl : blockLocations) {
				System.out.println("block-length:" + bl.getLength() + "--" + "block-offset:" + bl.getOffset());
				String[] hosts = bl.getHosts();
				for (String host : hosts) {
					System.out.println(host);
				}
			}
			System.out.println("--------------为angelababy打印的分割线--------------");
		}
	}

	/**
	 * 查看文件及文件夹信息
	 * @throws IOException
	 * @throws IllegalArgumentException
	 * @throws FileNotFoundException
	 */
	@Test
	public void testListAll() throws FileNotFoundException, IllegalArgumentException, IOException {
		FileStatus[] listStatus = fs.listStatus(new Path("/"));
		String flag = "d--             ";
		for (FileStatus fstatus : listStatus) {
			if (fstatus.isFile())  flag = "f--         ";
			System.out.println(flag + fstatus.getPath().getName());
		}
	}
}

/**
 * 相对那些封装好的方法而言的更底层一些的操作方式
 * 上层那些mapreduce   spark等运算框架，去hdfs中获取数据的时候，就是调的这种底层的api
 * @author 郑清
 */
public class StreamAccess {
	FileSystem fs = null;
	@Before
	public void init() throws Exception {
		Configuration conf = new Configuration();
		fs = FileSystem.get(new URI("hdfs://hdp-node01:9000"), conf, "hadoop");
	}
	
    /**
	 * 通过流的方式上传文件到hdfs
	 * @throws Exception
	 */
	@Test
	public void testUpload() throws Exception {
		FSDataOutputStream outputStream = fs.create(new Path("/angelababy.love"), true);
		FileInputStream inputStream = new FileInputStream("c:/angelababy.love");
		IOUtils.copy(inputStream, outputStream);
	}
	
	@Test
	public void testDownLoadFileToLocal() throws IllegalArgumentException, IOException{
		//先获取一个文件的输入流----针对hdfs上的
		FSDataInputStream in = fs.open(new Path("/jdk-7u65-linux-i586.tar.gz"));
		//再构造一个文件的输出流----针对本地的
		FileOutputStream out = new FileOutputStream(new File("c:/jdk.tar.gz"));
		//再将输入流中数据传输到输出流
		IOUtils.copyBytes(in, out, 4096);
	}
	
	/**
	 * hdfs支持随机定位进行文件读取，而且可以方便地读取指定长度
	 * 用于上层分布式运算框架并发处理数据
	 * @throws IllegalArgumentException
	 * @throws IOException
	 */
	@Test
	public void testRandomAccess() throws IllegalArgumentException, IOException{
		//先获取一个文件的输入流----针对hdfs上的
		FSDataInputStream in = fs.open(new Path("/iloveyou.txt"));
		//可以将流的起始偏移量进行自定义
		in.seek(22);
		
		//再构造一个文件的输出流----针对本地的
		FileOutputStream out = new FileOutputStream(new File("c:/iloveyou.line.2.txt"));
		IOUtils.copyBytes(in,out,19L,true);
	}
	
	/**
	 * 显示hdfs上文件的内容
	 * @throws IOException 
	 * @throws IllegalArgumentException 
	 */
	@Test
	public void testCat() throws IllegalArgumentException, IOException{
		FSDataInputStream in = fs.open(new Path("/iloveyou.txt"));
		IOUtils.copyBytes(in, System.out, 1024);
	}
}

	@Test
	public void testCat() throws IllegalArgumentException, IOException{
		FSDataInputStream in = fs.open(new Path("/weblog/input/access.log.10"));
		//拿到文件信息
		FileStatus[] listStatus = fs.listStatus(new Path("/weblog/input/access.log.10"));
		//获取这个文件的所有block的信息
		BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0L, listStatus[0].getLen());
		//第一个block的长度
		long length = fileBlockLocations[0].getLength();
		//第一个block的起始偏移量
		long offset = fileBlockLocations[0].getOffset();
		
		System.out.println(length);
		System.out.println(offset);
		
		//获取第一个block写入输出流
//		IOUtils.copyBytes(in, System.out, (int)length);
		byte[] b = new byte[4096];
		
		FileOutputStream os = new FileOutputStream(new File("d:/block0"));
		while(in.read(offset, b, 0, 4096)!=-1){
			os.write(b);
			offset += 4096;
			if(offset>=length) return;
		};
		os.flush();
		os.close();
		in.close();
	}