Movle

MapReduce工作流程

2019-01-01

Hadoop

1.流程示意图：

MapReduce工作流程一
Mapreduce工作流程二

2.流程详解

上面的流程是整个mapreduce最全工作流程，但是shuffle过程只是从第7步开始到第16步结束，具体shuffle过程详解，如下：

(1)maptask收集我们的map()方法输出的kv对，放到内存缓冲区中
(2)从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
(3)多个溢出文件会被合并成大的溢出文件
(4)在溢出过程中，及合并的过程中，都要调用partitioner进行分区和针对key进行排序
(5)reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据
(6)reducetask会取到同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并（归并排序）
(7)合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

3.注意

Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。
缓冲区的大小可以通过参数调整，参数：io.sort.mb 默认100M。

HDFS的其他功能

2019-01-01

Hadoop

一.集群间数据拷贝

1.scp实现两个远程主机之间的文件复制

1
2
3

scp -r hello.txt [root@bigdata111:/user/itstar/hello.txt](mailto:root@bigdata111:/user/itstar/hello.txt)          // 推 push
scp -r [root@bigdata112:/user/itstar/hello.txt  hello.txt](mailto:root@hadoop103:/user/atguigu/hello.txt%20%20hello.txt)        // 拉 pull
scp -r root@bigdata112:/opt/module/hadoop-2.8.4/LICENSE.txt root@bigdata113:/opt/module/hadoop-2.8.4/LICENSE.txt   //是通过本地主机中转实现两个远程主机的文件复制；如果在两个远程主机之间ssh没有配置的情况下可以使用该方式。

2.采用discp命令实现两个hadoop集群之间的递归数据复制(注:不用设置其他，直接写IP)

1	bin/hadoop distcp hdfs://192.168.1.51:9000/LICENSE.txt hdfs://192.168.1.111:9000/HAHA

二.Hadoop存档

1.理论概述

每个文件均按块存储，每个块的元数据存储在namenode的内存中，因此hadoop存储小文件会非常低效。因为大量的小文件会耗尽namenode中的大部分内存。但注意，存储小文件所需要的磁盘容量和存储这些文件原始内容所需要的磁盘空间相比也不会增多。例如，一个1MB的文件以大小为128MB的块存储，使用的是1MB的磁盘空间，而不是128MB。
Hadoop存档文件或HAR文件，是一个更高效的文件存档工具，它将文件存入HDFS块，在减少namenode内存使用的同时，允许对文件进行透明的访问。具体说来，Hadoop存档文件可以用作MapReduce的输入。

三.快照管理

快照相当于对目录做一个备份。并不会立即复制所有文件，而是指向同一个文件。当写入发生时，才会产生新文件。

1.基本语法

hdfs dfsadmin -allowSnapshot 路径   （功能描述：开启指定目录的快照功能）
hdfs dfsadmin -disallowSnapshot 路径 （功能描述：禁用指定目录的快照功能，默认是禁用）
hdfs dfs -createSnapshot 路径        （功能描述：对目录创建快照）
hdfs dfs -createSnapshot 路径 名称   （功能描述：指定名称创建快照）
hdfs dfs -renameSnapshot 路径 旧名称 新名称 （功能描述：重命名快照）
hdfs lsSnapshottableDir         （功能描述：列出当前用户所有已快照目录）
hdfs snapshotDiff 路径1 路径2 （功能描述：比较两个快照目录的不同之处）
hdfs dfs -deleteSnapshot <path> <snapshotName>  （功能描述：删除快照）

四. 回收站

1.默认回收站

默认值fs.trash.interval=0，0表示禁用回收站，可以设置删除文件的存活时间。
默认值fs.trash.checkpoint.interval=0，检查回收站的间隔时间。
要求fs.trash.checkpoint.interval<=fs.trash.interval。

2.启用回收站

修改core-site.xml，配置垃圾回收时间为1分钟。

<property>
    <name>fs.trash.interval</name>
    <value>1</value>
</property>

3.查看回收站

回收站在集群中的；路径：/user/itstar/.Trash/….

4.修改访问垃圾回收站用户名称

进入垃圾回收站用户名称，默认是dr.who，修改为itstar用户
[core-site.xml]

<property>
  <name>hadoop.http.staticuser.user</name>
  <value>itstar</value>
</property>

5.通过程序删除的文件不会经过回收站，需要调用moveToTrash()才进入回收站

Trash trash = New Trash(conf);
trash.moveToTrash(path);

6.恢复回收站数据

1	hadoop fs -mv /user/itstar/.Trash/Current/user/itstar/input /user/itstar/input

7.清空回收站

1	hdfs dfs -expunge

MapReduce的基本概念

2019-01-01

Hadoop

1. MapReduce定义

Mapreduce是一个分布式运算程序的编程框架，是用户开发“基于hadoop的数据分析应用”的核心框架。
Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上。

2. MapReduce优缺点

(1) 优点:

(a)MapReduce 易于编程。

它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。

(b)良好的扩展性。

当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

(c )高容错性。

MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由 Hadoop内部完成的。

(d)适合PB级以上海量数据的离线处理。

这里加红字体离线处理，说明它适合离线处理而不适合在线处理。比如像毫秒级别的返回一个结果，MapReduce很难做到。

(2) 缺点:

MapReduce不擅长做实时计算、流式计算、DAG（有向图）计算。

(a)实时计算。

MapReduce无法像Mysql一样，在毫秒或者秒级内返回结果。

(b)流式计算。

流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。

(c )DAG（有向图）计算。多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

3.MapReduce核心思想

MapReduce核心思想

(1)分布式的运算程序往往需要分成至少2个阶段。

(2)第一个阶段的maptask并发实例，完全并行运行，互不相干。

(3)第二个阶段的reduce task并发实例互不相干，但是他们的数据依赖于上一个阶段的所有maptask并发实例的输出。

