大数据核心框架Hadoop_haoop

Hadoop两大核心框架HDFS与MapReduce的原理、发展情况、常见操作与行业应用

Hadoop简介

1、什么是Hadoop？

Hadoop是一个由Apache基金会所开发的分布式系统基础架构，开源的、可靠的，可扩展的，分布式的运算存储系统。Hadoop实现了一个分布式文件系统（Hadoop Distributed File System），简称HDFS。HDFS有容错性的特点，并且设计用来部署在低廉的（low-cost）硬件上；而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。

Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，而 MapReduce为海量的数据提供了计算。

2、Haoop的特点？

扩容能力（Scalable）：能可靠地（reliably）存储和处理千兆字节（PB）数据。
成本低（Economical）：可以通过普通机器组成的服务器群来分发以及处理数据。这些服务器群总计可达数千个节点。
高效率（Efficient）：通过分发数据，Hadoop可以在数据所在的节点上并行地（parallel）处理它们，这使得处理非常的快速。
可靠性（Reliable）：Hadoop能自动地维护数据的多份副本，并且在任务失败后能自动地重新部署（redeploy）计算任务。
低成本：与一体机、商用数据仓库以及QlikView、Yonghong Z-Suite等数据集市相比，Hadoop是开源的，项目的软件成本因此会大大降低。

3、Hadoop项目组成部分

• Hadoop Common:支撑其他模块
• Hadoop Distributed File System：分布式系统对应用提供高吞吐量的访问
• Hadoop Yarn：资源管理和任务调度的一个框架。
• Hadoop MapReduce：能够并行处理大数据集的yarn基本系统

4、Hadoop主要应用场景

Hadoop适用于海量数据、离线数据和负责数据，应用场景如下：

场景1：数据分析，如Facebook就用Hive来进行日志分析，2009年时Facebook就有非编程人员的30%的人使用HiveQL进行数据分析。淘宝搜索中的自定义筛选也使用的Hive；利用Pig还可以做高级的数据处理，包括Twitter、LinkedIn上用于发现您可能认识的人，可以实现类似Amazon.com的协同过滤的推荐效果。淘宝的商品推荐也是。在Yahoo！的40%的Hadoop作业是用pig运行的，包括垃圾邮件的识别和过滤，还有用户特征建模。

场景2：离线计算，(异构计算+分布式计算)天文计算。

场景3：海量数据存储，如京东的存储集群。

5、Hadoo部署方式

(1)单例模式： 单例模式就是只有Hadoop被配置成以非分布式模式运行的一个独立Java进程。单机模式是Hadoop的默认模式。当首次解压Hadoop的源码包时，Hadoop无法了解硬件安装环境，便保守地选择了最小配置。在这种默认模式下所有3个XML文件均为空。当配置文件为空时，Hadoop会完全运行在本地。因为不需要与其他节点交互，单机模式就不使用HDFS，也不加载任何Hadoop的守护进程。该模式主要用于开发调试MapReduce程序的应用逻辑。

(2)伪分布模式： 伪分布式是使用多个守护线程模拟分布的伪分布运行模式，然后修改Hadoop的一些配置文件和基本操作。
1、Hadoop的配置文件：

  conf/hadoop-env.sh  配置JAVA_HOME；
  core-site.xml  配置HDFS节点名称和地址；
  hdfs-site.xml  配置HDFS存储目录，复制数量；
  mapred-site.xml   配置mapreduce的jobtracker地址；
  yarn-site.xml 配置yarn的地址。

1
2
3
4
5
6

2、配置ssh，生成密匙，使到ssh可以免密码连接。 (RSA算法，基于因数不对称加密：公钥加密私钥才能解密，私钥加密公钥才能解密)。
3、启动Hadoop： bin/start-all.sh。

(3)集群模式： 集群分布式就是真正多台机器来搭建分布式集群，计算机集群是一种计算机系统， 它通过一组松散集成的计算机软件和/或硬件连接起来高度紧密地协作完成计算工作。集群系统中的单个计算机通常称为节点，通常通过局域网连接。
集群技术的特点：
通过多台计算机完成同一个工作，达到更高的效率。
两机或多机内容、工作过程等完全一样。如果一台死机，另一台可以起作用。
其操作和步骤如下：

• 配置etc/hosts文件,使主机名解析为IP 或者使用DNS服务解析主机名
• 建立hadoop运行用户: useradd grid>passwd grid
• 配置ssh密码连入: 每个节点用grid登录，进入主工作目录，ssh -keygen -t rsa生产公钥，然后将每个节点的公钥复制到同一个文件中，再将这个包含所有节点公钥的文件复制到每个节点authorized_keys目录，这个每个节点之间彼此可以免密码连接
• 下载并解压hadoop安装包
• 配置namenode，修改site文件
• 配置hadoop-env.sh
• 配置masters和slaves文件
• 向各个节点复制hadoop
• 格式化namenode
• 启动hadoop
• 用jps检验各后台进程是否成功启动

Linux环境准备

什么是虚拟软件？

虚拟原件是一个可以使你在一台机器上同时运行二个或更多Windows、LINUX等系统。它可以模拟一个标准PC环境。这个环境和真实的计算机一样，都有芯片组、CPU、内存、显卡、声卡、网卡、软驱、硬盘、光驱、串口、并口、USB控制器等。
常用的虚拟软件：VMware workstation和VirtualBox

VMware workstation安装

1、双击VMware-workstation-full-14.1.2-8497320.exe
2、点击下一步进行安装

3、在弹出的“最终用户协议”窗口中，勾选“我接受许可协议中的条款”复选框，然后点击“下一步”按钮进入下一步

4、在弹出的“自定义安装”窗口中，可以点击“更改”按钮选择VMware Workstation的安装目录（本教程中采用默认目录）。选定安装位置后，勾选“增强型键盘驱动程序”，然后点击“下一步”按钮进入下一步

5、在弹出的“用户体验设置”窗口中，去掉“启动时检查产品更新”和“加入客户体验改进计划”复选框前的勾，然后点击“下一步”按钮进入下一步

如勾选，有新版时会提示你跟新版本，点next进行下一步

6、创建快捷方式

选择后点击下一步
7、配置完成，开始安装程序

点击continue
8、开始安装虚拟原件

9、完成，点击完成完成安装

10、在弹出的“系统重启提示”窗口中，点击“是”按钮重启系统（这里也可以点击“否”按钮，等激活VMware Workstation 14 Pro后再重启。）

11、双击桌面上的“VMware Workstation Pro”图标，在弹出的“VMware Workstation 14激活”窗口中，输入许可证“YC592-8VF55-M81AZ-FWW5T-WVRV0”（解压VMware-workstation-full-14.0.0目录下的Crack.zip,打开解压后的_keys.txt,可以见到所有可用的许可证），然后点击“继续”按钮请求激活

12、软件成功激活之后，可以见到VMware Workstation 14 Pro的主界面。在该界面中，可以点击帮助菜单中的“关于”菜单，查看软件的激活信息

安装Linux系统

Centos安装步骤

1、文件→新建虚拟机或直接点击创建新的虚拟机图标
2、选择典型（推荐）→下一步

3、安装程序光盘映像

4、填写linux安装基本信息

5、输入虚拟机名称和安装路径

6、设置磁盘大小

7、点击完成

8、开始安装

Linux系统相关配置

