hadoop之HDFS学习笔记（一）

[-setfacl [-R] [{-b|-k} {-m|-x } ]|[--set ]]

[-setfattr {-n name [-v value] | -x name} ] [-setrep [-R] [-w] ...] [-stat [format] ...] [-tail [-f] ]

[-test -[defsz] ]

[-text [-ignoreCrc] ...] [-touchz ...]

[-truncate [-w] ...] [-usage [cmd ...]]

6、hdfs的java客户端编程HDFS客户端编程应⽤场景：数据采集业务系统中⽇志⽣成机制

数据采集程序其实就是通过对java客户端编程，将数据不断的上传到hdfs。在windows开发环境中做⼀些准备⼯作：

1、在windows的某个路径中解压⼀份windows版本的hadoop安装包

2、将解压出的hadoop⽬录配置到windows的环境变量中：HADOOP_HOME

原因：若不配置环境变量，会在下载hdfs⽂件是出错，是由于使⽤hadoop的FileSystem保存⽂件到本地的时候出于效率的考虑，会使⽤hadoop安装包中的c语⾔库，显然没有配置hadoop环境变量时是找不到该c语⾔类库中的⽂件的；然⽽上传⽂件到hdfs没有类似问题；

6.1、核⼼代码1、将hdfs客户端开发所需的jar导⼊⼯程（jar包可在hadoop安装包中找到common和hdfs）2、写代码

6.1.1、获取hdfs客户端要点：要对hdfs中的⽂件进⾏操作，代码中⾸先需要获得⼀个hdfs的客户端对象Configuration conf = new Configuration();