(4)MapReduce编程模型只能包含一个map阶段和一个reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个mapreduce程序，串行运行

4 MapReduce进程

一个完整的mapreduce程序在分布式运行时有三类实例进程：
(1)MrAppMaster：负责整个程序的过程调度及状态协调。
(2)MapTask：负责map阶段的整个数据处理流程。
(3)ReduceTask：负责reduce阶段的整个数据处理流程。

5 MapReduce编程规范

用户编写的程序分成三个部分：Mapper，Reducer，Driver(提交运行mr程序的客户端)

(1)Mapper阶段

(a)用户自定义的Mapper要继承自己的父类
(b)Mapper的输入数据是KV对的形式（KV的类型可自定义）
(c)Mapper中的业务逻辑写在map()方法中
(d)Mapper的输出数据是KV对的形式（KV的类型可自定义）
(e)map()方法（maptask进程）对每一个<K,V>调用一次

(2)Reducer阶段

(a)用户自定义的Reducer要继承自己的父类
(b)Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
(c )Reducer的业务逻辑写在reduce()方法中
(d)Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法

(3)Driver阶段

整个程序需要一个Drvier来进行提交，提交的是一个描述了各种必要信息的job对象

HDFS之DateNode工作机制

2019-01-01

Hadoop

一.NameNode & DataNode工作机制

Datanode工作机制

1.一个数据块在datanode上以文件形式存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。

2.DataNode启动后向namenode注册，通过后，周期性（1小时）的向namenode上报所有的块信息。

3.心跳是每3秒一次，心跳返回结果带有namenode给该datanode的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10分钟没有收到某个datanode的心跳，则认为该节点不可用。

4.集群运行中可以安全加入和退出一些机器

二.数据完整性

1.当DataNode读取block的时候，它会计算checksum校验和

2.如果计算后的checksum，与block创建时值不一样，说明block已经损坏。

3.client读取其他DataNode上的block.

4.datanode在其文件创建后周期验证checksum校验和

三. 掉线时限参数设置

datanode进程死亡或者网络故障造成datanode无法与namenode通信，namenode不会立即把该节点判定为死亡，要经过一段时间，这段时间暂称作超时时长。HDFS默认的超时时长为10分钟+30秒。如果定义超时时间为timeout，则超时时长的计算公式为：

1	timeout = 2 * dfs.namenode.heartbeat.recheck-interval + 10 * dfs.heartbeat.interval。

而默认的dfs.namenode.heartbeat.recheck-interval 大小为5分钟，dfs.heartbeat.interval默认为3秒。
需要注意的是hdfs-site.xml 配置文件中的heartbeat.recheck.interval的单位为毫秒，dfs.heartbeat.interval的单位为秒。

<property>
    <name>dfs.namenode.heartbeat.recheck-interval</name>
    <value>300000</value>
</property>
<property>
    <name> dfs.heartbeat.interval </name>
    <value>3</value>
</property>

四.DataNode的目录结构

和namenode不同的是，datanode的存储目录是初始阶段自动创建的，不需要额外格式化。

1.在/opt/module/hadoop-2.8.4/data/dfs/data/current这个目录下查看版本号

cat VERSION 

storageID=DS-1b998a1d-71a3-43d5-82dc-c0ff3294921b
clusterID=CID-1f2bf8d1-5ad2-4202-af1c-6713ab381175
cTime=0
datanodeUuid=970b2daf-63b8-4e17-a514-d81741392165
storageType=DATA_NODE
layoutVersion=-56

2.具体解释

storageID：存储id号
clusterID集群id，全局唯一
cTime属性标记了datanode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级之后，该值会更新到新的时间戳。
datanodeUuid：datanode的唯一识别码
storageType：存储类型
layoutVersion是一个负整数。通常只有HDFS增加新特性时才会更新这个版本号。

3.在/opt/module/hadoop-2.8.4/data/dfs/data/current/BP-97847618-192.168.10.102-1493726072779/current这个目录下查看该数据块的版本号

cat VERSION 

#Mon May 08 16:30:19 CST 2017
namespaceID=1933630176
cTime=0
blockpoolID=BP-97847618-192.168.10.102-1493726072779
layoutVersion=-56

4.具体解释

namespaceID：是datanode首次访问namenode的时候从namenode处获取的storageID对每个datanode来说是唯一的（但对于单个datanode中所有存储目录来说则是相同的），namenode可用这个属性来区分不同datanode。

cTime属性标记了datanode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级之后，该值会更新到新的时间戳。

blockpoolID：一个block pool id标识一个block pool，并且是跨集群的全局唯一。当一个新的Namespace被创建的时候(format过程的一部分)会创建并持久化一个唯一ID。在创建过程构建全局唯一的BlockPoolID比人为的配置更可靠一些。NN将BlockPoolID持久化到磁盘中，在后续的启动过程中，会再次load并使用。

layoutVersion是一个负整数。通常只有HDFS增加新特性时才会更新这个版本号。

五.Datanode多目录配置

1.datanode也可以配置成多个目录，每个目录存储的数据不一样。即：数据不是副本。

2.具体配置如下：

hdfs-site.xml

<property>
<name>dfs.datanode.data.dir</name>  
<value>file:///${hadoop.tmp.dir}/dfs/data1,file:///${hadoop.tmp.dir}/dfs/data2</value>
</property>

HDFS之NameNode工作机制

2019-01-01

Hadoop

一.NameNode&Secondary NameNode工作机制

1.第一阶段：namenode启动

(a)第一次启动namenode格式化后，创建fsimage和edits文件。如果不是第一次启动，直接加载编辑日志(edits)和镜像文件(fsimage)到内存

(b)客户端对元数据进行增删改的请求

(c)namenode记录操作日志，更新滚动日志

