hadoop写文本文件java

① HDFS文件

Hadoop支持的文件系统由很多（见下图），HDFS只是其中一种实现。java抽象类 org.apache.hadoop.fs.FileSystem 定义了Hadoop中一个文件系统的客户端接口，并且该抽象类有几个具体实现。Hadoop一般使用URI（下图）方案来选取合适的文件系统实例进行交互。

特别的，HDFS文件系统的操作可以使用 FsSystem shell 、客户端（http rest api、Java api、C api等）。

FsSystem shell 的用法基本同本地shell类似，命令可参考 FsSystem shell

Hadoop是用Java写的，通过Java Api（ FileSystem 类）可以调用大部分Hadoop文件系统的交互操作。更详细的介绍可参考 hadoop Filesystem 。

非Java开发的应用可以使用由WebHDFS协议提供的HTTP REST API，但是HTTP比原生的Java客户端要慢，所以不到万不得已尽量不要使用HTTP传输特大数据。通过HTTP来访问HDFS有两种方法：

两种如图

在第一种情况中，namenode和datanode内嵌的web服务作为WebHDFS的端节点运行（是否启用WebHDFS可通过dfs.webhdfs.enabled设置，默认为true）。文件元数据在namenode上，文件读写操作首先被发往namenode，有namenode发送一个HTTP重定向至某个客户端，指示以流的方式传输文件数据的目的或源datanode。

第二种方法依靠一个或多个独立代理服务器通过HTTP访问HDFS。所有集群的网络通信都需要通过代理，因此客户端从来不直接访问namenode或datanode。使用代理后可以使用更严格的防火墙策略和带宽策略。

HttpFs代理提供和WebHDFS相同的HTTP接口，这样客户端能够通过webhdfs URI访问接口。HttpFS代理启动独立于namenode和datanode的守护进程，使用httpfs.sh 脚本，默认在一个不同的端口上监听（14000）。

下图描述了

读文件时客户端与 HDFS 中的 namenode， datanode 之间的数据流动。

对上图的解释如下：

在读取过程中， 如果 FSDataInputStream 在和一个 datanode 进行交流时出现了一个错误，他就去试一试下一个最接近的块，他当然也会记住刚才发生错误的 datanode 以至于之后不会再在这个 datanode 上进行没必要的尝试。 DFSInputStream 也会在 datanode 上传输出的数据上核查检查数(checknums).如果损坏的块被发现了， DFSInputStream 就试图从另一个拥有备份的 datanode 中去读取备份块中的数据。

在这个设计中一个重要的方面就是客户端直接从 datanode 上检索数据，并通过 namenode 指导来得到每一个块的最佳 datanode。这种设计允许 HDFS 扩展大量的并发客户端，因为数据传输只是集群上的所有 datanode 展开的。期间，namenode 仅仅只需要服务于获取块位置的请求（块位置信息是存放在内存中，所以效率很高）。如果不这样设计，随着客户端数据量的增长，数据服务就会很快成为一个瓶颈。

我们知道，相对于客户端(之后就是 maprece task 了)，块的位置有以下可能性：

我们认为他们对于客户端的带宽递减，距离递增（括号中表示距离）。示意图如下：

如果集群中的机器都在同一个机架上，我们无需其他配置，若集群比较复杂，由于hadoop无法自动发现网络拓扑，所以需要额外配置网络拓扑。

基本读取程序，将文件内容输出到console

FileSystemCat

随机读取

展开原码

下图描述了写文件时客户端与 HDFS 中的 namenode， datanode 之间的数据流动。

对上图的解释如下：