FileSystem fs = FileSystem.get(new URI(\"hdfs://hdp-01:9000\"),conf,\"root\");

完整代码如下：

/**

* Configuration参数对象的机制：

* 构造时，会加载jar包中的默认配置 xx-default.xml（core-default.xmlhdfs-default.xml） * 再加载 ⽤户配置xx-site.xml ，覆盖掉默认参数

* 构造完成之后，还可以conf.set(\"p\，会再次覆盖⽤户配置⽂件中的参数值 */

// new Configuration()会从项⽬的classpath中加载core-default.xml hdfs-default.xml core-site.xml hdfs-site.xml等⽂件 Configuration conf = new Configuration();

// 指定本客户端上传⽂件到hdfs时需要保存的副本数为：2 conf.set(\"dfs.replication\

// 指定本客户端上传⽂件到hdfs时切块的规格⼤⼩：64M conf.set(\"dfs.blocksize\

// 构造⼀个访问指定HDFS系统的客户端对象: 参数1:——HDFS系统的URI，参数2：——客户端要特别指定的参数，参数3：客户端的⾝份（⽤户名） FileSystem fs = FileSystem.get(new URI(\"hdfs://hdp-01:9000/\"), conf, \"root\");

// 上传⼀个⽂件到HDFS中

fs.copyFromLocalFile(new Path(\"D:/install-pkgs/hbase-1.2.1-bin.tar.gz\"), new Path(\"/aaa/\"));

fs.close();View Code

6.1.2、对⽂件进⾏操作上传、下载⽂件；⽂件夹的创建和删除、⽂件的移动和复制、查看⽂件夹和⽂件等。3、利⽤fs对象的⽅法进⾏⽂件操作⽅法均与命令⾏⽅法对应，⽐如：上传⽂件

fs.copyFromLocalFile(new Path(\"本地路径\"),new Path(\"hdfs的路径\")); 下载⽂件

fs.copyToLocalFile(new Path(\"hdfs的路径\"),new Path(\"本地路径\")) 对⽂件的增删改查如下，对⽂件数据的操作后续介绍。

FileSystem fs = null;

@Before

public void init() throws Exception{

Configuration conf = new Configuration(); conf.set(\"dfs.replication\ conf.set(\"dfs.blocksize\

fs = FileSystem.get(new URI(\"hdfs://hdp-01:9000/\"), conf, \"root\"); } /**

* 从HDFS中下载⽂件到客户端本地磁盘 * @throws IOException

* @throws IllegalArgumentException */

@Test

public void testGet() throws IllegalArgumentException, IOException{

fs.copyToLocalFile(new Path(\"/hdp20-05.txt\"), new Path(\"f:/\")); fs.close(); } /**

* 在hdfs内部移动⽂件\\修改名称 */

@Test

public void testRename() throws Exception{

fs.rename(new Path(\"/install.log\"), new Path(\"/aaa/in.log\"));

fs.close(); } /**

* 在hdfs中创建⽂件夹 */

@Test

public void testMkdir() throws Exception{

fs.mkdirs(new Path(\"/xx/yy/zz\"));

fs.close(); } /**

* 在hdfs中删除⽂件或⽂件夹 */

@Test

public void testRm() throws Exception{

fs.delete(new Path(\"/aaa\"), true);

fs.close(); } /**

* 查询hdfs指定⽬录下的⽂件信息 */

@Test

public void testLs() throws Exception{

// 只查询⽂件的信息,不返回⽂件夹的信息

RemoteIterator iter = fs.listFiles(new Path(\"/\"), true);

while(iter.hasNext()){

LocatedFileStatus status = iter.next();

System.out.println(\"⽂件全路径：\"+status.getPath()); System.out.println(\"块⼤⼩：\"+status.getBlockSize()); System.out.println(\"⽂件长度：\"+status.getLen());

System.out.println(\"副本数量：\"+status.getReplication());

System.out.println(\"块信息：\"+Arrays.toString(status.getBlockLocations()));

System.out.println(\"--------------------------------\"); }

fs.close(); } /**

* 查询hdfs指定⽬录下的⽂件和⽂件夹信息 */

@Test

public void testLs2() throws Exception{

FileStatus[] listStatus = fs.listStatus(new Path(\"/\"));

for(FileStatus status:listStatus){

System.out.println(\"⽂件全路径：\"+status.getPath());

System.out.println(status.isDirectory()?\"这是⽂件夹\":\"这是⽂件\"); System.out.println(\"块⼤⼩：\"+status.getBlockSize()); System.out.println(\"⽂件长度：\"+status.getLen());

System.out.println(\"副本数量：\"+status.getReplication());

System.out.println(\"--------------------------------\"); }

fs.close(); }View Code

6.1.3、对⽂件数据进⾏操作 同过客户端使⽤open打开流对象来读取hdfs中⽂件的具体数据，包括指定偏移量来读取特定范围的数据；通过客户端向hdfs⽂件追加数据。

/**

* 读取hdfs中的⽂件的内容 *

* @throws IOException

* @throws IllegalArgumentException */

@Test

public void testReadData() throws IllegalArgumentException, IOException { FSDataInputStream in = fs.open(new Path(\"/test.txt\"));

BufferedReader br = new BufferedReader(new InputStreamReader(in, \"utf-8\")); String line = null;

while ((line = br.readLine()) != null) { System.out.println(line); }

br.close(); in.close(); fs.close(); }

/**

* 读取hdfs中⽂件的指定偏移量范围的内容 * *

* 作业题：⽤本例中的知识，实现读取⼀个⽂本⽂件中的指定BLOCK块中的所有数据 *

* @throws IOException

* @throws IllegalArgumentException */

@Test

public void testRandomReadData() throws IllegalArgumentException, IOException { FSDataInputStream in = fs.open(new Path(\"/xx.dat\")); // 将读取的起始位置进⾏指定 in.seek(12); // 读16个字节

byte[] buf = new byte[16]; in.read(buf);

System.out.println(new String(buf)); in.close(); fs.close(); }View Code

写数据，create提供了丰富的重载函数，轻松实现覆盖，追加，以及指定缓存⼤⼩，副本数量等等信息。

/**

* 往hdfs中的⽂件写内容 *

* @throws IOException

* @throws IllegalArgumentException */

@Test

public void testWriteData() throws IllegalArgumentException, IOException { FSDataOutputStream out = fs.create(new Path(\"/zz.jpg\"), false); // D:\\images\\006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg

FileInputStream in = new FileInputStream(\"D:/images/006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg\"); byte[] buf = new byte[1024]; int read = 0;

while ((read = in.read(buf)) != -1) { out.write(buf,0,read); }

in.close(); out.close(); fs.close(); }View Code

7、HDFS实例hdfs版本wordcount程序。任务描述:

1、从hdfs⽂件中读取数据，每次读取⼀⾏数据；

2、将数据交给具体的单词统计业务去作业（使⽤⾯向接⼝编程，当业务逻辑改变时，⽆需修改主程序代码）；3、并将该⾏数据产⽣的结果存⼊缓存中（可以⽤hashmap模拟）数据采集设计：

1、流程

启动⼀个定时任务：——定时探测⽇志源⽬录——获取需要采集的⽂件

——移动这些⽂件到⼀个待上传临时⽬录

——遍历待上传⽬录中各⽂件，逐⼀传输到HDFS的⽬标路径，同时将传输完成的⽂件移动到备份⽬录启动⼀个定时任务：

——探测备份⽬录中的备份数据，检查是否已超出最长备份时长，如果超出，则删除

2、规划各种路径

⽇志源路径： d:/logs/accesslog/待上传临时⽬录： d:/logs/toupload/备份⽬录： d:/logs/backup/⽇期/HDFS存储路径： /logs/⽇期

HDFS中的⽂件的前缀：access_log_HDFS中的⽂件的后缀：.log将路径配置写⼊属性⽂件

MAPPER_CLASS=cn.edu360.hdfs.wordcount.CaseIgnorWcountMapperINPUT_PATH=/wordcount/input

OUTPUT_PATH=/wordcount/output2

View Code

主程序代码⽰例：

import java.io.BufferedReader;import java.io.InputStreamReader;import java.net.URI;

import java.util.HashMap;import java.util.Map.Entry;import java.util.Properties;import java.util.Set;

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.LocatedFileStatus;import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

public class HdfsWordcount {

public static void main(String[] args) throws Exception{ /**

* 初始化⼯作 */

Properties props = new Properties();

props.load(HdfsWordcount.class.getClassLoader().getResourceAsStream(\"job.properties\"));

Path input = new Path(props.getProperty(\"INPUT_PATH\")); Path output = new Path(props.getProperty(\"OUTPUT_PATH\"));

Class mapper_class = Class.forName(props.getProperty(\"MAPPER_CLASS\")); Mapper mapper = (Mapper) mapper_class.newInstance();

Context context = new Context(); /**

* 处理数据 */

FileSystem fs = FileSystem.get(new URI(\"hdfs://hdp-01:9000\"), new Configuration(), \"root\"); RemoteIterator iter = fs.listFiles(input, false);

while(iter.hasNext()){

LocatedFileStatus file = iter.next();

FSDataInputStream in = fs.open(file.getPath());

BufferedReader br = new BufferedReader(new InputStreamReader(in)); String line = null; // 逐⾏读取

while ((line = br.readLine()) != null) {

// 调⽤⼀个⽅法对每⼀⾏进⾏业务处理 mapper.map(line, context); }

br.close(); in.close(); } /**

* 输出结果 */

HashMap contextMap = context.getContextMap();

if(fs.exists(output)){

throw new RuntimeException(\"指定的输出⽬录已存在，请更换......!\"); }

FSDataOutputStream out = fs.create(new Path(output,new Path(\"res.dat\")));

Set> entrySet = contextMap.entrySet(); for (Entry entry : entrySet) {

out.write((entry.getKey().toString()+\"\\"+entry.getValue()+\"\\n\").getBytes()); }

out.close();

fs.close();

System.out.println(\"恭喜！数据统计完成.....\"); } }

View Code

⾃定义的业务接⼝

public interface Mapper {

public void map(String line,Context context); }

View Code

业务实现类1

public class WordCountMapper implements Mapper{ @Override

public void map(String line, Context context) {

String[] words = line.split(\" \");

for (String word : words) {

Object value = context.get(word); if(null==value){

context.write(word, 1); }else{

int v = (int)value;

context.write(word, v+1); } } }}

View Code

业务实现类2

public class CaseIgnorWcountMapper implements Mapper { @Override

public void map(String line, Context context) { String[] words = line.toUpperCase().split(\" \"); for (String word : words) {

Object value = context.get(word); if (null == value) {

context.write(word, 1); } else {

int v = (int) value;

context.write(word, v + 1); } } }}

View Code

缓存模拟

import java.util.HashMap;

public class Context {

private HashMap contextMap = new HashMap<>();

public void write(Object key,Object value){

contextMap.put(key, value); }

public Object get(Object key){

return contextMap.get(key); }

public HashMap getContextMap(){ return contextMap; }}

View Code

8、实战描述 需求描述：

在业务系统的服务器上，业务程序会不断⽣成业务⽇志（⽐如⽹站的页⾯访问⽇志）业务⽇志是⽤log4j⽣成的，会不断地切出⽇志⽂件

需要定期（⽐如每⼩时）从业务服务器上的⽇志⽬录中，探测需要采集的⽇志⽂件(access.log，不是直接采集数据)，发往HDFS

注意点：业务服务器可能有多台(hdfs上的⽂件名不能直接⽤⽇志服务器上的⽂件名)当天采集到的⽇志要放在hdfs的当天⽬录中

采集完成的⽇志⽂件，需要移动到到⽇志服务器的⼀个备份⽬录中

定期检查（⼀⼩时检查⼀次）备份⽬录，将备份时长超出24⼩时的⽇志⽂件清除Timer timer = new Timer()timer.schedual() 简易版⽇志采集主程序

import java.util.Timer;

public class DataCollectMain {

public static void main(String[] args) {

Timer timer = new Timer();

timer.schedule(new CollectTask(), 0, 60*60*1000L);

timer.schedule(new BackupCleanTask(), 0, 60*60*1000L); }}

View Code

⽇志收集定时任务类

import java.io.File;

import java.io.FilenameFilter;import java.net.URI;

import java.text.SimpleDateFormat;import java.util.Arrays;import java.util.Date;

import java.util.Properties;import java.util.TimerTask;import java.util.UUID;

import org.apache.commons.io.FileUtils;

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.log4j.Logger;

public class CollectTask extends TimerTask { @Override

public void run() {

/**

* ——定时探测⽇志源⽬录 ——获取需要采集的⽂件 ——移动这些⽂件到⼀个待上传临时⽬录

* ——遍历待上传⽬录中各⽂件，逐⼀传输到HDFS的⽬标路径，同时将传输完成的⽂件移动到备份⽬录 * */ try {

// 获取配置参数

Properties props = PropertyHolderLazy.getProps(); // 构造⼀个log4j⽇志对象

Logger logger = Logger.getLogger(\"logRollingFile\");

// 获取本次采集时的⽇期

SimpleDateFormat sdf = new SimpleDateFormat(\"yyyy-MM-dd-HH\"); String day = sdf.format(new Date());

File srcDir = new File(props.getProperty(Constants.LOG_SOURCE_DIR)); // 列出⽇志源⽬录中需要采集的⽂件

File[] listFiles = srcDir.listFiles(new FilenameFilter() {

@Override

public boolean accept(File dir, String name) {

if (name.startsWith(props.getProperty(Constants.LOG_LEGAL_PREFIX))) { return true; }

return false; } });

// 记录⽇志

logger.info(\"探测到如下⽂件需要采集：\" + Arrays.toString(listFiles));

// 将要采集的⽂件移动到待上传临时⽬录

File toUploadDir = new File(props.getProperty(Constants.LOG_TOUPLOAD_DIR)); for (File file : listFiles) {

FileUtils.moveFileToDirectory(file, toUploadDir, true); }

// 记录⽇志

logger.info(\"上述⽂件移动到了待上传⽬录\" + toUploadDir.getAbsolutePath()); // 构造⼀个HDFS的客户端对象

FileSystem fs = FileSystem.get(new URI(props.getProperty(Constants.HDFS_URI)), new Configuration(), \"root\"); File[] toUploadFiles = toUploadDir.listFiles();

// 检查HDFS中的⽇期⽬录是否存在，如果不存在，则创建

Path hdfsDestPath = new Path(props.getProperty(Constants.HDFS_DEST_BASE_DIR) + day); if (!fs.exists(hdfsDestPath)) { fs.mkdirs(hdfsDestPath); }

// 检查本地的备份⽬录是否存在，如果不存在，则创建

File backupDir = new File(props.getProperty(Constants.LOG_BACKUP_BASE_DIR) + day + \"/\"); if (!backupDir.exists()) { backupDir.mkdirs(); }

for (File file : toUploadFiles) {

// 传输⽂件到HDFS并改名access_log_

Path destPath = new Path(hdfsDestPath + \"/\" + UUID.randomUUID() + props.getProperty(Constants.HDFS_FILE_SUFFIX)); fs.copyFromLocalFile(new Path(file.getAbsolutePath()), destPath);

// 记录⽇志

logger.info(\"⽂件传输到HDFS完成：\" + file.getAbsolutePath() + \"-->\" + destPath); // 将传输完成的⽂件移动到备份⽬录

FileUtils.moveFileToDirectory(file, backupDir, true);

// 记录⽇志

logger.info(\"⽂件备份完成：\" + file.getAbsolutePath() + \"-->\" + backupDir); }

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

View Code

定期清理过时备份⽇志

import java.io.File;

import java.text.SimpleDateFormat;import java.util.Date;

import java.util.TimerTask;

import org.apache.commons.io.FileUtils;

public class BackupCleanTask extends TimerTask { @Override

public void run() {

SimpleDateFormat sdf = new SimpleDateFormat(\"yyyy-MM-dd-HH\"); long now = new Date().getTime(); try {

// 探测本地备份⽬录

File backupBaseDir = new File(\"d:/logs/backup/\"); File[] dayBackDir = backupBaseDir.listFiles(); // 判断备份⽇期⼦⽬录是否已超24⼩时 for (File dir : dayBackDir) {

long time = sdf.parse(dir.getName()).getTime(); if(now-time>24*60*60*1000L){ FileUtils.deleteDirectory(dir); } }

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

View Code

配置信息提取到属性配置⽂件中，并写成常量，以单例设计模式去加载配置信息。

LOG_SOURCE_DIR=d:/logs/accesslog/LOG_TOUPLOAD_DIR=d:/logs/toupload/LOG_BACKUP_BASE_DIR=d:/logs/backup/LOG_BACKUP_TIMEOUT=24

LOG_LEGAL_PREFIX=access.log.HDFS_URI=hdfs://hdp-01:9000/HDFS_DEST_BASE_DIR=/logs/HDFS_FILE_PREFIX=access_log_HDFS_FILE_SUFFIX=.logView Code

public class Constants {

/**

* ⽇志源⽬录参数key */

public static final String LOG_SOURCE_DIR = \"LOG_SOURCE_DIR\"; /**

* ⽇志待上传⽬录参数key */

public static final String LOG_TOUPLOAD_DIR = \"LOG_TOUPLOAD_DIR\";

public static final String LOG_BACKUP_BASE_DIR = \"LOG_BACKUP_BASE_DIR\";

public static final String LOG_BACKUP_TIMEOUT = \"LOG_BACKUP_TIMEOUT\";

public static final String LOG_LEGAL_PREFIX = \"LOG_LEGAL_PREFIX\";

public static final String HDFS_URI = \"HDFS_URI\";

public static final String HDFS_DEST_BASE_DIR = \"HDFS_DEST_BASE_DIR\";

public static final String HDFS_FILE_PREFIX = \"HDFS_FILE_PREFIX\";

public static final String HDFS_FILE_SUFFIX = \"HDFS_FILE_SUFFIX\";}

View Code

import java.util.Properties;

/**

* 单例模式：懒汉式——考虑了线程安全 * @author ThinkPad * */

public class PropertyHolderLazy { private static Properties prop = null;

public static Properties getProps() throws Exception { if (prop == null) {

synchronized (PropertyHolderLazy.class) { if (prop == null) {

prop = new Properties();

prop.load(PropertyHolderLazy.class.getClassLoader().getResourceAsStream(\"collect.properties\")); } } }

return prop; }}

View Code

import java.util.Properties;

/**

* 单例设计模式，⽅式⼀： 饿汉式单例 * @author ThinkPad * */

public class PropertyHolderHungery {

private static Properties prop = new Properties();

static { try {

prop.load(PropertyHolderHungery.class.getClassLoader().getResourceAsStream(\"collect.properties\")); } catch (Exception e) { } }

public static Properties getProps() throws Exception { return prop; }}

View Code

9、总结hdfs有服务端和客户端；服务端：

　　成员：namenode 管理元数据，datanode存储数据

　　配置：需要指定使⽤的⽂件系统（默认的配置在core-default.xml，为本地⽂件系统，需要修改服务器core-site.xml修改为hdfs⽂件系统，并指定namenode），namenode和datanode的

⼯作⽬录（服务器的默认配置在hdfs-default.xml中，默认是在/temp下，需要就该hdfs-site.xml来覆盖默认值。）；　　细节：第⼀次启动集群时需要格式化namenode客户端：

　　形式：⽹页端，命令⾏，java客户端api；客户端可以运⾏在任何地⽅。

　　功能：指定上传⽂件的信息报括切块⼤⼩（hdfs-default.xml中默认值128m，可以在hdfs-site.xml中修改，也可以咋java api 中创建客户端对象的时候指定，总之由客户端来指定），副本数量（hdfs-default.xml中默认值3，同样可以修改覆盖）；完成hdfs中⽂件的系列操作，如上传，下载

虽然服务端和客户端的共⽤配置 core-default.xml core-site.xml；hdfs-default.xml hdfs-site.xml，但是不同的程序所需要的参数不同，只不过为了⽅便，所有参数都写在⼀个⽂件中了。即是在服务器的hdfs-site.xml中配置了切块⼤⼩和副本数量，服务器的namenode和datanode根本不关⼼也不使⽤这些参数，只有启动服务器上的命令⾏客户端时，该参数才可能起作⽤。