(d)namenode在内存中对数据进行增删改查

2.第二阶段：Secondary NameNode工作

(a)Secondary NameNode询问namenode是否需要checkpoint。直接带回namenode是否检查结果。

(b)Secondary NameNode请求执行checkpoint。

(c)namenode滚动正在写的edits日志

(d)将滚动前的编辑日志和镜像文件拷贝到Secondary NameNode

(e)Secondary NameNode加载编辑日志和镜像文件到内存，并合并。

(f)生成新的镜像文件fsimage.chkpoint

(g)拷贝fsimage.chkpoint到namenode

(h)namenode将fsimage.chkpoint重新命名成fsimage

3.chkpoint检查时间参数设置

(1)通常情况下，SecondaryNameNode每隔一小时执行一次。

[hdfs-default.xml]

<property>
  <name>dfs.namenode.checkpoint.period</name>
  <value>3600</value>
</property>

(2)一分钟检查一次操作次数，当操作次数达到1百万时，SecondaryNameNode执行一次。

<property>
  <name>dfs.namenode.checkpoint.txns</name>
  <value>1000000</value>
  <description>操作动作次数</description>
</property>

<property>
  <name>dfs.namenode.checkpoint.check.period</name>
  <value>60</value>
  <description> 1分钟检查一次操作次数</description>
</property>

二.镜像文件和编辑日志文件

1.概念

namenode被格式化之后，将在/opt/module/hadoop-2.8.4/data/dfs/name/current目录中产生如下文件,注只能在NameNode所在的节点才能找到此文件
可以执行find . -name edits* 来查找文件

edits_0000000000000000000
fsimage_0000000000000000000.md5
seen_txid
VERSION

(1)Fsimage文件：HDFS文件系统元数据的一个永久性的检查点，其中包含HDFS文件系统的所有目录和文件idnode的序列化信息。

(2)Edits文件：存放HDFS文件系统的所有更新操作的路径，文件系统客户端执行的所有写操作首先会被记录到edits文件中。

(3)seen_txid文件保存的是一个数字，就是最后一个edits_的数字

(4)每次Namenode启动的时候都会将fsimage文件读入内存，并从00001开始到seen_txid中记录的数字依次执行每个edits里面的更新操作，保证内存中的元数据信息是最新的、同步的，可以看成Namenode启动的时候就将fsimage和edits文件进行了合并。

2.oiv查看fsimage文件

(1)查看oiv和oev命令

hdfs

hdfs

(2)基本语法

1	hdfs oiv -p 文件类型 -i镜像文件 -o 转换后文件输出路径

(3)案例实操

pwd
/opt/module/hadoop-2.8.4/data/dfs/name/current

hdfs oiv -p XML -i fsimage_0000000000000000316 -o /opt/fsimage.xml

cat /opt/module/hadoop-2.8.4/fsimage.xml

将显示的xml文件内容拷贝到IDEA中创建的xml文件中，并格式化。

3.oev查看edits文件

(1)基本语法

hdfs oev -p 文件类型 -i编辑日志 -o 转换后文件输出路径

-p	–processor <arg>   指定转换类型: binary (二进制格式), xml (默认，XML格式),stats
-i	–inputFile <arg>     输入edits文件，如果是xml后缀，表示XML格式，其他表示二进制
-o 	–outputFile <arg> 输出文件，如果存在，则会覆盖

三.滚动编辑日志

正常情况HDFS文件系统有更新操作时，就会滚动编辑日志。也可以用命令强制滚动编辑日志。
(1)滚动编辑日志（前提必须启动集群）

1	hdfs dfsadmin -rollEdits

举例：原文件名edits_inprogress_0000000000000000010
执行以下命令后

(2)镜像文件什么时候产生
Namenode启动时加载镜像文件和编辑日志

四.namenode版本号

1.查看namenode版本号

在/opt/module/hadoop-2.8.4/data/dfs/name/current这个目录下查看VERSION

namespaceID=1778616660
clusterID=CID-bc165781-d10a-46b2-9b6f-3beb1d988fe0
cTime=1552918200296
storageType=NAME_NODE
blockpoolID=BP-274621862-192.168.1.111-1552918200296
layoutVersion=-63

2.namenode版本号具体解释

(1)namespaceID在HDFS上，会有多个Namenode，所以不同Namenode的namespaceID是不同的，分别管理一组blockpoolID。

(2)clusterID集群id，全局唯一

(3)cTime属性标记了namenode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级之后，该值会更新到新的时间戳。

(4)storageType属性说明该存储目录包含的是namenode的数据结构。

(5)blockpoolID：一个block pool id标识一个block pool，并且是跨集群的全局唯一。当一个新的Namespace被创建的时候(format过程的一部分)会创建并持久化一个唯一ID。在创建过程构建全局唯一的BlockPoolID比人为的配置更可靠一些。NN将BlockPoolID持久化到磁盘中，在后续的启动过程中，会再次load并使用。

(6)layoutVersion是一个负整数。通常只有HDFS增加新特性时才会更新这个版本号。

(7)storageID (存储ID)：是DataNode的ID,不唯一

五.SecondaryNameNode目录结构

Secondary NameNode用来监控HDFS状态的辅助后台程序，每隔一段时间获取HDFS元数据的快照。
在/opt/module/hadoop-2.8.4/data/dfs/namesecondary/current这个目录中查看SecondaryNameNode目录结构

edits_0000000000000000001-0000000000000000002
fsimage_0000000000000000002
fsimage_0000000000000000002.md5
VERSION

SecondaryNameNode的namesecondary/current目录和主namenode的current目录的布局相同。
好处：在主namenode发生故障时（假设没有及时备份数据），可以从SecondaryNameNode恢复数据。

1
2