如果在任何一个 datanode 在写入数据的时候失败了，接下来所做的一切对客户端都是透明的：首先， pipeline 被关闭，在确认队列中的剩下的包会被添加进数据队列的起始位置上，以至于在失败的节点下游的任 何节点都不会丢失任何的包。然后与 namenode 联系后，当前在一个好的 datanode 会联系 namenode， 给失败节点上还未写完的块生成一个新的标识ID， 以至于如果这个失败的 datanode 不久后恢复了，这个不完整的块将会被删除。失败节点会从 pipeline 中移除，然后剩下两个好的 datanode 会组成一个的新的 pipeline ，剩下的 这些块的包(也就是刚才放在数据队列队首的包)会继续写进 pipeline 中好的 datanode 中。最后，namenode 注意到块备份数小于规定的备份数，他就安排在另一个节点上创建完成备份，直接从已有的块中复制就可以。然后一直到满足了备份数( dfs.replication )。如果有多个节点的写入失败了，如果满足了最小备份数的设置( dfs.namenode.repliction.min ),写入也将会成功，然后剩下的备份会被集群异步的执行备份，直到满足了备份数( dfs.replication )。

创建目录

文件压缩有两大好处：

Hadoop 对于压缩格式的是自动识别。如果我们压缩的文件有相应压缩格式的扩展名（比如 lzo，gz，bzip2 等）。Hadoop 会根据压缩格式的扩展名自动选择相对应的解码器来解压数据，此过程完全是 Hadoop 自动处理，我们只需要确保输入的压缩文件有扩展名。

Hadoop中有多种压缩格式、算法和工具，下图列出了常用的压缩方法。

表中的“是否可切分”表示对应的压缩算法是否支持切分，也就是说是否可以搜索数据流的任意位置并进一步往下读取数据，可切分的压缩格式尤其适合MapRece。

所有的压缩算法都需要权衡空间/时间：压缩和解压缩速度更快，其代价通常是只能节省少量的空间。不同的压缩工具有不同的特性：

更详细的比较如下

1.压缩性能比较

2.优缺点

另外使用hadoop原生（native）类库比其他java实现有更快的压缩和解压缩速度。特征比较如下：

使用容器文件格式结合压缩算法也能更好的提高效率。顺序文件、Arvo文件、ORCFiles、Parqurt文件同时支持压缩和切分。

压缩举例（Java）

压缩

解压缩

六、文件序列化

序列化是指将结构化数据转换为字节流以便在网络上传输或写到磁盘进行永久存储。反序列化狮子将字节流转换回结构化对象的逆过程。

序列化用于分布式数据处理的两大领域：进程间通信和永久存储。

对序列化的要求时是格式紧凑（高效使用存储空间）、快速（读写效率高）、可扩展（可以透明地读取老格式数据）且可以互操作（可以使用不同的语言读写数据）。

Hadoop使用的是自己的序列化格式 Writable ，它绝对紧凑、速度快，但不太容易用java以外的语言进行扩展或使用。

当然，用户也可以使用其他序列化框架或者自定义序列化方式，如 Avro 框架。

Hadoop内部还使用了 Apache Thrift 和 Protocal Buffers 来实现RPC和数据交换。

② Hadoop读写文件时内部工作机制是怎样的

客户端通过调用FileSystem对象（对应于HDFS文件系统，调用DistributedFileSystem对象）的open()方法来打开文件（也即图中的第一步），DistributedFileSystem通过RPC（Remote Procere Call）调用询问NameNode来得到此文件最开始几个block的文件位置（第二步）。对每一个block来说，namenode返回拥有此block备份的所有namenode的地址信息（按集群的拓扑网络中与客户端距离的远近排序，关于在Hadoop集群中如何进行网络拓扑请看下面介绍）。如果客户端本身就是一个datanode（如客户端是一个maprece任务）并且此datanode本身就有所需文件block的话，客户端便从本地读取文件。

以上步骤完成后，DistributedFileSystem会返回一个FSDataInputStream（支持文件seek），客户端可以从FSDataInputStream中读取数据。FSDataInputStream包装了一个DFSInputSteam类，用来处理namenode和datanode的I/O操作。

客户端然后执行read()方法（第三步），DFSInputStream（已经存储了欲读取文件的开始几个block的位置信息）连接到第一个datanode（也即最近的datanode）来获取数据。通过重复调用read()方法（第四、第五步），文件内的数据就被流式的送到了客户端。当读到该block的末尾时，DFSInputStream就会关闭指向该block的流，转而找到下一个block的位置信息然后重复调用read()方法继续对该block的流式读取。这些过程对于用户来说都是透明的，在用户看来这就是不间断的流式读取整个文件。

