大数据Hadoop之MapReduce
Posted Frankdeng
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了大数据Hadoop之MapReduce相关的知识，希望对你有一定的参考价值。
一 HDFS客户端环境准备
1.1 jar包准备
1）解压hadoop-2.7.6.tar.gz到非中文目录
2）进入share文件夹，查找所有jar包，并把jar包拷贝到_lib文件夹下
3）在全部jar包中查找sources.jar，并剪切到_source文件夹。
4）在全部jar包中查找tests.jar，并剪切到_test文件夹
1.2 Eclipse准备
1）根据自己电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径（例如：E:\\02_software\\hadoop-2.7.6）。（如果不生效，重新启动eclipse）
2）配置HADOOP_HOME环境变量
3）创建第一个java工程HdfsClientDemo1
4）创建lib文件夹，然后添加jar包
5）创建包，HdfsClient测试类
public class HdfsClient {
// 上传文件
public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件系统
Configuration configuration = new Configuration();
// 配置在集群上运行
// configuration.set("fs.defaultFS", "hdfs://node21:9000");
// FileSystem fs = FileSystem.get(configuration);
FileSystem fs = FileSystem.get(new URI("hdfs://node21:9000"), configuration, "admin");
// 2 上传文件
fs.copyFromLocalFile(new Path("e:/hello.txt"), new Path("/hello2.txt"));
// 3 关闭资源
fs.close();
System.out.println("over");
}
}
6）执行程序
运行时需要配置用户名称，客户端去操作hdfs时，是有一个用户身份的。默认情况下，hdfs客户端api会从jvm中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=admin，admin为用户名称。
7）注意：如果eclipse打印不出日志，在控制台上只显示
1.log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell). 2.log4j:WARN Please initialize the log4j system properly. 3.log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
需要在项目的src目录下，新建一个文件，命名为“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
二 HDFS的高级API编程
2.1 HDFS文件上传（测试参数优先级）
1.代码
@Test
public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件系统
Configuration configuration = new Configuration();
configuration.set("dfs.replication", "2");
FileSystem fs = FileSystem.get(new URI("hdfs://node21:9000"), configuration, "admin");
// 2 上传文件
fs.copyFromLocalFile(new Path("e:/hello.txt"), new Path("/hello5.txt"));
// 3 关闭资源
fs.close();
System.out.println("over");
}
2.将hdfs-site.xml拷贝到项目的根目录下
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<property>
<name>dfs.replication</name>
<value>1</value>
</property>
</configuration>
3.测试参数优先级
参数优先级： （1）客户端代码中设置的值 >（2）classpath下的用户自定义配置文件 >（3）然后是服务器的默认配置
2.2 文件的上传和下载
package com.xyg.hdfs.api;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFS_GET_AND_PUT {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://node21:9000");
conf.set("dfs.replication", "2");
FileSystem fs = FileSystem.get(conf);
/**
* 更改操作用户有两种方式:
* 1、直接设置运行换种的用户名为hadoop
* VM arguments ; -DHADOOP_USER_NAME=admin
* 2、在代码中进行声明
* System.setProperty("HADOOP_USER_NAME", "admin");
*/
System.setProperty("HADOOP_USER_NAME", "admin");
// 上传
fs.copyFromLocalFile(new Path("c:/sss.txt"), new Path("/a/ggg.txt"));
/**
* .crc ： 校验文件
* 每个块的元数据信息都只会记录合法数据的起始偏移量： qqq.txt blk_41838 : 0 - 1100byte
* 如果进行非法的数据追加。最终是能够下载合法数据。
* 由于你在数据的中间， 也就是说在 0 -1100 之间的范围进行了数据信息的更改。 造成了采用CRC算法计算出来校验值，和最初存入进HDFS的校验值
* 不一致。HDFS就认为当前这个文件被损坏了。
*/
// 下载
fs.copyToLocalFile(new Path("/a/qqq.txt"), new Path("c:/qqq3.txt"));
/**
* 上传和下载的API的底层封装其实就是 ： FileUtil.copy(....)
*/
fs.close();
}
}
2、配置文件conf
package com.xyg.hdfs;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map.Entry;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
public class TestConf1 {
public static void main(String[] args) throws Exception {
/**
* 底层会加载一堆的配置文件：
*
* core-default.xml
* hdfs-default.xml
* mapred-default.xml
* yarn-default.xml
*/
Configuration conf = new Configuration();
// conf.addResource("hdfs-default.xml");
/**
* 当前这个hdfs-site.xml文件就放置在这个项目中的src下。也就是classpath路径下。
* 所以 FS在初始化的时候，会把hdfs-site.xml这个文件中的name-value对解析到conf中
* 但是：
* 1、如果hdfs-site.xml 不在src下， 看是否能加载？？？ 不能
* 2、如果文件名不叫做 hdfs-default.xml 或者 hdsf-site.xml 看是否能自动加载？？？ 不能
* 得出的结论：
* 如果需要项目代码自动加载配置文件中的信息，那么就必须把配置文件改成-default.xml或者-site.xml的名称
* 而且必须放置在src下
*
* 那如果不叫这个名，或者不在src下，也需要加载这些配置文件中的参数：
* 必须使用conf对象提供的一些方法去手动加载
*/
// conf.addResource("hdfs-site.xml");
conf.set("dfs.replication", "1");
conf.addResource("myconfig/hdfs-site.xml");
/**
* 依次加载的参数信息的顺序是：
* 1、加载 core/hdfs/mapred/yarn-default.xml
* 2、加载通过conf.addResources()加载的配置文件
* 3、加载conf.set(name, value)
*/
FileSystem fs = FileSystem.get(conf);
System.out.println(conf.get("dfs.replication"));
Iterator<Entry<String, String>> iterator = conf.iterator();
while(iterator.hasNext()){
Entry<String, String> e = iterator.next();
System.out.println(e.getKey() + "\\t" + e.getValue());
}
}
}
输出结果
View Code3、列出指定目录下的文件以及块的信息
package com.xyg.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
public class TestHDFS1 {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
System.setProperty("HADOOP_USER_NAME", "admin");
conf.set("fs.defaultFS", "hdfs://node21:9000");
FileSystem fs = FileSystem.get(conf);
/**
* 列出指定的目录下的所有文件
*/
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
while(listFiles.hasNext()){
LocatedFileStatus file = listFiles.next();
System.out.println(file.getPath()+"\\t");
System.out.println(file.getPath().getName()+"\\t");
System.out.println(file.getLen()+"\\t");
System.out.println(file.getReplication()+"\\t");
/**
* blockLocations的长度是几？ 是什么意义？
*
* 块的数量
*/
BlockLocation[] blockLocations = file.getBlockLocations();
System.out.println(blockLocations.length+"\\t");
for(BlockLocation bl : blockLocations){
String[] hosts = bl.getHosts();
System.out.print(hosts[0] + "-" + hosts[1]+"\\t");
}
System.out.println();
}
}
}
输出结果
hdfs://hadoop1:9000/aa/bb/cc/hadoop.tar.gz hadoop.tar.gz 199007110 2 3 hadoop3-hadoop1 hadoop1-hadoop2 hadoop1-hadoop4
4、上传文件
package com.xyg.hdfs;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import org.apache.hadoop.conf.Configuration;
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 UploadDataByStream {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
System.setProperty("HADOOP_USER_NAME", "admin");
conf.set("fs.defaultFS", "hdfs://node21:9000");
FileSystem fs = FileSystem.get(conf);
InputStream in = new FileInputStream(new File("d:/abc.tar.gz"));
FSDataOutputStream out = fs.create(new Path("/aa/abc.tar.gz"));
IOUtils.copyBytes(in, out, 4096, true);
fs.close();
}
}
5、下载文件
package com.xyg.hdfs;
import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
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;
public class DownloadDataByStream {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
System.setProperty("HADOOP_USER_NAME", "admin");
conf.set("fs.defaultFS", "hdfs://node21:9000");
FileSystem fs = FileSystem.get(conf);
FSDataInputStream in = fs.open(new Path("/aa/abc.tar.gz"));
OutputStream out = new FileOutputStream(new File("D:/abc.sh"));
IOUtils.copyBytes(in, out, 4096, true);
fs.close();
}
}
6、删除某个路径下特定类型的文件，比如class类型文件，比如txt类型文件
package com.xyg.hdfs;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFS_DELETE_CLASS {
public static final String FILETYPE = "tar.gz";
public static final String DELETE_PATH = "/aa";
public static void main(String[] args) throws Exception {
new HDFS_DELETE_CLASS().rmrClassFile(new Path(DELETE_PATH));
}
public void rmrClassFile(Path path) throws Exception{
// 首先获取集群必要的信息，以得到FileSystem的示例对象fs
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://node2:9000"), conf, "admin");
// 首先检查path本身是文件夹还是目录
FileStatus fileStatus = fs.getFileStatus(path);
boolean directory = fileStatus.isDirectory();
// 根据该目录是否是文件或者文件夹进行相应的操作
if(directory){
// 如果是目录
checkAndDeleteDirectory(path, fs);
}else{
// 如果是文件，检查该文件名是不是FILETYPE类型的文件
checkAndDeleteFile(path, fs);
}
}
// 处理目录
public static void checkAndDeleteDirectory(Path path, FileSystem fs) throws Exception{
// 查看该path目录下一级子目录和子文件的状态
FileStatus[] listStatus = fs.listStatus(path);
for(FileStatus fStatus: listStatus){
Path p = fStatus.getPath();
// 如果是文件，并且是以FILETYPE结尾，则删掉，否则继续遍历下一级目录
if(fStatus.isFile()){
checkAndDeleteFile(p, fs);
}else{
checkAndDeleteDirectory(p, fs);
}
}
}
// 檢查文件是否符合刪除要求，如果符合要求則刪除，不符合要求则不做处理
public static void checkAndDeleteFile(Path path, FileSystem fs) throws Exception{
String name = path.getName();
System.out.println(name);
/*// 直接判断有没有FILETYPE这个字符串,不是特别稳妥，并且会有误操作，所以得判断是不是以FILETYPE结尾
if(name.indexOf(FILETYPE) != -1){
fs.delete(path, true);
}*/
// 判断是不是以FILETYPE结尾
int startIndex = name.length() - FILETYPE.length();
int endIndex = name.length();
// 求得文件后缀名
String fileSuffix = name.substring(startIndex, endIndex);
if(fileSuffix.equals(FILETYPE)){
fs.delete(path, true);
}
}
}
7、删除HDFS集群中的所有空文件和空目录
public class DeleteEmptyDirAndFile {
static FileSystem fs = null;
public static void main(String[] args) throws Exception {
initFileSystem();
// 创建测试数据
// makeTestData();
// 删除测试数据
// deleteTestData();
// 删除指定文件夹下的空文件和空文件夹
deleteEmptyDirAndFile(new Path("/aa"));
}
/**
* 删除指定文件夹下的 空文件 和 空文件夹
* @throws Exception
*/
public static void deleteEmptyDirAndFile(Path path) throws Exception {
//当是空文件夹时
FileStatus[] listStatus = fs.listStatus(path);
if(listStatus.length == 0){
fs.delete(path, true);
return;
}
// 该方法的结果：包括指定目录的 文件 和 文件夹
RemoteIterator<LocatedFileStatus> listLocatedStatus = fs.listLocatedStatus(path);
while (listLocatedStatus.hasNext()) {
LocatedFileStatus next = listLocatedStatus.next();
Path currentPath = next.getPath();
// 获取父目录
Path parent = next.getPath().getParent();
// 如果是文件夹，继续往下遍历，删除符合条件的文件（空文件夹）
if (next.isDirectory()) {
// 如果是空文件夹
if(fs.listStatus(currentPath).length == 0){
// 删除掉
fs.delete(currentPath, true);
}else{
// 不是空文件夹，那么则继续遍历
if(fs.exists(currentPath)){
deleteEmptyDirAndFile(currentPath);
}
}
// 如果是文件
} else {
// 获取文件的长度
long fileLength = next.getLen();
// 当文件是空文件时， 删除
if(fileLength == 0){
fs.delete(currentPath, true);
}
}
// 当空文件夹或者空文件删除时，有可能导致父文件夹为空文件夹，
// 所以每次删除一个空文件或者空文件的时候都需要判断一下，如果真是如此，那么就需要把该文件夹也删除掉
int length = fs.listStatus(parent).length;
if(length == 0){
fs.delete(parent, true);
}
}
}
/**
* 初始化FileSystem对象之用
*/
public static void initFileSystem() throws Exception{
Configuration conf = new Configuration();
System.setProperty("HADOOP_USER_NAME", "admin");
conf.addResource("config/core-site.xml");
conf.addResource("config/hdfs-site.xml");
fs = FileSystem.get(conf);
}
/**
* 创建 测试 数据之用
*/
public static void makeTestData() throws Exception {
String emptyFilePath = "D:\\\\bigdata\\\\1704mr_test\\\\empty.txt";
String notEmptyFilePath = "D:\\\\bigdata\\\\1704mr_test\\\\notEmpty.txt";
// 空文件夹 和 空文件 的目录
String path1 = "/aa/bb1/cc1/dd1/";
fs.mkdirs(new Path(path1));
fs.mkdirs(new Path("/aa/bb1/cc1/dd2/"));
fs.copyFromLocalFile(new Path(emptyFilePath), new Path(path1));
fs.copyFromLocalFile(new Path(notEmptyFilePath), new Path(path1));
// 空文件 的目录
String path2 = "/aa/bb1/cc2/dd2/";
fs.mkdirs(new Path(path2));
fs.copyFromLocalFile(new Path(emptyFilePath), new Path(path2));
// 非空文件 的目录
String path3 = "/aa/bb2/cc3/dd3";
fs.mkdirs(new Path(path3));
fs.copyFromLocalFile(new Path(notEmptyFilePath), new Path(path3));
// 空 文件夹
String path4 = "/aa/bb2/cc4/dd4";
fs.mkdirs(new Path(path4));
System.out.println("测试数据创建成功");
}
/**
* 删除 指定文件夹
* @throws Exception
*/
public static void deleteTestData() throws Exception {
boolean delete = fs.delete(new Path("/aa"), true);
System.out.println(delete ? "删除数据成功" : "删除数据失败");
}
}
8、手动拷贝某个特定的数据块（比如某个文件的第二个数据块）
/**
* 手动拷贝某个特定的数据块（比如某个文件的第二个数据块）
* */
public static void copyBlock(String str,int num) {
Path path = new Path(str);
BlockLocation[] localtions = new BlockLocation[0] ;
try {
FileStatus fileStatus = fs.getFileStatus(path);
localtions = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
/*for(int i=0;i<localtions.length;i++) {
System.out.println(localtions[i]);
}*/
/*System.out.println(localtions[num-1].getOffset());
System.out.println(localtions[num-1].getLength());
String[] hosts = localtions[num-1].getHosts();*/
FSDataInputStream open = fs.open(path);
open.seek(localtions[num-1].getOffset());
OutputStream out = new FileOutputStream(new File("D:/abc.tar.gz"));
IOUtils.copyBytes(open, out,4096,true);
} catch (IOException e) {
e.printStackTrace();
}
}
9、编写程序统计出HDFS文件系统中文件大小小于HDFS集群中的默认块大小的文件占比
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
/**
* 编写程序统计出HDFS文件系统中文件大小小于HDFS集群中的默认块大小的文件占比
* 比如：大于等于128M的文件个数为98，小于128M的文件总数为2，所以答案是2%
*/
public class Exam1_SmallFilePercent {
private static int DEFAULT_BLOCKSIZE = 128 * 1024 * 1024;
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://node21:9000");
System.setProperty("HADOOP_USER_NAME", "admin");
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/");
float smallFilePercent = getSmallFilePercent(fs, path);
System.out.println(smallFilePercent);
fs.close();
}
/**
* 该方法求出指定目录下的小文件和总文件数的对比
* @throws Exception
*/
private static float getSmallFilePercent(FileSystem fs, Path path) throws Exception {
// TODO Auto-generated method stub
int smallFile = 0;
int totalFile = 0;
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);
while(listFiles.hasNext()){
totalFile++;
LocatedFileStatus next = listFiles.next();
long len = next.getLen();
if(len < DEFAULT_BLOCKSIZE){
smallFile++;
}
}
System.out.println(smallFile+" : "+totalFile);
return smallFile * 1f /totalFile;
}
}
10、编写程序统计出HDFS文件系统中的平均数据块数（数据块总数/文件总数）
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
/**
* 编写程序统计出HDFS文件系统中的平均数据块数（数据块总数/文件总数）
* 比如：一个文件有5个块，一个文件有3个块，那么平均数据块数为4
* 如果还有一个文件，并且数据块就1个，那么整个HDFS的平均数据块数就是3
*/
public class Exam2_HDSFAvgBlocks {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://node21:9000");
System.setProperty("HADOOP_USER_NAME", "admin");
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/");
float avgHDFSBlocks = getHDFSAvgBlocks(fs, path);
System.out.println("HDFS的平均数据块个数为：" + avgHDFSBlocks);
fs.close();
}
/**
* 求出指定目录下的所有文件的平均数据块个数
*/
private static float getHDFSAvgBlocks(FileSystem fs, Path path) throws Exception {
// TODO Auto-generated method stub
int totalFiles = 0; // 总文件数
int totalBlocks = 0; // 总数据块数
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);
while(listFiles.hasNext()){
LocatedFileStatus next = listFiles.next();
int length = next.getBlockLocations().length;
totalBlocks += length;
if(next.getLen() != 0){
totalFiles++;
}
}
System.out.println(totalBlocks+" : "+totalFiles);
return totalBlocks * 1f / totalFiles;
}
}
11、编写程序统计出HDFS文件系统中的平均副本数（副本总数/总数据块数）
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
/**
* 编写程序统计出HDFS文件系统中的平均副本数（副本总数/总数据块数）
* 比如：总共两个文件，一个文件5个数据块，每个数据块3个副本，第二个文件2个数据块，每个文件2个副本，最终的平均副本数 = （3*3 + 2*2）/（3+2）= 2.8
*/
public class Exam3_HDSFAvgBlockCopys {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://node21:9000");
System.setProperty("HADOOP_USER_NAME", "admin");
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/");
float avgHDFSBlockCopys = getHDFSAvgBlockCopys(fs, path);
System.out.println("HDFS的平均数据块个数为：" + avgHDFSBlockCopys);
fs.close();
}
/**
* 求出指定目录下的所有文件的平均数据块个数
*/
private static float getHDFSAvgBlockCopys(FileSystem fs, Path path) throws Exception {
// TODO Auto-generated method stub
int totalCopy = 0; // 总副本数
int totalBlocks = 0; // 总数据块数
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);
while(listFiles.hasNext()){
LocatedFileStatus next = listFiles.next();
int length = next.getBlockLocations().length;
short replication = next.getReplication();
totalBlocks += length;
totalCopy += length * replication;
}
System.out.println(totalCopy+" : "+totalBlocks);
return totalCopy * 1f / totalBlocks;
}
}
12、统计HDFS整个文件系统中的不足指定数据块大小的数据块的比例
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
/**
* 统计HDFS整个文件系统中的不足指定数据块大小的数据块的比例
* 比如指定的数据块大小是128M，总数据块有100个，不是大小为完整的128M的数据块有5个，那么不足指定数据块大小的数据块的比例就为5%
* 注意：千万注意考虑不同文件的指定数据块大小可能不一致。所以千万不能用默认的128M一概而论
*/
public class Exam4_LTBlockSize {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://node21:9000");
System.setProperty("HADOOP_USER_NAME", "admin");
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/");
float avgHDFSBlockCopys = getLessThanBlocksizeBlocks(fs, path);
System.out.println("HDFS的不足指定数据块大小的数据块数目为：" + avgHDFSBlockCopys);
fs.close();
}
private static float getLessThanBlocksizeBlocks(FileSystem fs, Path path) throws Exception {
// TODO Auto-generated method stub
int totalBlocks = 0; // 总副本数
int lessThenBlocksizeBlocks = 0; // 总数据块数
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);
while(listFiles.hasNext()){
LocatedFileStatus next = listFiles.next();
BlockLocation[] blockLocations = next.getBlockLocations();
int length = blockLocations.length;
if(length != 0){
totalBlocks += length;
long lastBlockSize = blockLocations[length - 1].getLength();
long blockSize = next.getBlockSize();
if(lastBlockSize < blockSize){
lessThenBlocksizeBlocks++;
}
}
}
System.out.println(lessThenBlocksizeBlocks+" : "+totalBlocks);
return lessThenBlocksizeBlocks * 1f / totalBlocks;
}
}
13、统计出一个给定数组的蓄水总量（把数组的每个位置的数看是做地势高低）
/**
统计出一个给定数组的蓄水总量（把数组的每个位置的数看是做地势高低）
比如：int[] intArray = new int[]{4,3,2,5,6,4,4,7}
能蓄水：[0,1,2,0,0,2,2,0] 所以总量是：7
核心思路：把数组切成很多个 01数组，每一层一个01数组，统计每个01数组中的合法0的总个数（数组的左边第一个1的中间区间中的0的个数）即可
*/
public class Exam5_WaterStoreOfArray {
public static void main(String[] args) {
// int[] intArray = new int[]{4,3,2,5,6,4,4,7};
// int[] intArray = new int[]{1,2,3,4,5,6};
int[] intArray = new int[]{3,1,2,7,3,8,4,9,5,6};
int totalWater = getArrayWater(intArray);
System.out.println(totalWater);
}
/**
* 求出数组中的水数
*/
private static int getArrayWater(int[] intArray) {
int findMaxValueOfArray = findMaxValueOfArray(intArray);
int findMinValueOfArray = findMinValueOfArray(intArray);
int length = intArray.length;
int totalWater = 0;
// 循环次数就是最大值和最小值的差
for(int i=findMinValueOfArray; i<findMaxValueOfArray; i++){
// 循环构造每一层的01数组
int[] tempArray = new int[length];
for(int j=0; j<length; j++){
if(intArray[j] > i){
tempArray[j] = 1;
}else{
tempArray[j] = 0;
}
}
// 获取每一个01数组的合法0个数
int waterOfOneZeroArray = getWaterOfOneZeroArray(tempArray);
totalWater += waterOfOneZeroArray;
}
return totalWater;
}
/**
* 寻找逻辑是：从左右开始各找一个1，然后这两个1之间的所有0的个数，就是水数
*/
private static int getWaterOfOneZeroArray(int[] tempArray) {
int length = tempArray.length;
int toatalWater = 0;
// 找左边的1
int i = 0;
while(i < length){
if(tempArray[i] == 1){
break;
}
i++;
}
// 从右边开始找1
int j=length-1;
while(j >= i){
if(tempArray[j] == 1){
break;
}
j--;
}
// 找以上两个1之间的0的个数。
if(i == j || i + 1 == j){
return 0;
}else{
for(int k=i+1; k<j; k++){
if(tempArray[k] == 0){
toatalWater++;
}
}
return toatalWater;
}
}
/**
*
* 描述：找出一个数组中的最大值
*/
public static int findMaxValueOfArray(int[] intArray){
int length = intArray.length;
if(length == 0){
return 0;
}else if(length == 1){
return intArray[0];
}else{
int max = intArray[0];
for(int i=1; i<length; i++){
if(intArray[i] > max){
max = intArray[i];
}
}
return max;
}
}
/**
* 找出一个数组中的最小值
*/
public static int findMinValueOfArray(int[] intArray){
int length = intArray.length;
if(length == 0){
return 0;
}else if(length == 1){
return intArray[0];
}else{
int min = intArray[0];
for(int i=1; i<length; i++){
if(intArray[i] < min){
min = intArray[i];
}
}
return min;
}
}
}
以上是关于大数据Hadoop之MapReduce的主要内容，如果未能解决你的问题，请参考以下文章
大数据之Hadoop(MapReduce)：MapReduce扩展案例
大数据之Hadoop(MapReduce): MapReduce概述
大数据之Hadoop(MapReduce)：KeyValueTextInputFormat使用案例
大数据之Hadoop(MapReduce)：数据清洗（ETL）