方法一：将SecondaryNameNode中数据拷贝到namenode存储数据的目录；
方法二：使用-importCheckpoint选项启动namenode守护进程，从而将SecondaryNameNode中数据拷贝到namenode目录中。

六.集群安全模式操作

1.概述

    Namenode启动时，首先将映像文件（fsimage）载入内存，并执行编辑日志（edits）中的各项操作。一旦在内存中成功建立文件系统元数据的映像，则创建一个新的fsimage文件和一个空的编辑日志。此时，namenode开始监听datanode请求。但是此刻，namenode运行在安全模式，即namenode的文件系统对于客户端来说是只读的。
    系统中的数据块的位置并不是由namenode维护的，而是以块列表的形式存储在datanode中。在系统的正常操作期间，namenode会在内存中保留所有块位置的映射信息。在安全模式下，各个datanode会向namenode发送最新的块列表信息，namenode了解到足够多的块位置信息之后，即可高效运行文件系统。
    如果满足“最小副本条件”，namenode会在30秒钟之后就退出安全模式。所谓的最小副本条件指的是在整个文件系统中99.9%的块满足最小副本级别（默认值：dfs.replication.min=1）。在启动一个刚刚格式化的HDFS集群时，因为系统中还没有任何块，所以namenode不会进入安全模式。

2.基本语法

集群处于安全模式，不能执行重要操作（写操作）。集群启动完成后，自动退出安全模式。

bin/hdfs dfsadmin -safemode get		（功能描述：查看安全模式状态）
bin/hdfs dfsadmin -safemode enter  	（功能描述：进入安全模式状态）
bin/hdfs dfsadmin -safemode leave	（功能描述：离开安全模式状态）
bin/hdfs dfsadmin -safemode wait	（功能描述：等待安全模式状态）

七.Namenode多目录配置

1.namenode的本地目录可以配置成多个，且每个目录存放内容相同，增加了可靠性。

2.具体配置如下：

hdfs-site.xml

<property>
<name>dfs.namenode.name.dir</name>
<value>file:///${hadoop.tmp.dir}/dfs/name1,file:///${hadoop.tmp.dir}/dfs/name2</value>
</property>

(1)步骤：