当真个文件读取完毕时，客户端调用FSDataInputSteam中的close（）方法关闭文件输入流（第六步）。

如果在读某个block是DFSInputStream检测到错误，DFSInputSteam就会连接下一个datanode以获取此block的其他备份，同时他会记录下以前检测到的坏掉的datanode以免以后再无用的重复读取该datanode。DFSInputSteam也会检查从datanode读取来的数据的校验和，如果发现有数据损坏，它会把坏掉的block报告给namenode同时重新读取其他datanode上的其他block备份。

这种设计模式的一个好处是，文件读取是遍布这个集群的datanode的，namenode只是提供文件block的位置信息，这些信息所需的带宽是很少的，这样便有效的避免了单点瓶颈问题从而可以更大的扩充集群的规模。

Hadoop中的网络拓扑

在Hadoop集群中如何衡量两个节点的远近呢？要知道，在高速处理数据时，数据处理速率的唯一限制因素就是数据在不同节点间的传输速度：这是由带宽的可怕匮乏引起的。所以我们把带宽作为衡量两个节点距离大小的标准。

但是计算两个节点之间的带宽是比较复杂的，而且它需要在一个静态的集群下才能衡量，但Hadoop集群一般是随着数据处理的规模动态变化的（且两两节点直接相连的连接数是节点数的平方）。于是Hadoop使用了一个简单的方法来衡量距离，它把集群内的网络表示成一个树结构，两个节点之间的距离就是他们离共同祖先节点的距离之和。树一般按数据中心(datacenter)，机架(rack)，计算节点(datanode)的结构组织。计算节点上的本地运算速度最快，跨数据中心的计算速度最慢（现在跨数据中心的Hadoop集群用的还很少，一般都是在一个数据中心内做运算的）。

假如有个计算节点n1处在数据中心c1的机架r1上，它可以表示为/c1/r1/n1，下面是不同情况下两个节点的距离：

• distance(/d1/r1/n1, /d1/r1/n1) = 0 (processes on the same node)

• distance(/d1/r1/n1, /d1/r1/n2) = 2 (different nodes on the same rack)

• distance(/d1/r1/n1, /d1/r2/n3) = 4 (nodes on different racks in the same data center)

• distance(/d1/r1/n1, /d2/r3/n4) = 6 (nodes in different data centers)

如下图所示：

Hadoop

写文件

现在我们来看一下Hadoop中的写文件机制解析，通过写文件机制我们可以更好的了解一下Hadoop中的一致性模型。

Hadoop

上图为我们展示了一个创建一个新文件并向其中写数据的例子。

首先客户端通过DistributedFileSystem上的create()方法指明一个欲创建的文件的文件名（第一步），DistributedFileSystem再通过RPC调用向NameNode申请创建一个新文件（第二步，这时该文件还没有分配相应的block）。namenode检查是否有同名文件存在以及用户是否有相应的创建权限，如果检查通过，namenode会为该文件创建一个新的记录，否则的话文件创建失败，客户端得到一个IOException异常。DistributedFileSystem返回一个FSDataOutputStream以供客户端写入数据，与FSDataInputStream类似，FSDataOutputStream封装了一个DFSOutputStream用于处理namenode与datanode之间的通信。

当客户端开始写数据时（第三步），DFSOutputStream把写入的数据分成包（packet）, 放入一个中间队列——数据队列（data queue）中去。DataStreamer从数据队列中取数据，同时向namenode申请一个新的block来存放它已经取得的数据。namenode选择一系列合适的datanode（个数由文件的replica数决定）构成一个管道线（pipeline），这里我们假设replica为3，所以管道线中就有三个datanode。DataSteamer把数据流式的写入到管道线中的第一个datanode中（第四步），第一个datanode再把接收到的数据转到第二个datanode中（第四步），以此类推。

DFSOutputStream同时也维护着另一个中间队列——确认队列（ack queue），确认队列中的包只有在得到管道线中所有的datanode的确认以后才会被移出确认队列（第五步）。

如果某个datanode在写数据的时候当掉了，下面这些对用户透明的步骤会被执行：