1、配置VMWare虚拟软件网卡，保证Windows机器能和虚拟机linux正常通信

点击VMware快捷方式，右键打开文件所在位置 -> 双击vmnetcfg.exe -> VMnet1 仅主机->修改子网 ip 设置网段：192.168.0.0 子网掩码：255.255.255.0 -> 应用-> 确定
双击vmnetcfg.exe进入下面设置界面

回到windows --> 打开网络和共享中心 -> 更改适配器设置 -> 右键VMnet1 -> 属性 -> 双击IPv4 -> 设置windows的IP：192.168.0.100 子网掩码：255.255.255.0 -> 点击确定

在虚拟软件上 –我的计算机-> 选中虚拟机 -> 右键 -> 设置-> 网络适配器-> 仅主机-> 确定

2、修改主机名

vim /etc/sysconfig/network

修改内容如下：

NETWORKING=yes

HOSTNAME=iflytek001
1
2
3
4
5
6
7

3、设置linux机器IP

第一种：通过Linux图形界面进行修改：

进入Linux图形界面 -> 右键点击右上方的两个小电脑 -> 点击vpn connections -> 选中当前网络System eth0 -> 点击edit按钮 -> 选择IPv4 -> method选择为manual -> 点击add按钮 -> 添加IP：192.168.0.2 子网掩码：255.255.255.0 网关：192.168.0.1 -> apply

第二种：修改配置文件方式（推荐使用）：

vim /etc/sysconfig/network-scripts/ifcfg-eth0

修改内容如下：

DEVICE="eth0"

BOOTPROTO="static"              

HWADDR="00:0C:29:3C:BF:E7"

IPV6INIT="yes"

NM_CONTROLLED="yes"

ONBOOT="yes"

TYPE="Ethernet"

UUID="ce22eeca-ecde-4536-8cc2-ef0dc36d4a8c"

IPADDR="192.168.0.2"           

NETMASK="255.255.255.0"          

GATEWAY="192.168.0.1" 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

修改后重启虚拟机，查看ip是否修改成功。

说明：在搭建Hadoop集群时，需克隆虚拟机，在克隆的虚拟机中如果ifconfig时，显示的网卡驱动名字不是默认的eth0。

解决办法：

首先需要修改70-persistent-net.rules文件：

vim /etc/udev/rules.d/70-persistent-net.rules

然后修改ifcfg-eth0文件：

vim /etc/sysconfig/network-scripts/ifcfg-eth0

4、修改主机名和IP的映射关系 vim /etc/hosts
修改内容如下：
192.168.0.2 iflytek001

5、关闭防火墙

#查看防火墙状态

service iptables status

#关闭防火墙

service iptables stop

#查看防火墙开机启动状态

chkconfig iptables --list

#关闭防火墙开机启动

chkconfig iptables off

#查看所有的开机状态

chkconfig –list
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

6、重启系统

reboot/init 6

7、测试

#测试用户名是否测通

ping iflytek001

#查看主机名
hostname -f ------只有重启后才有效

JDK环境

1、上传JDK压缩包，使用SSH工具上传压缩包
2、解压jdk

#创建文件夹：mkdir /iflytek

#解压：tar -zxvf  jdk-8u11-linux-x64.tar.gz -C  /iflytek
1
2
3

把指定的压缩包解压到指定的目录中，-C表示如果文件夹不存在则创建

3、将java添加到环境变量中

vim /etc/profile   进入编辑状态，添加如下内容：

#在文件中添加以下内容

export JAVA_HOME=/iflytek /jdk1.8.0_11

export PATH=$PATH:$JAVA_HOME/bin


1
2
3
4
5
6
7
8
9

4、刷新配置
source /etc/profile

5、测试jdk是否成功
java -version

6、测试案例

创建一个HelloWorld.java文件并编辑内容

编译和执行代码

Hadoop下载

1、访问hadoop网址http://hadoop.apache.org，点击“download”，进入下载页面，接着点击“Apache relwase archive”

2、在下载hadoop页面找到要下载的版本2.5.0,点击下图红框中的文件下载即可

以上是Apache的下载方式
3、下载CDH版本的Hadoop，网址为：

4、点击cdh à5进入版本下载。找到hadoop-2.5.0-cdh5.3.6进行下载

以下涉及到Hadoop版本的环境均基于此Hadoop版本

Hadoop单例模式搭建

1、将Hadoop包进行解压缩。参考命令如下

tar -zxvf hadoop-2.5.0-cdh5.3.6.tar.gz -C /iflytek
1

2、对Hadoop目录进行重命名。参考命令如下

mv hadoop-2.5.0-cdh5.3.6 hadoop-2.5.0.
1

3、修改hadoop-env.sh。在$HADOOP_HOME/etc/hadoop目录下执行命令：vi hadoop-env.sh，按i键之后进入编辑状态，参考命令如下：

vim hadoop-env.sh
export JAVA_HOME=/iflytek/jdk1.8.0_11
1
2

4、配置Hadoop相关环境变量。参考命令如下

vim /etc/profile

export HADOOP_HOME=/iflytek/hadoop-2.5.0

export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

#完成/etc/profile文件下的修改后，刷新/etc/profile文件

source /etc/profile
1
2
3
4
5
6
7
8
9

5、测试
在/opt中创建文件夹data，并在data文件夹中创建hello.txt文件，具体内容如下

运行hadoop jar命令。参考命令如下：

格式为：hadoop jar *.jar 包名.类名 input output

Input为本地文件输入路径；

Output为本地文件输出路径；
命令如下：

hadoop jar \

/iflytek/hadoop-2.5.0/share/hadoop/mapreduce/  \

hadoop-mapreduce-examples-2.5.0-cdh5.3.6.jar  wordcount \

/opt/data/hello.txt  /wordcount
1
2
3
4
5
6
7
8
9
10
11
12
13
14

6、执行结果如下，hadoop单例模式即安装成功

Hadoop伪分布式搭建

1、将下载后的软件，上传到虚拟机的/usr/local目录下。（下载目录http://archive.cloudera.com/cdh5/cdh/5/）

2、将Hadoop包进行解压缩。参考命令如下:

tar -zxvf hadoop-2.5.0-cdh5.3.6.tar.gz -C /usr/local/
1

3、对Hadoop目录进行重命名。参考命令如下:

mv hadoop-2.5.0-cdh5.3.6 hadoop-2.5.0
1

4、修改hadoop-env.sh。在$HADOOP_HOME/etc/hadoop目录下执行命令：vim hadoop-env.sh，按i键之后进入编辑状态，参考命令如下：

vim hadoop-env.sh
export JAVA_HOME=/iflytek/jdk1.8.0_11
1
2
3

5、修改core-site.xml。参考命令如下

<property>   
     <name>fs.default.name</name>
   <!--Hadoop的URI和端口-->    
   <value>hdfs://iflytek001:9000</value>
</property>
1
2
3
4
5

6、修改hdfs-site.xml。参考命令如下

<property>   
      <name>dfs.name.dir</name>
   <!--映射信息的保存路径-->   
   <value>/iflytek/data/namenode</value>
  </property>
 <property>   
   <name>dfs.data.dir</name>
   <!--真正的datanode数据保存路径-->   
   <value>/iflytek/data/datanode</value>
   </property>
 <property>   
    <name>dfs.tmp.dir</name>
   <!--临时目录设定-->   
   <value>/iflytek/data/tmp</value>
  </property>
 <property>   
     <name>dfs.replication</name>
   <!--缺省的块复制数量-->   
   <value>1</value>
 </property>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