a.停止集群 删除data 和 logs  rm -rf data/* logs/*

b.hdfs namenode -format

c.start-dfs.sh

d.去展示

(2)具体解释：

格式化做了哪些事情？

在NameNode节点上，有两个最重要的路径，分别被用来存储元数据信息和操作日志，而这两个路径来自于配置文件，它们对应的属性分别是dfs.name.dir和dfs.name.edits.dir，同时，它们默认的路径均是/tmp/hadoop/dfs/name。格式化时，NameNode会清空两个目录下的所有文件，之后，格式化会在目录dfs.name.dir下创建文件

hadoop.tmp.dir 这个配置，会让dfs.name.dir和dfs.name.edits.dir会让两个目录的文件生成在一个目录里

(3)思考2：非NN上如果生成了name1和name2，那么他和NN上生成得有没有差别？

答：有区别、NN节点上会产生新得edits_XXX，非NN不会fsimage会更新，而非NN不会，只会产生一个仅初始化得到得fsimage，不会生成edits,更不会发生日志滚动。

HDFS的读写数据流程

2019-01-01

Hadoop

一.HDFS写数据流程

1.剖析文件写入

HDFS的写数据流程

(1)客户端向namenode请求上传文件，namenode检查目标文件是否已存在，父目录是否存在。

(2)namenode返回是否可以上传。

(3)客户端请求第一个 block上传到哪几个datanode服务器上。

(4)namenode返回3个datanode节点，分别为dn1、dn2、dn3。

(5)客户端请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成

(6)dn1、dn2、dn3逐级应答客户端

(7)客户端开始往dn1上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位，dn1收到一个packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答

(8)当一个block传输完成之后，客户端再次请求namenode上传第二个block的服务器。（重复执行3-7步）

文件的写入

1.	客户端通过调用DistributedFileSystem的create方法创建新文件。

2.	DistributedFileSystem通过RPC调用namenode去创建一个没有blocks关联的新文件，创建前， namenode会做各种校验，比如文件是否存在，客户端有无权限去创建等。如果校验通过， namenode就会记录下新文件，否则就会抛出IO异常。

3.	前两步结束后，会返回FSDataOutputStream的对象，与读文件的时候相似， FSDataOutputStream被封装成DFSOutputStream。DFSOutputStream可以协调namenode和 datanode。客户端开始写数据到DFSOutputStream，DFSOutputStream会把数据切成一个个小的packet，然后排成队 列data quene（数据队列）。

4.	DataStreamer会去处理接受data quene，它先询问namenode这个新的block最适合存储的在哪几个datanode里（比如重复数是3，那么就找到3个最适合的 datanode），把他们排成一个pipeline。DataStreamer把packet按队列输出到管道的第一个datanode中，第一个 datanode又把packet输出到第二个datanode中，以此类推。

5.	DFSOutputStream还有一个对列叫ack quene，也是由packet组成，等待datanode的收到响应，当pipeline中的所有datanode都表示已经收到的时候，这时ack quene才会把对应的packet包移除掉。 
如果在写的过程中某个datanode发生错误，会采取以下几步： 
  (1)pipeline被关闭掉； 
  (2)	为了防止防止丢包ack quene里的packet会同步到data quene里； 
  (3)把产生错误的datanode上当前在写但未完成的block删掉； 
  (4)block剩下的部分被写到剩下的两个正常的datanode中； 
  (5)namenode找到另外的datanode去创建这个块的复制。当然，这些操作对客户端来说是无感知的。

6.	客户端完成写数据后调用close方法关闭写入流。

7.	DataStreamer把剩余得包都刷到pipeline里，然后等待ack信息，收到最后一个ack后，通知datanode把文件标视为已完成。

    注意：客户端执行write操作后，写完的block才是可见的(注:和下面的一致性所对应)，正在写的block对客户端是不可见的，只有 调用sync方法，客户端才确保该文件的写操作已经全部完成，当客户端调用close方法时，会默认调用sync方法。是否需要手动调用取决你根据程序需 要在数据健壮性和吞吐率之间的权衡

二.HDFS读数据流程

HDFS读数据流程

(1)客户端向namenode请求下载文件，namenode通过查询元数据，找到文件块所在的datanode地址。

(2)挑选一台datanode（就近原则，然后随机）服务器，请求读取数据。

(3)datanode开始传输数据给客户端（从磁盘里面读取数据放入流，以packet为单位来做校验）。

(4)客户端以packet为单位接收，先在本地缓存，然后写入目标文件

1.	首先调用FileSystem对象的open方法，其实是一个DistributedFileSystem的实例。
2.	DistributedFileSystem通过rpc获得文件的第一批block的locations，同一个block按照重复数会返回多个locations，这些locations按照hadoop拓扑结构排序，距离客户端近的排在前面。
3.	前两步会返回一个FSDataInputStream对象，该对象会被封装DFSInputStream对象，DFSInputStream可 以方便的管理datanode和namenode数据流。客户端调用read方法，DFSInputStream最会找出离客户端最近的datanode 并连接。
4.	数据从datanode源源不断的流向客户端。
5.	如果第一块的数据读完了，就会关闭指向第一块的datanode连接，接着读取下一块。这些操作对客户端来说是透明的，客户端的角度看来只是读一个持续不断的流。
6.	如果第一批block都读完了， DFSInputStream就会去namenode拿下一批block的locations，然后继续读，如果所有的块都读完，这时就会关闭掉所有的流。 
7.	如果在读数据的时候， DFSInputStream和datanode的通讯发生异常，就会尝试正在读的block的排序第二近的datanode,并且会记录哪个 datanode发生错误，剩余的blocks读的时候就会直接跳过该datanode。 DFSInputStream也会检查block数据校验和，如果发现一个坏的block,就会先报告到namenode节点，然后 DFSInputStream在其他的datanode上读该block的镜像。
8.	该设计就是客户端直接连接datanode来检索数据并且namenode来负责为每一个block提供最优的datanode， namenode仅仅处理block location的请求，这些信息都加载在namenode的内存中，hdfs通过datanode集群可以承受大量客户端的并发访问。

三.一致性模型

(1)debug调试如下代码

   @Test
public void writeFile() throws Exception{
	// 1 创建配置信息对象
	Configuration configuration = new Configuration();
	fs = FileSystem.get(configuration);
	
	// 2 创建文件输出流
	Path path = new Path("F:\\date\\H.txt");
	FSDataOutputStream fos = fs.create(path);
	
	// 3 写数据
	fos.write("hello Andy".getBytes());
       // 4 一致性刷新
	fos.hflush();
	
	fos.close();
}

(2)总结
写入数据时，如果希望数据被其他client立即可见，调用如下方法

1	FSDataOutputStream. hflush (); //清理客户端缓冲区数据，被其他client立即可见

Hadoop之通过API操作HDFS

2019-01-01

Hadoop

HDFS客户端操作

一.IDEA环境准备

1.修改$MAVEN_HOME/conf/settings.xml

  <!--本地仓库所在位置-->
<localRepository>F:\m2\repository</localRepository>

<!--使用阿里云镜像去下载Jar包，速度更快-->
  <mirrors>
    <mirror>
      <id>alimaven</id>
      <name>aliyun maven</name>
      <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
      <mirrorOf>central</mirrorOf>        
    </mirror>
  </mirrors>

<!--本地配置JDK8版本-->
	<profiles>
		<profile>  
		  <id>jdk-1.8</id>  
		   <activation>  
		     <activeByDefault>true</activeByDefault>  
		     <jdk>1.8</jdk>  
		   </activation>  
			<properties>  
				<maven.compiler.source>1.8</maven.compiler.source>  
				<maven.compiler.target>1.8</maven.compiler.target>  
				<maven.compiler.compilerVersion>1.8</maven.compiler.compilerVersion>
			</properties>
		</profile>
	</profiles>

2.Maven准备,在在项目文件pom.xml文件中添加

<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.8.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.8.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.8.4</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.10</version>
        </dependency>

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.7</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/junit/junit -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
   </dependencies>

二. 通过API操作HDFS

1.HDFS获取文件系统

(1)详细代码：

 /**
 * 打印本地hadoop地址值
 * IO的方式写代码
 */
@Test
public void intiHDFS() throws IOException {
    //F2 可以快速的定位错误
    // alt + enter自动找错误
    //1.创建配信信息对象 ctrl + alt + v  后推前  ctrl + shitl + enter 补全
    Configuration conf = new Configuration();

    //2.获取文件系统
    FileSystem fs = FileSystem.get(conf);

    //3.打印文件系统
    System.out.println(fs.toString());
}

2.HDFS文件上传

(1)详细代码

    /**
     * 上传代码
     * 注意：如果上传的内容大于128MB,则是2块
     */
    @Test
    public void putFileToHDFS() throws Exception {
        //注：import org.apache.hadoop.conf.Configuration;
        //ctrl + alt + v 推动出对象
        //1.创建配置信息对象
        Configuration conf = new Configuration();

        //2.设置部分参数
        conf.set("dfs.replication","2");

        //3.找到HDFS的地址
        FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

        //4.上传本地Windows文件的路径
        Path src = new Path("D:\\hadoop-2.7.2.rar");

        //5.要上传到HDFS的路径
        Path dst = new Path("hdfs://bigdata111:9000/Andy");

        //6.以拷贝的方式上传，从src -> dst
        fs.copyFromLocalFile(src,dst);

        //7.关闭
        fs.close();

        System.out.println("上传成功");
}

