首先要在windows下解压一个windows版本的hadoop
然后在配置他的环境变量，同时要把hadoop的share目录下的hadoop下的相关jar包拷贝到esclipe
然后build path
下面上代码
我要评论import java.io.bufferedreader;
import java.io.fileinputstream;
import java.io.ioexception;
import java.io.inputstreamreader;
import java.net.uri;
import java.net.urisyntaxexception;
import java.util.arrays;
import org.apache.hadoop.conf.configuration;
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.locatedfilestatus;
import org.apache.hadoop.fs.path;
import org.apache.hadoop.fs.remoteiterator;
import org.junit.before;
import org.junit.test;
public class hdfsclientdemo {
public static void main(string[] args) throws exception {
/**
* configuration参数对象的机制：
* 构造时，会加载jar包中的默认配置 xx-default.xml
* 再加载 用户配置xx-site.xml ，覆盖掉默认参数
* 构造完成之后，还可以conf.set("p","v")，会再次覆盖用户配置文件中的参数值
*/
// 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", "2");
// 指定本客户端上传文件到hdfs时切块的规格大小：64m
conf.set("dfs.blocksize", "64m");
// 构造一个访问指定hdfs系统的客户端对象: 参数1:——hdfs系统的uri，参数2：——客户端要特别指定的参数，参数3：客户端的身份（用户名）
filesystem fs = filesystem.get(new uri("hdfs://172.31.2.38:9000/"), conf, "root");
// 上传一个文件到hdfs中
fs.copyfromlocalfile(new path("d:/install-pkgs/hbase-1.2.1-bin.tar.gz"), new path("/aaa/"));
fs.close();
}
filesystem fs = null;
@before
public void init() throws exception{
configuration conf = new configuration();
conf.set("dfs.replication", "2");
conf.set("dfs.blocksize", "64m");
fs = filesystem.get(new uri("hdfs://172.31.2.38:9000/"), conf, "root");
}
/**
* 从hdfs中下载文件到客户端本地磁盘
* @throws ioexception
* @throws illegalargumentexception
*/
@test
public void testget() throws illegalargumentexception, ioexception{
fs.copytolocalfile(new path("/test"), new path("d:/"));
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<locatedfilestatus> 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中的文件的内容
*
* @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中文件的指定偏移量范围的内容
*
*
*
* @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();
}
/**
* 往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();
}
}
练习：从一个文件里面不断地采集日志上传到hdfs里面
1.流程介绍
---启动一个定时任务
--定时探测日志原目录
--获取文件上传到一个待上传的临时目录
--逐一上传到hdfs目标路径，同时移动到备份目录里
--启动一个定时任务：
--探测备份目录中的备份数据是否已经超出，如果超出就删除
主类为：
import java.util.timer;
public class datacollectmain {
public static void main(string[] args) {
timer timer = new timer();
//第一个为task类，第二个开始时间 第三个没隔多久执行一次
timer.schedule(new collecttask(), 0, 60*60*1000l);
timer.schedule(new backupcleantask(), 0, 60*60*1000l);
}
}
collecttask类：
这个类要继承timertask，重写run方法，主要内容就是不断收集日志文件
package cn.edu360.hdfs.datacollect;
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() {
try {
// 获取配置参数
properties props = propertyholderlazy.getprops();
// 获取本次采集时的日期
simpledateformat sdf = new simpledateformat("yyyy-mm-dd-hh");
string day = sdf.format(new date());
file srcdir = new file("d:/logs/accesslog");
// 列出日志源目录中需要采集的文件
//里面传了一个文件过滤器，重写accept方法,return true就要
file[] listfiles = srcdir.listfiles(new filenamefilter() {
@override
public boolean accept(file dir, string name) {
if (name.startswith("access.log")) {
return true;
}
return false;
}
});
// 将要采集的文件移动到待上传临时目录
file touploaddir = new file("d:/logs/toupload");
for (file file : listfiles) {
//这里如果是 file.renameto(touploaddir)是不对的，因为会生成一个toupload的文件而不是文件夹
//要用renameto的话你要自己加上文件的新名字比较麻烦
//用fileutiles是对file操作的一些工具类
//第一个目标文件，第二个路径，第三个是否存在覆盖
fileutils.movefiletodirectory(file, touploaddir, true);
}
// 构造一个hdfs的客户端对象
filesystem fs = filesystem.get(new uri("hdfs://hdp-01:9000"), new configuration(), "root");
file[] touploadfiles = touploaddir.listfiles();
// 检查hdfs中的日期目录是否存在，如果不存在，则创建
path hdfsdestpath = new path("/logs" + day);
if (!fs.exists(hdfsdestpath)) {
fs.mkdirs(hdfsdestpath);
}
// 检查本地的备份目录是否存在，如果不存在，则创建
file backupdir = new file("d:/logs/backup" + day + "/");
if (!backupdir.exists()) {
backupdir.mkdirs();
}
for (file file : touploadfiles) {
// 传输文件到hdfs并改名access_log_
fs.copyfromlocalfile(new path(file.getabsolutepath()), new path("/logs"+day+"/access_log_"+uuid.randomuuid()+".log"));
// 将传输完成的文件移动到备份目录
//注意这里依然不要用renameto
fileutils.movefiletodirectory(file, backupdir, true);
}
} catch (exception e) {
e.printstacktrace();
}
}
/**
* 读取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();
}
/**
* 往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();
}
}
backupcleantask类
package cn.edu360.hdfs.datacollect;
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();
}
}
}
hdfs中namenode中储存元数据（对数据的描述信息）是在内存中以树的形式储存的，并且每隔一段时间都会把这些元数据序列化到磁盘中。序列化的东西在磁盘中叫 fsimage文件。
元数据可能会很大很大，所以只能是定期的序列化
问题1：序列化的时候，发生了元数据的修改怎么办
答：namenode会把每次用户的操作都记录下来，记录成日志文件，存在edits日志文件中
其中edits日志文件也会像log4j滚动日志文件一样，当文件太大的时候会另起一个文件并改名字
问题2：当edits文件太多的时候，一次宕机也会花大量的时间从edits里恢复，怎么办
答：会定期吧edits文件重放fsimage文件，并记录edits的编号，把那些重放过的日志文件给删除。这样也相当于重新序列化了，
所以namenode并不会做这样的事情，是由secondary node做的，他会定期吧namenode的fsimage文件和edits文件下载下来
并把fsimage文件反序列化，并且读日志文件更新元数据，然后序列化到磁盘，然后把他上传给namenode。
这个机制叫做checkpoint机制
这里secondarynode 相当一一个小秘书
额外知识点
注意，在windows里面不要写有些路径不要写绝对路径，因为程序放到linux下面可能会找不到，因此报错
一般使用class加载器，这样当这个class加载的时候就会知道这个class在哪
类加载器的一些使用例子
比如我加载一个配置文件，为了避免出现绝对路径，我们可以是用类加载器
properties props = new properties();
//加载配置文件，这样写的目的是为了避免在windows里出现绝对路径，用类加载器，再把文件传化成流
props.load(hdfswordcount.class.getclassloader().getresourceasstream("job.properties"));
而对于一些功能性的类，我们最好在写逻辑的时候也不要直接去导入这个包，而是使用class.forname
//这样不直接导入这个包，直接用类加载器，是面向接口编程的一种思想。这里我并不是在开始import xxxx.mapper,这里mapper是一个接口，这里我用了多态
class<?> mapper_class = class.forname(props.getproperty("mapper_class"));
mapper mapper = (mapper) mapper_class.newinstance();
单例模式
https://www.cnblogs.com/crazy-wang-android/p/9054771.html
只有个一实例，必须自己创建自己这个实例，必须为别人提供这个实例
饿汉式单例：就算没有人调用这个class，他也会加载进去；
如对于一个配置文件的加载
import java.util.properties;
/**
* 单例设计模式，方式一： 饿汉式单例
*
*/
public class propertyholderhungery {
private static properties prop = new properties();
static {
try {
//将一个文件prop.load(stram)
//这里面如果传一个io流不好，因为要用到绝对路径，使用了类加载器 这种不管有没有使用这个类都会加载
prop.load(propertyholderhungery.class.getclassloader().getresourceasstream("collect.properties"));
} catch (exception e) {
}
}
public static properties getprops() throws exception {
return prop;
}
}
懒汉式：只有调用的时候才会有，但会有线程安全问题
/**
* 单例模式：懒汉式——考虑了线程安全
* */
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;
}
}
如对本文有疑问， 点击进行留言回复！！
HBase Filter 过滤器之FamilyFilter详解
去 HBase，Kylin on Parquet 性能表现如何？
如何找到Hive提交的SQL相对应的Yarn程序的applicationId
网友评论