7、修改mapred-site.xml。需要将mapred-site.xml.template的名字改为mapred-site.xml。命令为：mv mapred-site.xml.template mapred-site.xml。参考命令如下：

<property>
   <name>mapreduce.framework.name</name>
   <!--使用yarn运行mapreduce程序-->
   <value>yarn</value>
</property>
1
2
3
4
5

8、修改yarn-site.xml。参考命令如下：

<property>
   <name>yarn.resourcemanager.hostname</name>
   <!--运行ResourceManager机器所在的节点位置-->    
   <value>iflytek001</value>
</property>
<property>
   <name>yarn.nodemanager.aux-services</name>
   <value>mapreduce_shuffle</value>
</property>
1
2
3
4
5
6
7
8
9

9、配置Hadoop相关环境变量。参考命令如下：

vim /etc/profile

#添加内容

export HADOOP_HOME=/iflytek/hadoop-2.5.0

export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin

#刷新

source /etc/profile

#完成profile文件下的修改后，刷新profile文件
1
2
3
4
5
6
7
8
9
10
11
12
13

10、格式化namenode，在主机上执行以下命令。参考命令如下：

hdfs namenode -format
1

11、启动HDFS。参考命令如下：

start-dfs.sh
验证启动是否成功：jps、50070端口。
启动进程：

 namenode、secondarynamenode、datanode

在网站中输入主机名加端口号，主机名:端口
1
2
3
4
5
6
7

12、启动YARN与验证：
启动HDFS，参考命令如下：

start-yarn.sh

验证启动是否成功：jps、8088端口。

启动进程：resourcemanager、nodemanager

在网站中输入主机名加端口号，主机名:端口

Hadoop集群搭建

除了上述两种Hadoop环境搭建方式，在生产环境中，我们一般搭建Hadoop集群来处理海量数据问题，集群技术的特点：通过多台计算机完成同一个工作。达到更高的效率；两机或多机内容、工作过程等完全一样。如果一台死机，另一台可以起作用。
hadoop集群搭建准备

Hadoop集群搭建流程如下：
1、首先创建一台虚拟机iflytek001，Linux系统安装、环境配置和JDK安装、与Hadoop伪分布式的Linux系统安装、Linux环境配置和JDK安装一致，在此不再赘述。

2、克隆两台虚拟机iflytek002，iflytek003， 克隆完成无误后（把iflytek001内存改成1536MB）修改iflytek002和iflytek003的ip和主机名。

3、设置SSH免密码登录，在1.3.5中已介绍过SSH免密码登录配置，在此不再赘述。

4、设置集群时间同步，以iflytek001作为时间服务器。

1）看环境 #rpm -qa|grep ntp

2）编辑环境 #vim /etc/ntp.conf

3）编辑/etc/sysconfig/ntpd文件

4）查看ntpd状态：service ntpd status

5）打开iflytek001 ntpd状态：service ntpd start

6）永久打开iflytek001 ntpd状态：chkconfig ntpd on

7）分别在iflytek002和iflytek003配置时间更新crontab -e，内容如下：

  0-59/10 * * * * /usr/sbin/ntpdate iflytek001

8）分别在iflytek002与iflytek003同步

  /usr/sbin/ntpdate iflytek001
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

注意： 同步前把三台虚拟机的防火墙关掉 chkconfig iptables off
5、搭建iflytek001的hadoop环境，配置文件如下
1）hadoop-en.sh

export JAVA_HOME=/iflytek/jdk1.8.0_11
# The jsvc implementation to use. Jsvc is required to run secure datanodes.
#export JSVC_HOME=${JSVC_HOME}
  
export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-"/etc/hadoop"}
  