3.HDFS文件下载

(1)详细代码：

 /**
 * hadoop fs -get /HDFS文件系统
 * @throws Exception
 */
@Test
public void getFileFromHDFS() throws Exception {
    //1.创建配置信息对象  Configuration:配置
    Configuration conf = new Configuration();

    //2.找到文件系统
    //final URI uri     ：HDFS地址
    //final Configuration conf：配置信息
    // String user ：Linux用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.下载文件
    //boolean delSrc:是否将原文件删除
    //Path src ：要下载的路径
    //Path dst ：要下载到哪
    //boolean useRawLocalFileSystem ：是否校验文件
    fs.copyToLocalFile(false,new Path("hdfs://bigdata111:9000/README.txt"),
            new Path("F:\\date\\README.txt"),true);

    //4.关闭fs
    //alt + enter 找错误
    //ctrl + alt + o  可以快速的去除没有用的导包
    fs.close();
    System.out.println("下载成功");
}

4.HDFS目录创建

(1)详细代码

 /**
 * hadoop fs -mkdir /xinshou
 */
@Test
public void mkmdirHDFS() throws Exception {
    //1.创新配置信息对象
    Configuration configuration = new Configuration();

    //2.链接文件系统
    //final URI uri  地址
    //final Configuration conf  配置
    //String user   Linux用户
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), configuration, "root");

    //3.创建目录
    fs.mkdirs(new Path("hdfs://bigdata111:9000/Good/Goog/Study"));

    //4.关闭
    fs.close();
    System.out.println("创建文件夹成功");
}

5.HDFS文件夹删除

(1)详细代码

/**
 * hadoop fs -rm -r /文件
 */
@Test
public void deleteHDFS() throws Exception {
    //1.创建配置对象
    Configuration conf = new Configuration();

    //2.链接文件系统
    //final URI uri, final Configuration conf, String user
    //final URI uri  地址
    //final Configuration conf  配置
    //String user   Linux用户
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.删除文件
    //Path var1   : HDFS地址
    //boolean var2 : 是否递归删除
    fs.delete(new Path("hdfs://bigdata111:9000/a"),false);

    //4.关闭
    fs.close();
    System.out.println("删除成功啦");
}

6.HDFS文件名更改

(1)详细代码：

    @Test
    public void renameAtHDFS() throws Exception{
	// 1 创建配置信息对象
	Configuration configuration = new Configuration();	
	FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"),configuration, "itstar");
		
	//2 重命名文件或文件夹
	fs.rename(new Path("hdfs://bigdata111:9000/user/itstar/hello.txt"), new Path("hdfs://bigdata111:9000/user/itstar/hellonihao.txt"));
	fs.close();
}

7.HDFS文件详情查看

查看文件名称、权限、长度、块信息

/**
 * 查看【文件】名称、权限等
 */