1）管道线关闭，所有确认队列上的数据会被挪到数据队列的首部重新发送，这样可以确保管道线中当掉的datanode下流的datanode不会因为当掉的datanode而丢失数据包。

2）在还在正常运行的datanode上的当前block上做一个标志，这样当当掉的datanode重新启动以后namenode就会知道该datanode上哪个block是刚才当机时残留下的局部损坏block，从而可以把它删掉。

3）已经当掉的datanode从管道线中被移除，未写完的block的其他数据继续被写入到其他两个还在正常运行的datanode中去，namenode知道这个block还处在under-replicated状态（也即备份数不足的状态）下，然后他会安排一个新的replica从而达到要求的备份数，后续的block写入方法同前面正常时候一样。

有可能管道线中的多个datanode当掉（虽然不太经常发生），但只要dfs.replication.min（默认为1）个replica被创建，我们就认为该创建成功了。剩余的replica会在以后异步创建以达到指定的replica数。

当客户端完成写数据后，它会调用close()方法（第六步）。这个操作会冲洗（flush）所有剩下的package到pipeline中，等待这些package确认成功，然后通知namenode写入文件成功（第七步）。这时候namenode就知道该文件由哪些block组成（因为DataStreamer向namenode请求分配新block，namenode当然会知道它分配过哪些blcok给给定文件），它会等待最少的replica数被创建，然后成功返回。

replica是如何分布的

Hadoop在创建新文件时是如何选择block的位置的呢，综合来说，要考虑以下因素：带宽（包括写带宽和读带宽）和数据安全性。如果我们把三个备份全部放在一个datanode上，虽然可以避免了写带宽的消耗，但几乎没有提供数据冗余带来的安全性，因为如果这个datanode当机，那么这个文件的所有数据就全部丢失了。另一个极端情况是，如果把三个冗余备份全部放在不同的机架，甚至数据中心里面，虽然这样数据会安全，但写数据会消耗很多的带宽。Hadoop 0.17.0给我们提供了一个默认replica分配策略（Hadoop 1.X以后允许replica策略是可插拔的，也就是你可以自己制定自己需要的replica分配策略）。replica的默认分配策略是把第一个备份放在与客户端相同的datanode上（如果客户端在集群外运行，就随机选取一个datanode来存放第一个replica），第二个replica放在与第一个replica不同机架的一个随机datanode上，第三个replica放在与第二个replica相同机架的随机datanode上。如果replica数大于三，则随后的replica在集群中随机存放，Hadoop会尽量避免过多的replica存放在同一个机架上。选取replica的放置位置后，管道线的网络拓扑结构如下所示：

Hadoop

总体来说，上述默认的replica分配策略给了我们很好的可用性（blocks放置在两个rack上，较为安全），写带宽优化（写数据只需要跨越一个rack），读带宽优化（你可以从两个机架中选择较近的一个读取）。

一致性模型

HDFS某些地方为了性能可能会不符合POSIX（是的，你没有看错，POSIX不仅仅只适用于linux/unix， Hadoop 使用了POSIX的设计来实现对文件系统文件流的读取 ），所以它看起来可能与你所期望的不同，要注意。

创建了一个文件以后，它是可以在命名空间（namespace）中可以看到的：

Path p = new Path("p");

fs.create(p);

assertThat(fs.exists(p), is(true));

但是任何向此文件中写入的数据并不能保证是可见的，即使你flush了已经写入的数据，此文件的长度可能仍然为零：

Path p = new Path("p");

OutputStream out = fs.create(p);

out.write("content".getBytes("UTF-8"));

out.flush();

assertThat(fs.getFileStatus(p).getLen(), is(0L));

这是因为，在Hadoop中，只有满一个block数据量的数据被写入文件后，此文件中的内容才是可见的（即这些数据会被写入到硬盘中去），所以当前正在写的block中的内容总是不可见的。

Hadoop提供了一种强制使buffer中的内容冲洗到datanode的方法，那就是FSDataOutputStream的sync()方法。调用了sync()方法后，Hadoop保证所有已经被写入的数据都被冲洗到了管道线中的datanode中，并且对所有读者都可见了：

Path p = new Path("p");