# Extra Java CLASSPATH elements.  Automatically insert capacity-scheduler.
for f in $HADOOP_HOME/contrib/capacity-scheduler/*.jar; do
  if [ "$HADOOP_CLASSPATH" ]; then
    export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:$f
  else
    export HADOOP_CLASSPATH=$f
1
2
3
4
5
6
7
8
9
10
11
12

2）core-site.xml

<configuration>
<property>
      <name>fs.defaultFS</name>
      <!—Hadoop主节点URI和端口-->
      <value>hdfs://iflytek001:9000</value>
   </property>
      <property>
      <name>hadoop.tmp.dir</name>
        <!--临时文件存放目录-->
      <value>/iflytek/hadoop-2.5.0/data/tmp</value>
   </property>
      <property>
      <name>fs.trash.interval</name>
      <!--文件废弃标识设定，0为禁止此功能,以秒为单位-->
      <value>420</value>
    </property>
</configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3）hdfs-site.xml

<configuration>
    <property>
        <name>dfs.namenode.secondary.http-address</name>
            <!--secondaryNameNode地址-->
        <value>iflytek003:50090</value>
    </property>
</configuration>
1
2
3
4
5
6
7

4）mapred-site.xml

<configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <!--使用yarn运行mapreduce程序-->
        <value>yarn</value>
    </property>
        <property>
        <name>mapreduce.jobhistory.address</name>
        <!--jobhistory 端口-->
        <value>iflytek001:10020</value>
    </property>
        <property>
        <name>mapreduce.jobhistory.webapp.address</name>
            <!--jobhistory 网页端口->
        <value>iflytek001:19888</value>
    </property>
</configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

5）yarn-site.xml

<configuration>
   <property>
        <name>yarn.resourcemanager.hostname</name>
    <!--运行ResourceManager机器所在的节点位置-->
        <value>iflytek002</value>
   </property>
     <property>
        <name>yarn.nodemanager.aux-services</name>
        <!--执行结束后收集各个container本地的日志 -->
        <value>mapreduce_shuffle</value>
   </property>
   <property>
        <name>yarn.log-aggregation-enable</name>
        <!--收集的日志的保留时间，以秒为单位 -->
        <value>true</value>
   </property>
      <property>
        <name>yarn.log-aggregation.retain-seconds</name>
        <value>420</value>
    </property>
</configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

6）Slaves

iflytek001
iflytek002
iflytek003

6、搭建完成后，拷贝到另外两台虚拟机（iflytek002,iflytek003）中，命令如下：

scp -r hadoop-2.5.0/ root@iflytek002:/iflytek/

scp -r hadoop-2.5.0/ root@iflytek003:/iflytek/
1
2
3

接着格式化：hdfs namenode -format

格式化成功后，分别在iflytek001，iflytek002中输入以下命令：

start-dfs.sh，start-yarn.sh
1

iflytek001：DataNode、NameNode、NodeManager

iflytek002：RedourceManager、NodeManager、DataNode

iflytek003：DataNode、SecondaryNameNode、NodeManager

完成后在iflytek001浏览器中进行测试：
主机：50070

主机：8088

SSH简介和使用

SSH(Secure Shell的缩写)是建立在应用层和传输层基础上的安全协议。SSH 是目前较可靠，专为远程登录会话和其他网络服务提供安全性的协议。利用 SSH 协议可以有效防止远程管理过程中的信息泄露问题。下面简单介绍SSH的使用方法。
1、无选项参数运行 SSH
ssh 192.168.0.2
注意： 当你第一次使用ssh登录远程主机时，会出现没有找到主机密钥的提示信息。如果你回答的是 NO，SSH 将不会继续连接，只有回答 Yes才会继续。输入"yes"后，系统会将远程主机的密钥加入到你的主目录下的 .ssh/hostkeys下，这样你就可以继续操作了。下一次再登录此主机时，SSH 就不会提示确认消息了。对此主机的真实验证信息已经默认保存在每个用户的 /用户主目录/.ssh 文件里。

2、指定登录用户

ssh -l root 192.168.0.2或者 ssh root@192.168.0.2。

3、指定端口
SSH 默认使用的端口号是 22。大多现代的 Linux 系统 22 端口都是开放的。如果你运行 ssh 程序而没有指定端口号，它直接就是通过 22 端口发送请求的。 一些系统管理员会改变 SSH 的默认端口号。就要使用 -p 选项，后面在加上 SSH 端口号 ssh root@192.168.0.2 -p 1234

SSH免密码登录原理——非对称加密算法

在日常的工作生产中， 我们经常需要进行数据的通讯，开发人员经常需要对数据进行加解密操作，以保证数据的安全。数据的加密算法非为对称加密和非对称加密两种，常用的DES、三重DES、AES等都属于对称加密，即通过一个密钥可以进行数据的加解密，密钥一旦泄漏，传输的数据则不安全。

非对称加密算法的核心源于数学问题，它存在公钥和私钥的概念，要完成加解密操作，需要两个密钥同时参与。我们常说的“公钥加密，私钥加密”或“私钥加密， 公钥解密”都属于非对称加密的范畴，后文中讲到的RSA算法也一种典型的非对称加密算法。公钥加密的数据必须使用私钥才可以解密，同样，私钥加密的数据也 只能通过公钥进行解密。

相比对称加密，非对称加密的安全性得到了提升，但是也存在明显的缺点，非对称加解密的效率要远远小于对称加解密。所以非对称加密往往被用在一些安全性要求比较高的应用或领域中。

在SSH安全协议的原理中， 是一种非对称加密与对称加密算法的结合:

说明： server A免登录到server B:

1、 在A上生成公钥私钥；

2、 将公钥拷贝给server B,要重命名成authorized_keys；

3、 Server A向Server B发送一个连接请求；

4、 Server B得到Server A的信息后,在authorized_key中查找,如果有相应的用户名和IP,则随机生成一个字符串,并用Server A的公钥加密,发送给Server A；

5、 Server A得到Server B发来的消息后,使用私钥进行解密,然后将解密后的字符串发送给Server B。Server B进行和生成的对比,如果一致,则允许免登录；

总之:A要免密码登录到B,B首先要拥有A的公钥,然后B要做一次加密验证。对于非对称加密,公钥加密的密文不能公钥解开,只能私钥解开。

SSH免密码登录配置

1、生成ssh免登录密钥，具体命令如下：

cd ~/.ssh

ssh-keygen -t rsa （四个回车）
1
2
3

执行完这个命令后，会生成两个文件id_rsa（私钥）、id_rsa.pub（公钥），如下图

2、将公钥拷贝到要免登录的机器上

ssh-copy-id  iflytek001
1

从上面可以看出执行ssh-copy-id iflytek001之后，在.ssh目录下多个文件authorized_key其内容就是密码值，此时ssh访问就不需要密码了，如下图：

HDFS体系架构及分析

HDFS简介

Hadoop Distributed File System(简称HDFS)是Hadoop分布式文件系统。以流式数据访问模式来存储超大文件，运行与硬件的集群之中。

HDFS有着高容错性（fault-tolerant）的特点，并且设计用来部署在低廉的（low-cost）硬件上。而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。

对外部客户机而言，HDFS就像一个传统的分级文件系统。可以创建、删除、移动或重命名文件，等等。但是 HDFS 的架构是基于一组特定的节点构建的，这是由它自身的特点决定的。这些节点包括 NameNode（仅一个），它在 HDFS 内部提供元数据服务；DataNode，它为 HDFS 提供存储块。由于仅存在一个 NameNode，因此这是 HDFS 的一个缺点（单点失败）。

存储在 HDFS 中的文件被分成块，然后将这些块复制到多个计算机中（DataNode）。这与传统的RAID架构大不相同。块的大小（通常为 64MB或128M）和复制的块数量在创建文件时由客户机决定。NameNode 可以控制所有文件操作。HDFS 内部的所有通信都基于标准的TCP/IP 协议。

HDFS体系架构

HDFS集群有两类节点并以管理者-工作者模式运行，即一个管理者和多个工作者。NameNode管理文件系统的命名空间。它维护着文件系统树及整个树内的所有文件和目录。这些信息以两个文件保存磁盘中：命名空间镜像文件和编辑日志文件。Namenode同时也记录着每个文件中各个块所在的数据节点信息，但它并不永久保存块的位置信息，这些信息会在系统启动时由数据节点重建

数据块

与一般文件系统一样，HDFS也有块（Block）的概念，默认大小为64M或128M,HDFS上的文件也被划分为块大小的多个分块作为独立的存储单元。
与通常的磁盘文件系统不同的是：

HDFS中小于一个块大小的文件不会占据整个块的空间（当一个1MB的文件存储在一个128MB的块中时，文件只使用1MB的磁盘空间，而不是128MB）

设置数据块的好处：
1、一个文件的大小可以大于集群任意节点磁盘的容量
2、容易对数据进行备份，提高容错能力
3、使用抽象块而非整个文件作为存储单元，大大简化存储子系统的设计
检测数据块情况：hdfs fsck /hdfspath -blocks -files

NameNode：文件系统管理节点

NameNode是整个HDFS文件系统的管理节点。它维护着整个文件系统的文件目录树，文件/目录的元信息和每个文件对应的数据块列表并可以用来接收用户的操作请求。

NameNode 在一个称为 FsImage 的文件中存储所有关于文件系统名称空间的信息。这个文件和一个包含所有事务的记录文件（这里是 EditLog）将存储在 NameNode 的本地文件系统上。FsImage 和 EditLog 文件也需要复制副本，以防文件损坏或 NameNode 系统丢失。

fsimage:元数据镜像文件。存储某一时段NameNode内存元数据信息；
edits:操作日志文件；
fsimage:保存最近一次checkpoint的时间；
以上这些文件是保存在linux的文件系统中

DataNode：数据节点

DataNode 是一个在HDFS实例中的单独机器上运行的软件节点。通常Hadoop 集群包含一个NameNode和大量DataNode。DataNode通常以机架的形式组织，所有机架通过一个交换机将所有系统连接起来。

DataNode 响应来自 HDFS 客户机的读写请求同时它们还响应来自 NameNode 的创建、删除和复制块的命令。NameNode 依赖来自每个 DataNode 的定期心跳（heartbeat）消息。每条消息都包含一个块报告，NameNode 可以根据这个报告验证块映射和其他文件系统元数据。如果 DataNode 不能发送心跳消息，NameNode 将采取修复措施，重新复制在该节点上丢失的块。

DateNode提供真实文件数据的存储服务。

文件块（block）：最基本的存储单位。对于文件内容而言，一个文件的长度大小是size，那么从文件的０偏移开始，按照固定的大小，顺序对文件进行划分并编号，划分好的每一个块称一个Block。HDFS默认Block大小是128MB，以一个256MB文件，共有256/128=2个Block。

不同于普通文件系统的是，HDFS中，如果一个文件小于一个数据块的大小，并不占用整个数据块存储空间

Secondary NameNode：HA（热备）的一个解决方案

Secondary NameNode有如下作用：
一是镜像备份；
二是日志与镜像的定期合并，以防止编辑日志过大；
三是当namenode发生故障时启用，并把数据恢复到namenode上，避免数据丢失；
注意： 以上两个过程同时进行，称为checkpoint。
备注：
镜像备份的作用：备份fsimage(fsimage是元数据发送检查点时写入文件);
日志与镜像的定期合并的作用:
将Namenode中edits日志和fsimage合并,防止(如果Namenode节点故障，namenode下次启动的时候，会把fsimage加载到内存中，应用edit log,edit log往往很大，导致操作往往很耗时。)内存泄露。

什么时候checkpiont？

fs.checkpoint.period 指定两次checkpoint的最大时间间隔，默认3600秒。
fs.checkpoint.size规定edits文件的最大值，一旦超过这个值则强制checkpoint，不管是否到达最大时间间隔。默认大小是64M。
NameNode与Secondary NameNode的工作流程如下：

1）NameNode管理着元数据信息，元数据信息会定期的刷到磁盘中，其中的两个文件是edits即操作日志文件和fsimage即元数据镜像文件，新的操作日志不会立即与fsimage进行合并，也不会刷到NameNode的内存中，而是会先写到edits中(因为合并需要消耗大量的资源)。当edits文件的大小达到一个临界值(默认是64MB)或者间隔一段时间(默认是1小时)的时候checkpoint会触发SecondaryNameNode进行工作。

2）当触发一个checkpoint操作时，NameNode会生成一个新的edits即上图中的edits.new文件，同时SecondaryNameNode会将edits文件和fsimage复制到本地。

3）SecondaryNameNode将本地的fsimage文件加载到内存中，然后再与edits文件进行合并生成一个新的fsimage文件即上图中的Fsimage.ckpt文件。

4）SecondaryNameNode将新生成的Fsimage.ckpt文件复制到NameNode节点。

5）在NameNode结点的edits.new文件和Fsimage.ckpt文件会替换掉原来的edits文件和fsimage文件，至此，刚好是一个轮回即在NameNode中又是edits和fsimage文件了。

6）等待下一次checkpoint触发SecondaryNameNode进行工作，一直这样循环操作。

HDFS文件存储机制

上图说明：HDFS是一个的主从结构，一个HDFS集群是由一个名字节点，它是一个管理文件命名空间和调节客户端访问文件的主服务器，当然还有一些数据节点，通常是一个节点一个机器，它来管理对应节点的存储。HDFS对外开放文件命名空间并允许用户数据以文件形式存储。

HDFS将要存储的大文件进行切割，切割后存放在既定的存储块（Block）中，并通过预先设定的优化处理，模式对存储的数据进行预处理，从而攻克了大文件储存与计算的需求。

NameNode是集群的主server，主要是用于对HDFS中全部的文件及内容数据进行维护，并不断读取记录集群中DataNode主机情况与工作状态，并通过读取与写入镜像日志文件的方式进行存储。NameNode负责接受client发送过来的信息，然后将文件存储位置信息发送给提交请求的client。由client直接与DataNode进行联系，从而进行部分文件的运算与操作。

DataNode在HDFS集群中担任任务详细执行角色，是集群的工作节点。文件被分成若干个同样大小的数据块，分别存储在若干个DataNode上。DataNode会定期向集群内NameNode发送自己的执行状态与存储内容，并依据NameNode发送的指令进行工作。

HDFS还能够对已经存储的Block进行多副本备份，将每一个Block至少拷贝到3个相互独立的硬件上，这样能够高速恢复损坏的数据；

用户能够使用既定的API接口对HDFS中的文件进行操作；

当client的读取操作错误发生的时候。client会向NameNode报告错误，并请求NameNode排除错误的DataNode后又一次依据距离排序。从而获得一个新的DataNode的读取路径。假设全部的DataNode都报告读取失败。那么整个任务就读取失败；

对于写出操作过程中出现的问题。FSDataOutputStream并不会马上关闭。client向NameNode报告错误信息。并直接向提供备份的DataNode中写入数据。
备份DataNode被升级为首选DataNode，并在其余2个DataNode中备份复制数据。
NameNode对错误的DataNode进行标记以便兴许对其进行处理。

HDFS操作

HDFS Shell命令介绍

我们可以通过命令行和HDFS打交道，进一步增加对HDFS的认识，HDFS命令行接口是一种最直接，也比较简单的方式
调用文件系统(FS)Shell命令应使用 bin/hadoop fs 的形式。
所有的FS shell命令使用URI路径作为参数。
URI格式是scheme://authority/path。HDFS的scheme是hdfs，对本地文件系统，scheme是file。其中scheme和authority参数都是可选的，如果未加指定，就会使用配置中指定的默认scheme。
例如：/parent/child可以表示成hdfs://namenode:namenodePort/parent/child，或者更简单的/parent/child（假设配置文件是namenode:namenodePort）。
大多数FS Shell命令的行为和对应的Unix Shell命令类似。

命令格式

文件系统shell包含各种shell like命令，可以直接和hdfs文件系统进行交互，就像对其他系统的支持一样，例如： Local FS, HFTP FS, S3 FS 。 shell命令执行格式如下：
hadoop fs {args}
hadoop dfs {args}
hdfs dfs {args}
说明：
hadoop fs：使用面最广，可以操作任何文件系统。
hadoop dfs与hdfs dfs：只能操作HDFS文件系统相关（包括与Local FS间的操作），前者已经Deprecated，一般使用后者。
以上所有fs shell命令都需要指定一个URI路径作为参数，URI的格式是: scheme://authority/path。对于hdfs协议是hdfs，本地的协议是file。一般情况下scheme和authority是可以省略的。这时需要在配置文件中指定。
HDFS文件的格式可以使用： hdfs://namenodehost/parent/child or simply as /parent/child

HDFS命令操作测试

输入hdfs dfs -help命令，查看hdfs的命令

在Windows下Eclipse连接Hadoop

Hadoop是java编写的，通过JavaAPI可以调用所有Hadoop文件系统的交互操作，它可以使用FileSystem类来提供文件系统操作。

在windows下解压hadoop-2.5.0-cdh5.3.6.tar.gz，将winutils.exe文件放置hadoop的bin目录下，如下：

接下来在代码中加上System.setProperty(“hadoop.home.dir”, “E:\BigData\Linux\hadoop-2.5.0”);即可连接成功

在Linux下Eclipse安装

解压eclipse压缩包到/iflytek目录下：切换到要解压的目录下，然后执行解压命令

成功解压后，在simple目录下产生一个目录eclipse

进入到elicpse目录下，查看目录结构，启动eclipse工具

或者进入Linux系统，将Eclipse的快捷方式传递到桌面上，并启动

启动eclipse后，进入eclipse运行界面

文件上传

方法一：自定义输入输出流,（比较底层的写法）

//获得filesystem对象

Configuration conf = new Configuration();
FileSystem fs= FileSystem.get(conf);

1
2
3

//设置输出流

Path path = new Path("hdfs://iflytek001:9000/ ");
FSDataOutputStream out = fs.create(path);

1
2
3

//设置输入流

FileInputStream in = new FileInputStream("/opt/data/hello.txt");
1

//调用IOUtils的copy方法，复制文件

IOUtils.copy(in, out);
1

注意：IOUtils类所在的包是在：import org.apache.commons.io.IOUtils;

方法二:直接调用hadoop hdfs 的copyFromLocalFile（）方法（建议使用此方法）

public static void main(String[] args throws Exception){
Configuration conf =new Configuration();
conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)
FileSystem fs =FileSystem.get(conf);
fs.copyFromLocalFile(new Path(“/opt/data/hello.txt”),new Path(“hdfs://iflytek001:9000/”)); 
}

1
2
3
4
5
6
7
8

注意：如果权限问题，可以修改目录权限.

或者修改运行参数：VM arguments:-DHADOOP_USER_NAME=hadoop

或者在程序中设置：fs =FileSystem.get(new URI(“hdfs://iflytek001:9000/”),conf,”hadoop”)

文件下载

方法一：(比较底层的写法）

//获得filesystem对象

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
1
2

//打开输入流

Path src = new Path("hdfs://iflytek001:9000/hello.txt");
FSDataInputStream in = fs.open(src);
1
2

//设置输出流

FileOutputStream out = new FileOutputStream("/opt/data/new");
1

//调用IOUtils的copy方法，复制文件

IOUtils.copy(in, out);
1

注意：必须把$HADOOP_HOME/share/etc/hadoop/下的core-site.xml 和 hdfs-site.xml 导入工程的src下，或者在conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)

方法二：调用copyToLocalFile()方法（封装好的写法）

public static void main(String[] args throws Exception){
   Configuration conf =new Configuration();
   conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)
   FileSystem fs =FileSystem.get(conf);
   fs.copyToLocalFile(new Path(“hdfs://iflytek001:9000/hello.txt”),new Path(“/opt/new”));
}

1
2
3
4
5
6
7
8

其他常用API操作

1、 重命名HDFS文件

    public static void main(String[] args throws Exception){
    Configuration conf =new Configuration();
    conf.set(“fs.defaultFS”,”hdfs://iflytek:9000/”)
    FileSystem fs =FileSystem.get(conf);
    //前一个参数是要修改的文件路径，后一个参数是修改后的路径名
    fs.rename (new Path(“/hello.txt”), new Path(“/hellocp.txt”));
}
1
2
3
4
5
6
7
8

2、 查看某个节点在HDFS集群的状态

 public static void main(String[] args throws Exception){
    Configuration conf =new Configuration();
    conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)
    FileSystem fs =FileSystem.get(conf);
    FileStatus f [] = fs.listStatus(new Path(“hdfs://iflytek001:9000/”));
    for (FileStatus fss:f){
       System.out.println(fss);
    }
}
1
2
3
4
5
6
7
8
9

3、 HDFS删除目录和文件

  public static void main(String[] args throws Exception){
    Configuration conf =new Configuration();
    conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)
    FileSystem fs =FileSystem.get(conf);
    fs.delete(new Path(“/hello.txt”),true);
}
1
2
3
4
5
6
7

4、 HDFS显示目录下的文件

   public static void main(String[] args throws Exception){
   Configuration conf =new Configuration();
   conf.set(“fs.defaultFS”,”hdfs://iflytek001:9000/”)
   FileSystem fs =FileSystem.get(conf);
   RemoteIterator<LocatedFileStatus> listFiles =   
   fs.listFiles(new Path(“hdfs://iflytek001:9000/”),true);
   while(listFiles.hasNext()){
       System.out.println(listFiles.next().hetPath());
   }
   }
1
2
3
4
5
6
7
8
9
10

Hadoop MapReduc

MapReduce的总执行过程

MapReduce运行的时候，通过Mapper运行的任务读取HDFS中的数据文件，然后调用自己的方法，处理数据，最后输出。Reducer任务会接收Mapper任务输出的数据，作为自己的输入数据，调用自己的方法，最后输出到HDFS的文件中

Mapper任务的执行过程

每个Mapper任务是一个java进程，它会读取HDFS中的文件，解析成很多的键值对，经过我们覆盖的map方法处理后，转换为很多的键值对再输出。整个Mapper任务的处理过程又可以分为以下几个阶段，如图所示：

1）第一阶段是读取HDF中的文件。每一行解析成一个<k,v>。每一个键值对调用一次Map函数。把输入文件按照一定的标准分片(InputSplit)，每个输入片的大小是固定的。默认情况下，输入片(InputSplit)的大小与数据块(Block)的大小是相同的。如果数据块(Block)的大小是默认值64MB，输入文件有两个，一个是32MB，一个是72MB，那么小的文件是一个输入片，大文件会分为两个数据块64MB和8MB，一共产生三个输入片。每一个输入片由一个Mapper进程处理。这里的三个输入片，会有三个Mapper进程处理。

2）第二阶段是对输入片中的记录按照一定的规则解析成键值对。有个默认规则是把每一行文本内容解析成键值对。“键”是每一行的起始位置(单位是字节)，“值”是本行的文本内容。

3）第三阶段是调用Mapper类中的map方法。第二阶段中解析出来的每一个键值对，调用一次map方法。如果有1000个键值对，就会调用1000次map方法。每一次调用map方法会输出零个或者多个键值对。

4）第四阶段是按照一定的规则对第三阶段输出的键值对进行分区。比较是基于键进行的。默认是只有一个区。分区的数量就是Reducer任务运行的数量。

5）第五阶段是对每个分区中的键值对进行排序。首先，按照键进行排序，对于键相同的键值对，按照值进行排序。比如三个键值对<2,2>、<1,3>、<2,1>，键和值分别是整数。那么排序后的结果是<1,3>、<2,1>、<2,2>。如果有第六阶段，那么进入第六阶段；如果没有，直接输出到本地的linux文件中。

6）第六阶段是对数据进行归约处理，也就是reduce处理。键相等的键值对会调用一次reduce方法。经过这一阶段，数据量会减少。归约后的数据输出到本地的linux文件中。本阶段默认是没有的，需要用户自己增加这一阶段的代码。

Reducer任务的执行过程

每个Reducer任务是一个java进程。Reducer任务接收Mapper任务的输出，归约处理后写入到HDFS中，可以分为如下图所示的几个阶段

1）第一阶段是Reducer任务会主动从Mapper任务复制其输出的键值对。Mapper任务可能会有很多，因此Reducer会复制多个Mapper的输出。

2）第二阶段是把复制到Reducer本地数据，全部进行合并，即把分散的数据合并成一个大的数据。再对合并后的数据排序。

3）第三阶段是对排序后的键值对调用reduce方法,键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对。最后把这些输出的键值对写入到HDFS文件中。

注意：在整个MapReduce程序的开发过程中，我们最大的工作量是覆盖map函数和覆盖reduce函数。

MapReduce数据类型与格式

Writable接口与序列化机制

1、Writable接口

MapReduce的任意Key和Value必须实现Writable接口

 public interface Writable {
       void write(DataOutput out) throwas IOException;
       void readFilelds(DataInput in) throws IOException;
   }
1
2
3
4

MapReduce的任意key必须实现WritableComparable接口

 public interface WritableComparable<T> extends Writable,Comparable<T>{
   }
1
2

常用的Writable实现类

Text一般认为它等价于java.lang.String的Writable。针对UTF-8序列。例:

   Text test = new Text("test");
   IntWritable one = new IntWritable(1);
1
2

2、序列化概念

序列化（Serialization）是指把结构化对象转化为字节流，便于在网络上传输或写到磁盘进行永久存储。反序列化（Deserialization）是序列化的逆过程。即把字节流转回结构化对象。
序列化在分布式数据处理的领域经常使用：进程间通信和永久存储
Hadoop的序列化格式：Writable
序列化格式特点：
Ø 紧凑：高效使用存储空间。
Ø 快速：读写数据的额外开销小
Ø 可扩展：可透明地读取老格式的数据
Ø 互操作：支持多语言的交互

Hadoop序列化的作用
Ø 序列化在分布式环境的两大作用：进程间通信，永久存储。
Ø Hadoop节点间通信。

MapReduce处理类

1、FileInputFormat
FileInputFormat是所有以文件作为数据源的InputFormat实现的基类，FileInputFormat保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。
1）InputFormat

InputFormat类的层次结构

2）InputSplit
在执行mapreduce之前，原始数据被分割成若干split，每个split作为一个map任务的输入，在map执行过程中split会被分解成一个个记录（key-value对），map会依次处理每一个记录。

FileInputFormat只划分比HDFS block大的文件，所以FileInputFormat划分的结果是这个文件或者是这个文件中的一部分。

如果一个文件的大小比block小，将不会被划分，这也是Hadoop处理大文件的效率要比处理很多小文件的效率高的原因。

当Hadoop处理很多小文件（文件大小小于hdfs block大小）的时候，由于FileInputFormat不会对小文件进行划分，所以每一个小文件都会被当做一个split并分配一个map任务，导致效率底下。

例如：一个1G的文件，会被划分成16个64MB的split，并分配16个map任务处理，而10000个100kb的文件会被10000个map任务处理。

2、TextInputFormat
TextInputformat是默认的处理类，处理普通文本文件。

文件中每一行作为一个记录，他将每一行在文件中的起始偏移量作为key，每一行的内容作为value。

默认以\n或回车键作为一行记录。

TextInputFormat继承了FileInputFormat。

3、其他输入类

4、Hadoop的输出：

MapReduce案列应用

运行商流量数据主要包括3个维度，其中包括电话号码、上传流量、下行流量三个特征

经过mapreduce处理，对流量进行统计排序

导入相关的JAR包

导入$HADOOP-HOME/share/hadoop/mapreduce和${HADOOP-HOME}/share/hadoop/mapreduce/lib/*中的jar包。部分jar如下：

创建封装数据类

新建一个Java工程，导包，建立一个FlowBean类。在FlowBean中生成get、set方法、toString方法以及生成构造函数。另外还需要在此类中将对象数据序列化到流中，再从数据流中反序列出对象的数据。参考代码如下：

public class FlowBean  implements
WritableComparable<FlowBean>{
private String phoneNB;
private long up_flow;
private long d_flow;
private long s_flow;
1
2
3
4
5
6

//在反序列化时，反射机制需要调用空参数构造函数，所以显示定义了一个空参构造函数

public FlowBean(){}
1

//为了对象数据的初始化方便，加入一个带参的构造函数

public FlowBean(String phoneNB,long up_flow,long d_flow){
this.phoneNB = phoneNB;
this.up_flow = up_flow;
this.d_flow = d_flow;
this.s_flow=up_flow+d_flow;
} 
public String getPhoneNB() {
return phoneNB;
}
public void setPhoneNB(String phoneNB) {
this.phoneNB = phoneNB;
}
public long getUp_flow() {
return up_flow;
}
public void setUp_flow(long up_flow) {
this.up_flow = up_flow;
}
public long getD_flow() {
return d_flow;
}
public void setD_flow(long d_flow) {
this.d_flow = d_flow;
}
public long getS_flow() {
return s_flow;
}
public void setS_flow(long s_flow) {
this.s_flow = s_flow;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

//将对象数据序列化到流中

@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(phoneNB);
out.writeLong(up_flow);
out.writeLong(d_flow);
out.writeLong(s_flow)  
}
1
2
3
4
5
6
7

//从数据流中反序列出对象的数据

//从数据流中读出对象字段时，必须跟序列化时的顺序保持一致

@Override
public void readFields(DataInput in) throws IOException {
phoneNB = in.readUTF();
up_flow = in.readLong();
d_flow = in.readLong();
s_flow = in.readLong();
} 
@Override
public String toString() {
return phoneNB+" "+up_flow+" "+d_flow +" "+s_flow;
}
@Override
public int compareTo(FlowBean o) {
return s_flow>o.getS_flow()?1:-1;
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

创建SortMapper.class

创建SortMapper.class

public class SortMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{
@Override
protected void map(
LongWritable key,
Text value,
Mapper<LongWritable,Text,FlowBean,NullWritable>.Context context)
throws IOException, InterruptedException {
         //拿到一行数据，切分出各字段，封装为一个flowbean，作为key输出
String line = value.toString();
String[] fields=line.split(" ");
String phoneNB = fields[0];
Long up_flow = Long.parseLong(fields[1]);
Long d_flow = Long.parseLong(fields[2]); 
context.write(new FlowBean(phoneNB,up_flow,d_flow),
        NullWritable.get());
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Mapper处理的流程：

1）将文件拆分为splits，并由MapReduce框架自动完成按行进行分割，将每一行分割为<key,value>对；

2）每一对<key,value>调用一次map函数，处理后生产新的<key,value>对，由Context传递给reduce处理；

3）Mapper对<key，value>对进行按key值进行排序，如果需要可以执行Combine过程，将key值相同的value进行合并，以便提供效率。最后得到Mapper的最终输出结果；

4）reduce处理，处理后将新的<key,value>对输出。

创建SortReducer.class

创建SortReducer.class

public  class SortReducer extends Reducer<FlowBean,NullWritable,Text, FlowBean>{ 
   @Override    
   protected void reduce(FlowBean key, Iterable<NullWritable> values,Reducer<FlowBean, NullWritable, 
 Text,FlowBean>.Context content) 
        throws IOException, InterruptedException {
        String phoneNB = key.getPhoneNB();
       content.write(new Text(phoneNB), key);
    }
   }
1
2
3
4
5
6
7
8
9

Reduce执行流程：

Mapreduce的主要思想是分而治之(divide and conquer)，分治算法。

将一个大的问题切分成很多小的问题，然后在集群中的各个节点上执行，这即是Map过程。在Map过程结束之后，会有一个Ruduce的过程，这个过程即将所有的Map阶段产出的结果进行汇集。

写MapReduce程序的步骤：

1）把问题转化为MapReduce模型；

2）写map类；

3）写reduce类；

4）编写测试类并设置运行的参数；

5）提交job。

创建SortMr.class

public class SortMr {   
 public static void main(String[] args) throws IOException, 
ClassNotFoundException, InterruptedException {   
     Job job = Job.getInstance(new Configuration());      
     job.setJarByClass(SortMr.class);       
     //设置Mapper的类       
     job.setMapperClass(SortMapper.class);     
     //设置Reducer的类      
     job.setReducerClass(SortReducer.class);        
     //设置Map输出key的类型        
     job.setMapOutputKeyClass(FlowBean.class);
     //设置Map输出value的类型
     job.setMapOutputValueClass(NullWritable.class);
     //设置输出key的类型
     job.setOutputKeyClass(Text.class);
     //设置输出value的类型
     job.setOutputValueClass(FlowBean.class);
     //设置输入文件地址
       FileInputFormat.setInputPaths(job,
        new Path("file:///root/stu/data/flow.txt"));        
     //设置输出文件地址
       FileOutputFormat.setOutputPath(job,
         new Path("file:///root/stu/out01")); 
          job.waitForCompletion(true);
     }
   }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

MapReduce源码分析

1、MR执行流程

1）客户端提交一个mr的jar包给JobClient(提交方式：hadoop jar …)

2）JobClient通过RPC和JobTracker进行通信，返回一个存放jar包的地址（HDFS）和jobId

3）client将jar包写入到HDFS当中(path = hdfs上的地址 + jobId)

4）开始提交任务(任务的描述信息，不是jar, 包括jobid，jar存放的位置，配置信息等等)

5）JobTracker进行初始化任务

6）读取HDFS上的要处理的文件，开始计算输入分片，每一个分片对应一个MapperTask

7）TaskTracker通过心跳机制领取任务（任务的描述信息）

8）下载所需的jar，配置文件等

9） TaskTracker启动一个java child子进程，用来执行具体的任务（MapperTask或ReducerTask）

10） 将结果写入到HDFS当中

2、MapReduce过程各个角色的作用

Partitioner编程

Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。

Partitioner的作用是对Mapper产生的中间结果进行分片，以便将同一个分组的数据交给同一个Reducer处理，它直接影响Reducer阶段的复杂均衡。

Partitioner只提供了一个方法：

getPartition(Text key,Text value,int numPartitions)

前两个参数是Map的Key和Value，numPartitions为Reduce的个数。

接下来我们在一个例子的基础上，使用Partitioner编程对数据进行分区。

创建AreaPartitioner.class

使用Partitioner函数，将数据按照手机号的地区代码分区。参考代码如下:

public class AreaPartitioner<KEY, VALUE> extends Partitioner<KEY, VALUE> {
        //手机号，地区代码       
        private static HashMap<String, Integer> areaMap = new HashMap<>();
        //静态代码块，将数据先加载到内存中
        static{       
        areaMap.put("131", 0);       
        areaMap.put("186", 1);       
        areaMap.put("138", 2);       
        areaMap.put("137", 3);       
        }     
        @Override       
        public int getPartition(KEY key, VALUE value,int numPartitions) {
        //从key中拿到手机号，查询手机归属地字典，不同的手机号返回不同的组号          
        Integer provinceCode = areaMap.get(key.toString().substring(0,3));          
        return provinceCode==null?4:provinceCode;        
        }     
       }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

创建AreaPartitionMapper.class

使用Map函数，拿到数据集中的一行数据，切分各个字段，抽取出我们需要的字段：电话号码，上行流量、下行流量，然后封装成kv发送出去。参考代码如下：

public class AreaPartitionMapper extends Mapper        
 <LongWritable, Text, Text, FlowBean> {     
  @Override    
   protected void map(LongWritable key, Text value, Mapper<LongWritable,Text,Text,FlowBean>.Context context)throws
  IOException, InterruptedException {         
  // 拿到一行数据         
  String line = value.toString();         
  // 切分字段         
  String[] strings = line.split(" ");         
  // 拿到我们需要的若干个字段         
  String phoneNB = strings[0];         
  long up_flow = Long.parseLong(strings[1]);         
  long d_flow = Long.parseLong(strings[2]);         
  // 将数据封装到一个flowbean中         
  // 以手机号为key，将流量数据输出去         
  context.write(new Text(phoneNB), new FlowBean(phoneNB, up_flow, d_flow));       
  }     
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

创建AreaPartitionReducer.class

传一组数据调用一次我们的reduce方法，reduce中的业务逻辑就是遍历values，然后进行累加输出，参考代码如下：

public class  AreaPartitionReducer extends Reducer
        <Text, FlowBean, Text, FlowBean> {
      protected void reduce(Text key, Iterable<FlowBean> values,
       Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException,   
   InterruptedException {
     //传进来的实例<13345677654,beans>,即多个该电话的键值对
     //取出values获得上下行和总流量求和
     long up_flow_sum = 0;
     long down_flow_sum = 0;
     for (FlowBean bean : values) {
          up_flow_sum += bean.getUp_flow();
         down_flow_sum += bean.getD_flow();        
        }
    //封装数据并输出
    context.write(key, new  FlowBean(key.toString(),up_flow_sum,down_flow_sum));  
     }
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

创建AreaPartition.class

最后在main方法中调用执行job提交。参考代码如下:

public class AreaPartition{    
  public static void main(String[] args) throws 
   IOException,ClassNotFoundException, InterruptedException {        
   Job job = Job.getInstance(new Configuration());        
   job.setJarByClass(FlowCountPartition.class);        
   job.setJarByClass(SortMr.class);        
   job.setMapperClass(FlowCountPartitionMapper.class);        
   job.setReducerClass(FlowCountPartitionReducer.class);        
   /**         
    * 加入自定义分区定义：AreaPartitioner         
     */        
     job.setPartitionerClass(AreaPartitioner.class);        
     /**         
     * 设置reduce task的数量，要跟AreaPartitioner返回的partitioner个数匹配         
     * 若reduce task多，会产生多余的几个空文件         
     * 若reduce task少，就会发生异常，因为有一些key没有对应reduce task    接收         
     * 但reduce task数量为1时，不会产生异常，因为所有key都会给这一个reduce task         
     * reduce task和map task指的是reducer和mapper在集群中运行的实例         
     */        
     job.setNumReduceTasks(5);        
     job.setMapOutputKeyClass(Text.class);        
     job.setMapOutputValueClass(FlowBean.class);        
     job.setOutputKeyClass(Text.class);        
     job.setOutputValueClass(FlowBean.class);             
      FileInputFormat.setInputPaths(job,
         new Path("file:///root/stu/out01"));        
         FileOutputFormat.setOutputPath(job,
         new Path("file:///root/stu/out02"));        
         job.waitForCompletion(true);    
       }
      }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

Combiner编程

 Combiner最基本是实现本地key的归并，Combiner具有类似本地reduce功能。
如果不用Combiner，那么所有的结果都是reduce完成，效率会相对低下，使用Combiner，先完成的map会在本地聚合，提升速度。
注意：Combiner的输出是Reducer的输入，如果Combiner是可插拔的，添加Combiner决不能改变最终的计算结果，在这种情况下Combiner只适用于那种Reduce的输入key/value和Reduce的输出key/value类型完全一致，且不影响最终结果的场景，比如累加、最大值。
当然，Combiner也可以是不可插拔的，也就是说有与没有Combiner最终的计算结果是不一样的，不可插拔的应用场景是数据的筛选，比如我们可以在Map执行之后，Reduce执行之前先对数据进行筛选，把坏数据给过滤掉，这样传到Reduce的数据就都是好的数据了，这样最终的结果肯定是与没有筛选数据的结果是不一样的。
1
2
3
4

大三的专业课就是老师讲完每个人完成实验并提交实验报告，作为一个已经将实验报告写（超）完的人（手动狗头），想着干点什么呢，于是就把我们老师给的学习课程网站的学习知识做了一个搬运工，谁让我闲的没事干呢！！！