@Test
public void readListFiles() throws Exception {
    //1.创建配置对象
    Configuration conf = new Configuration();

    //2.链接文件系统
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.迭代器
    RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

    //4.遍历迭代器
    while (listFiles.hasNext()){
        //一个一个出
        LocatedFileStatus fileStatus = listFiles.next();

        //名字
        System.out.println("文件名：" + fileStatus.getPath().getName());
        //块大小
        System.out.println("大小：" + fileStatus.getBlockSize());
        //权限
        System.out.println("权限：" + fileStatus.getPermission());
        System.out.println(fileStatus.getLen());


        BlockLocation[] locations = fileStatus.getBlockLocations();

        for (BlockLocation bl:locations){
            System.out.println("block-offset:" + bl.getOffset());
            String[] hosts = bl.getHosts();
            for (String host:hosts){
                System.out.println(host);
            }
        }

        System.out.println("------------------华丽的分割线----------------");
    }

8.HDFS文件和文件夹判断

/**
 * 判断是否是个文件还是目录，然后打印
 * @throws Exception
 */
@Test
public void judge() throws Exception {
    //1.创建配置文件信息
    Configuration conf = new Configuration();

    //2.获取文件系统
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.遍历所有的文件
    FileStatus[] liststatus = fs.listStatus(new Path("/Andy"));
    for(FileStatus status :liststatus)
    {
        //判断是否是文件
        if (status.isFile()){
            //ctrl + d:复制一行
            //ctrl + x 是剪切一行，可以用来当作是删除一行
            System.out.println("文件:" + status.getPath().getName());
        } else {
            System.out.println("目录:" + status.getPath().getName());
        }
    }
}

三. 通过IO流操作HDFS

1.HDFS文件上传

/**
 * IO流方式上传
 *
 * @throws URISyntaxException
 * @throws FileNotFoundException
 * @throws InterruptedException
 */
@Test
public void putFileToHDFSIO() throws URISyntaxException, IOException, InterruptedException {
    //1.创建配置文件信息
    Configuration conf = new Configuration();

    //2.获取文件系统
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.创建输入流
    FileInputStream fis = new FileInputStream(new File("F:\\date\\Sogou.txt"));

    //4.输出路径
    //注意：不能/Andy  记得后边写个名 比如：/Andy/Sogou.txt
    Path writePath = new Path("hdfs://bigdata111:9000/Andy/Sogou.txt");
    FSDataOutputStream fos = fs.create(writePath);

    //5.流对接
    //InputStream in    输入
    //OutputStream out  输出
    //int buffSize      缓冲区
    //boolean close     是否关闭流
    try {
        IOUtils.copyBytes(fis,fos,4 * 1024,false);
    } catch (IOException e) {
        e.printStackTrace();
    }finally {
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        System.out.println("上传成功啦");
    }
}

2.HDFS文件下载

 /**
 * IO读取HDFS到本地
 *
 * @throws URISyntaxException
 * @throws IOException
 * @throws InterruptedException
 */
@Test
public void getFileToHDFSIO() throws URISyntaxException, IOException, InterruptedException {
    Configuration configuration = new Configuration();

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://bigdata111:9000"),configuration,"root");


    FileOutputStream fos = new FileOutputStream(new File("/Users/macbook/TestInfo/downloadFileFromHdfsIO.txt"));

    Path readPath = new Path("hdfs://bigdata111:9000/aa.txt");

    FSDataInputStream fis = fileSystem.open(readPath);

    try {
        IOUtils.copyBytes(fis,fos,4*1024,false);
    } catch (IOException e) {
        e.printStackTrace();
    }finally {
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        System.out.println("下载成功！");
    }
}

3.定位文件读取

(1)下载第一块

 /**
 * IO读取第一块的内容
 *
 * @throws Exception
 */
@Test
public void  readFlieSeek1() throws Exception {
    //1.创建配置文件信息
    Configuration conf = new Configuration();

    //2.获取文件系统
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.输入
    Path path = new Path("hdfs://bigdata111:9000/Andy/hadoop-2.7.2.rar");
    FSDataInputStream fis = fs.open(path);

    //4.输出
    FileOutputStream fos = new FileOutputStream("F:\\date\\readFileSeek\\A1");

    //5.流对接
    byte[] buf = new byte[1024];
    for (int i = 0; i < 128 * 1024; i++) {
        fis.read(buf);
        fos.write(buf);
    }

    //6.关闭流
    IOUtils.closeStream(fos);
    IOUtils.closeStream(fis);
}

(2)下载第二块

/**
 * IO读取第二块的内容
 *
 * @throws Exception
 */
@Test
public void readFlieSeek2() throws Exception {
    //1.创建配置文件信息
    Configuration conf = new Configuration();

    //2.获取文件系统
    FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");

    //3.输入
    Path path = new Path("hdfs://bigdata111:9000/Andy/hadoop-2.7.2.rar");
    FSDataInputStream fis = fs.open(path);

    //4.输出
    FileOutputStream fos = new FileOutputStream("F:\\date\\readFileSeek\\A2");

    //5.定位偏移量/offset/游标/读取进度 (目的：找到第一块的尾巴，第二块的开头)
    fis.seek(128 * 1024 * 1024);

    //6.流对接
    IOUtils.copyBytes(fis, fos, 1024);

    //7.关闭流
    IOUtils.closeStream(fos);
    IOUtils.closeStream(fis);
}

(3)合并文件
在window命令窗口中执行
type A2 >> A1 然后更改后缀为rar即可

HDFS基础概念以及HDFS命令行操作

2019-01-01

Hadoop

一.HDFS基础概念

1.概念

HDFS，它是一个文件系统，全称：Hadoop Distributed File System，用于存储文件通过目录树来定位文件；其次，它是分布式的，由很多服务器联合起来实现其功能，集群中的服务器有各自的角色。

2.组成

(1)HDFS集群包括，NameNode和DataNode以及Secondary Namenode。

(2)NameNode负责管理整个文件系统的元数据，以及每一个路径（文件）所对应的数据块信息。

(3)DataNode 负责管理用户的文件数据块，每一个数据块都可以在多个datanode上存储多个副本。

(4)Secondary NameNode用来监控HDFS状态的辅助后台程序，每隔一段时间获取HDFS元数据的快照。

3 HDFS 文件块大小

    HDFS中的文件在物理上是分块存储（block），块的大小可以通过配置参数(dfs.blocksize)来规定，默认大小在hadoop2.x版本中是128M，老版本中是64M
    HDFS的块比磁盘的块大，其目的是为了最小化寻址开销。如果块设置得足够大，从磁盘传输数据的时间会明显大于定位这个块开始位置所需的时间。因而，传输一个由多个块组成的文件的时间取决于磁盘传输速率。
    如果寻址时间约为10ms，而传输速率为100MB/s，为了使寻址时间仅占传输时间的1%，我们要将块大小设置约为100MB。默认的块大小128MB。
    块的大小：10ms100100M/s = 100M
HDFS文件快的大小

二.HDFS命令行操作：

1.基本语法

1	bin/hadoop fs 具体命令

2.参数大全

bin/hadoop fs
        [-appendToFile <localsrc> ... <dst>]
        [-cat [-ignoreCrc] <src> ...]
        [-checksum <src> ...]
        [-chgrp [-R] GROUP PATH...]
        [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
        [-chown [-R] [OWNER][:[GROUP]] PATH...]
        [-copyFromLocal [-f] [-p] <localsrc> ... <dst>]
        [-copyToLocal [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-count [-q] <path> ...]
        [-cp [-f] [-p] <src> ... <dst>]
        [-createSnapshot <snapshotDir> [<snapshotName>]]
        [-deleteSnapshot <snapshotDir> <snapshotName>]
        [-df [-h] [<path> ...]]
        [-du [-s] [-h] <path> ...]
        [-expunge]
        [-get [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-getfacl [-R] <path>]
        [-getmerge [-nl] <src> <localdst>]
        [-help [cmd ...]]
        [-ls [-d] [-h] [-R] [<path> ...]]
        [-mkdir [-p] <path> ...]
        [-moveFromLocal <localsrc> ... <dst>]
        [-moveToLocal <src> <localdst>]
        [-mv <src> ... <dst>]
        [-put [-f] [-p] <localsrc> ... <dst>]
        [-renameSnapshot <snapshotDir> <oldName> <newName>]
        [-rm [-f] [-r|-R] [-skipTrash] <src> ...]
        [-rmdir [--ignore-fail-on-non-empty] <dir> ...]
        [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
        [-setrep [-R] [-w] <rep> <path> ...]
        [-stat [format] <path> ...]
        [-tail [-f] <file>]
        [-test -[defsz] <path>]
        [-text [-ignoreCrc] <src> ...]
        [-touchz <path> ...]
        [-usage [cmd ...]]

3.常用命令

(1)-help：输出这个命令参数

1	bin/hdfs dfs -help rm

(2)-ls: 显示目录信息

1	hadoop fs -ls /

(3)-mkdir：在hdfs上创建目录

1	hadoop fs -mkdir -p /hdfs路径

(4)-moveFromLocal从本地剪切粘贴到hdfs

1	hadoop fs -moveFromLocal 本地路径 /hdfs路径

(5)–appendToFile ：追加一个文件到已经存在的文件末尾

1	hadoop fs -appendToFile 本地路径 /hdfs路径

(6)-cat ：显示文件内容

1	hadoop fs -cat /hdfs路径

(7)-tail -f：监控文件

1	hadoop fs -tail -f /hdfs路径

(8)-chmod、-chown：linux文件系统中的用法一样，修改文件所属权限

1 2	hadoop fs -chmod 777 /hdfs路径 hadoop fs -chown someuser:somegrp /hdfs路径

(9)-cp ：从hdfs的一个路径拷贝到hdfs的另一个路径

1	hadoop fs -cp /hdfs路径1 / hdfs路径2

(10)-mv：在hdfs目录中移动/重命名文件

1	hadoop fs -mv /hdfs路径 / hdfs路径

(11)-get：等同于copyToLocal，就是从hdfs下载文件到本地

1	hadoop fs -get / hdfs路径 ./本地路径

(12)-getmerge ：合并下载多个文到linux本地，比如hdfs的目录 /aaa/下有多个文件:log.1, log.2,log.3,…（注：是合成到Linux本地）

1 2	hadoop fs -getmerge /aaa/log.* ./log.sum hadoop fs -getmerge /hdfs1路径 /hdfs2路径 / //合成到不同的目录：

(13)-put：等同于copyFromLocal

1	hadoop fs -put /本地路径 /hdfs路径

(14)-rm：删除文件或文件夹

1	hadoop fs -rm -r /hdfs路径

(15)-df ：统计文件系统的可用空间信息

1	hadoop fs -df -h /hdfs路径

(16)-du统计文件夹的大小信息

1	hadoop fs -du -s -h /hdfs路径

(17)-count：统计一个指定目录下的文件节点数量

1 2	hadoop fs -count /aaa/ hadoop fs -count /hdfs路径

(18)-setrep：设置hdfs中文件的副本数量：3是副本数，可改

1	hadoop fs -setrep 3 / hdfs路径

这里设置的副本数只是记录在namenode的元数据中，是否真的会有这么多副本，还得看datanode的数量。因为目前只有3台设备，最多也就3个副本，只有节点数的增加到10台时，副本数才能达到10。

厨艺日志

2019-01-01

厨艺

2021/01/05

1.今天晚上做了香煎豆腐，但是豆腐切得太厚了，下次切薄一点，油多放一点

2021/01/06

1.今天做的炒饼，酱油放少了一点，盐放多了一点，有点咸

2020/01/07

1.今天做了炒土豆丝和蚝油生菜，先配调料，碗里倒一些酱油，加入一些淀粉水，再加一勺蚝油，再加一些水，然后把水煮开，放一点油，然后把生菜烫熟捞出，然后锅烧干，放油，放入蒜末煸香，然后倒入调好的酱汁，之后把弄好的酱汁倒在生菜上即可，还蛮好吃的

Hadoop环境搭建-本地模式搭建

2019-01-01

Hadoop

1.Hadoop安装准备工作：

(1)安装好linux操作系统
(2)关闭防火墙
(3)在linux上安装JDK

2.本地模式具体步骤：

(1)解压hadoop安装包：

1	tar -zxvf hadoop-2.8.4.tar.gz -C /opt/module

(2)配置环境变量：

(a).修改命令：

1	vi ~/.bash_profile //修改环境变量的文件

(b).添加的内容为：

HADOOP_HOME=/opt/module/hadoop-2.8.4
export HADOOP_HOME
PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH
export PATH

(c ).使环境变量生效

输入命令：

1	source ~/.bash_profile

(3)修改配置文件hadoop-env.sh

(a).进入hadoop-2.8.4/etc/hadoop文件夹

输入命令：

1	cd /opt/module/hadoop-2.8.4/etc/hadoop //进入hadoop-2.8.4/etc/hadoop文件夹

(b).修改配置文件hadoop-env.sh：

输入命令：

1	vi hadoop-env.sh //修改hadoop-env.sh文件

修改内容为：

1	export JAVA_HOME=/opt/module/jdk1.8.0_144 //修改JAVAHOME地址，改为自己建的jdk地址，应该在25行

jdk

(c ).验证（运行一个MR程序）

(1)在/root/temp目录下创建一个a.txt文件
(2)修改a.txt文件

touch a.txt                                 //新建a.txt文件
vi a.txt                                    //修改a.txt

i am a tiger you are also a tiger           //添加内容

(3)进入/opt/module/hadoop-2.8.4/share/hadoop/mapreduce文件夹下
(4)运行wordcount程序 hadoop jar hadoop-mapreduce-examples-2.8.4.jar wordcount ~/temp/a.txt ~/temp/output/wc0717

cd /opt/module/hadoop-2.8.4/share/hadoop/mapreduce   //进入文件夹

//hadoop jar 程序包 函数名 输入文件地址（这里是本地linux路径，因为本地模式没有hdfs） 输出文件地址
hadoop jar hadoop-mapreduce-examples-2.8.4.jar wordcount ~/temp/a.txt  ~/temp/output/wc0717     //执行MR程序

运行成功后，会在/root/temp/output/文件夹下，生产wc0717文件下，此文件夹下有两个文件part-r-000000和_SUCCESS
wordcount程序运行结果