FSDataOutputStream out = fs.create(p);

out.write("content".getBytes("UTF-8"));

out.flush();

out.sync();

assertThat(fs.getFileStatus(p).getLen(), is(((long) "content".length())));

这个方法就像POSIX中的fsync系统调用（它冲洗给定文件描述符中的所有缓冲数据到磁盘中）。例如，使用java API写一个本地文件，我们可以保证在调用flush()和同步化后可以看到已写入的内容：

FileOutputStream out = new FileOutputStream(localFile);

out.write("content".getBytes("UTF-8"));

out.flush(); // flush to operating system

out.getFD().sync(); // sync to disk （getFD()返回与该流所对应的文件描述符）

assertThat(localFile.length(), is(((long) "content".length())));

在HDFS中关闭一个流隐式的调用了sync()方法：

Path p = new Path("p");

OutputStream out = fs.create(p);

out.write("content".getBytes("UTF-8"));

out.close();

assertThat(fs.getFileStatus(p).getLen(), is(((long) "content".length())));

由于Hadoop中的一致性模型限制，如果我们不调用sync()方法的话，我们很可能会丢失多大一个block的数据。这是难以接受的，所以我们应该使用sync()方法来确保数据已经写入磁盘。但频繁调用sync()方法也是不好的，因为会造成很多额外开销。我们可以再写入一定量数据后调用sync()方法一次，至于这个具体的数据量大小就要根据你的应用程序而定了，在不影响你的应用程序的性能的情况下，这个数据量应越大越好。

③ 运行hadoop的wordcount时没有hadoop-***-core.jar文件

你仿照书上写的wordcount代码其实是一个maprece程序，其运行在hadoop平台上，按照正常的开发实现步骤专，应属该现在linux搭建hadoop集群或者伪分布，然后当你在Eclipse里面写了maprece程序之后，将你的项目打成jar包之后再hadoop集群里面运行，或者用Eclipse集成hadoop做测试。你这样直接在Eclipse里面写，是没办法运行的，就像你看了Java书在书上有一个helloword程序，然后你照着敲了一遍，然后直接在本机上运行，却忽略了你没装jdk一样，你的helloword是肯定没办法运行的

④ hadoop的几个问题 1.将本地文件复制到hdfs中，那么在hdfs中这个文件是存放在namenode还是分开放在datanode

试着回答：
先说明一下：
1. namenode负责管理目录和文件信息，真正的文件块是存放在datanode上。
2. 每个map和rece（即task)都是java进程，默版认是有单独的jvm的，所以不可能同一个类的对象会在不同节点上。
看你的描述是把namenode,datanode和jobtracker,tasktracker有点混了。

所以：
问题1. 分块存放在datanode上
问题2.inputformat是在datanode上，确切的说是在tasktracker中。每权个map和rece都会有自己的对象，当多个map读入一个文件时，实际上不同的map是读的文件不同的块，rece也是一样，各个任务读入的数据是不相交的。
问题3.rece输出肯定是在hdfs上，和普通文件一样在datanode上。
问题4.每个recer会有自己的outputformat对象，与前面inputformat原因一样。

⑤ 如何编写hadoop java程序

1.编译java
# mkdir class
#Javac -classpath .:lib/hadoop-common-2.2.0.jar:lib/hadoop-annotations-2.2.0.jar -d class HADemo.java
2.生成jar包
#jar -cvf hademo.jar -C class/ .
added manifest
adding: com/(in = 0) (out= 0)(stored 0%)
adding: com/wan/(in = 0) (out= 0)(stored 0%)
adding: com/wan/demo/(in = 0) (out= 0)(stored 0%)
adding: com/wan/demo/HADemo.class(in = 844) (out= 520)(deflated 38%)
3.测试运行
#hadoop jar hademo.jar com.wan.demo.HADemo /test
检测：
#hadoop fs -ls /

⑥ 想转行大数据必须有两年Java开发经验吗

大数据主要分两个方向，一个是大数据的收集整理，一个是大数据的分析。收集的话目前主回要用hadoop，是答用java编写的，会java最好。如果是大数据分析的话用Python、R、matlab都有，不需要会java。