4、 client请求3台DataNode中的一台A上传数据（本质上是一个RPC调用，建立pipeline），A收到请求会继续调用B，然后B调用C，将整个pipeline建立完成，后逐级返回client；
5、 client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位（默认64K），A收到一个packet就会传给B，B传给C；A每传一个packet会放入一个应答队列等待应答。
6、 数据被分割成一个个packet数据包在pipeline上依次传输，在pipeline反方向上，逐个发送ack（命令正确应答），最终由pipeline中第一个DataNode节点A将pipelineack发送给client;
9、HDFS的API操作9.1、创建maven工程并导入jar包由于cdh版本的所有的软件涉及版权的问题，所以并没有将所有的jar包托管到maven仓库当中去，而是托管在了CDH自己的服务器上面，所以我们默认去maven的仓库下载不到，需要自己手动的添加repository去CDH仓库进行下载，以下两个地址是官方文档说明，请仔细查阅
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo.html
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo_514x.html
<repositories>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0-mr1-cdh5.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.6.0-cdh5.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.6.0-cdh5.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>2.6.0-cdh5.14.0</version>
</dependency>
<!--https://mvnrepository.com/artifact/junit/junit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
<version>RELEASE</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
<!-- <verbal>true</verbal>-->
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<minimizeJar>true</minimizeJar>
</configuration>
</execution>
</executions>
</plugin>
<!-- <plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>cn.itcast.hadoop.db.DBToHdfs2</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>-->
</plugins>
</build>
9.2、使用url的方式访问数据（了解）@Test
public void demo1()throws Exception{
//第一步：注册hdfs 的url，让java代码能够识别hdfs的url形式
URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
InputStream inputStream = null;
FileOutputStream outputStream =null;
//定义文件访问的url地址
Stringurl = "hdfs://192.168.52.100:8020/test/input/install.log";
//打开文件输入流
try {
inputStream = new URL(url).openStream();
outputStream = new FileOutputStream(newFile("c:\\hello.txt"));
IOUtils.copy(inputStream,outputStream);
} catch (IOExceptione) {
e.printStackTrace();
}finally {
IOUtils.closeQuietly(inputStream);
IOUtils.closeQuietly(outputStream);
}
}
如果执行出现以下错误，可以参见资料如何解决，也可以不用理会，不会影响程序的执行。记得配置完成环境变量之后重启开发工具
9.3、使用文件系统方式访问数据（掌握）
在 java 中操作 HDFS，主要涉及以下 Class：
Configuration：该类的对象封转了客户端或者服务器的配置; FileSystem：该类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作，通过 FileSystem 的静态方法 get 获得该对象。
FileSystem fs = FileSystem.get(conf)
get 方法从 conf 中的一个参数fs.defaultFS 的配置值判断具体是什么类型的文件系统。如果我们的代码中没有指定 fs.defaultFS，并且工程 classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml ，默 认 值 为 ： file:/// ， 则 获 取 的 将 不 是 一 个DistributedFileSystem的实例，而是一个本地文件系统的客户端对象
9.4、获取FileSystem的几种方式第一种方式获取FileSystem
@Test
public void getFileSystem() throws URISyntaxException,IOException {
Configuration configuration = new Configuration();
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),configuration);
System.out.println(fileSystem.toString());
}
第二种获取FileSystem类的方式
@Test
public void getFileSystem2() throws URISyntaxException,IOException {
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS","hdfs://192.168.52.100:8020");
FileSystem fileSystem = FileSystem.get(new URI("/"),configuration);
System.out.println(fileSystem.toString());
}
第三种获取FileSystem类的方式
@Test
public void getFileSystem3() throws URISyntaxException,IOException {
Configuration configuration = new Configuration();
FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://192.168.52.100:8020"),configuration);
System.out.println(fileSystem.toString());
}
第四种获取FileSystem类的方式
@Test
public void getFileSystem4() throws Exception{
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS","hdfs://192.168.52.100:8020");
FileSystem fileSystem = FileSystem.newInstance(configuration);
System.out.println(fileSystem.toString());
}
9.5、递归遍历文件系统当中的所有文件通过递归遍历hdfs文件系统
@Test
public void listFile() throws Exception{
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
for (FileStatusfileStatus : fileStatuses) {
if(fileStatus.isDirectory()){
Path path =fileStatus.getPath();
listAllFiles(fileSystem,path);
}else{
System.out.println("文件路径为"+fileStatus.getPath().toString());
}
}
}
public void listAllFiles(FileSystemfileSystem,Path path) throws Exception{
FileStatus[] fileStatuses =fileSystem.listStatus(path);
for (FileStatusfileStatus : fileStatuses) {
if(fileStatus.isDirectory()){
listAllFiles(fileSystem,fileStatus.getPath());
}else{
Path path1 =fileStatus.getPath();
System.out.println("文件路径为"+path1);
}
}
}
官方提供的API直接遍历
/**
* 递归遍历官方提供的API版本
* @throws Exception
*/
@Test
public void listMyFiles()throws Exception{
//获取fileSystem类
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), newConfiguration());
//获取RemoteIterator 得到所有的文件或者文件夹，第一个参数指定遍历的路径，第二个参数表示是否要递归遍历
RemoteIterator<LocatedFileStatus>locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
while (locatedFileStatusRemoteIterator.hasNext()){
LocatedFileStatus next =locatedFileStatusRemoteIterator.next();
System.out.println(next.getPath().toString());
}
fileSystem.close();
}
9.6、下载文件到本地程序执行的main方法
/**
* 拷贝文件的到本地
* @throws Exception
*/
@Test
public void getFileToLocal()throws Exception{
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
FSDataInputStream open = fileSystem.open(new Path("/test/input/install.log"));
FileOutputStream fileOutputStream = new FileOutputStream(newFile("c:\\install.log"));
IOUtils.copy(open,fileOutputStream);
IOUtils.closeQuietly(open);
IOUtils.closeQuietly(fileOutputStream);
fileSystem.close();
}
9.7、hdfs上创建文件夹@Test
public void mkdirs() throws Exception{
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
boolean mkdirs= fileSystem.mkdirs(new Path("/hello/mydir/test"));
fileSystem.close();
}
9.8、hdfs文件上传@Test
public void putData() throws Exception{
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
fileSystem.copyFromLocalFile(new Path("file:///c:\\install.log"),new Path("/hello/mydir/test"));
fileSystem.close();
}
9.9、HDFS权限问题以及伪造用户首先停止hdfs集群，在node01机器上执行以下命令
cd /export/servers/hadoop-2.6.0-cdh5.14.0
sbin/stop-dfs.sh
修改node01机器上的hdfs-site.xml当中的配置文件
cd/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
vim hdfs-site.xml
<property>
<name>dfs.permissions</name>
<value>true</value>
</property>
修改完成之后配置文件发送到其他机器上面去
scp hdfs-site.xml node02:$PWD
scp hdfs-site.xml node03:$PWD
重启hdfs集群
cd /export/servers/hadoop-2.6.0-cdh5.14.0
sbin/start-dfs.sh
随意上传一些文件到我们hadoop集群当中准备测试使用
cd/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
hdfs dfs -mkdir /config
hdfs dfs -put *.xml /config
hdfs dfs -chmod 600 /config/core-site.xml
使用代码准备下载文件
@Test
public void getConfig()
throws Exception{
FileSystem fileSystem = FileSystem.
get(
new URI(
"hdfs://192.168.52.100:8020"),
new Configuration(),
"[url=]root[/url][a1] ");
fileSystem.copyToLocalFile(
new Path(
"/config/core-site.xml"),
new Path(
"file:///c:/core-site.xml"));
fileSystem.close();
}
9.10、HDFS的小文件合并由于hadoop擅长存储大文件，因为大文件的元数据信息比较少，如果hadoop集群当中有大量的小文件，那么每个小文件都需要维护一份元数据信息，会大大的增加集群管理元数据的内存压力，所以在实际工作当中，如果有必要一定要将小文件合并成大文件进行一起处理
在我们的hdfs 的shell命令模式下，可以通过命令行将很多的hdfs文件合并成一个大文件下载到本地，命令如下
cd /export/servers
hdfs dfs -getmerge /config/*.xml ./hello.xml
既然可以在下载的时候将这些小文件合并成一个大文件一起下载，那么肯定就可以在上传的时候将小文件合并到一个大文件里面去
代码如下：
/**
* 将多个本地系统文件，上传到hdfs，并合并成一个大的文件
* @throws Exception
*/
@Test
public void mergeFile() throws Exception{
//获取分布式文件系统
FileSystemfileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), new Configuration(),"root");
FSDataOutputStream outputStream =fileSystem.create(new Path("/bigfile.xml"));
//获取本地文件系统
LocalFileSystemlocal = FileSystem.getLocal(new Configuration());
//通过本地文件系统获取文件列表，为一个集合
FileStatus[]fileStatuses = local.listStatus(new Path("file:///F:\\传智播客大数据离线阶段课程资料\\3、大数据离线第三天\\上传小文件合并"));
for (FileStatusfileStatus : fileStatuses) {
FSDataInputStream inputStream =local.open(fileStatus.getPath());
IOUtils.copy(inputStream,outputStream);
IOUtils.closeQuietly(inputStream);
}
IOUtils.closeQuietly(outputStream);
local.close();
fileSystem.close();
}
收藏
淘帖
踩
