Hadoop 笔记

• 适合大文件存储，支持TB、PB级的数据存储，并有副本策略
• 可以构建在廉价的机器上，并有一定的容错和恢复机制
• 支持流式数据访问，一次写入，多次读取最高效

HDFS 缺点

• 不适合大量小文件存储
• 不适合并发写入，不支持文件随机修改
• 不支持随机读低延时的访问方式

HDFS 写流程

1. 客户端向NameNode发起写数据请求
2. 分块写入DataNode节点，DataNode自动完成副本备份
3. DataNode向NameNode汇报存储完成，NameNode通知客户端

HDFS 读流程

1. 客户端向NameNode发起读数据请求
2. NameNode找出距离最近的DataNode节点信息
3. 客户端从DataNode分块下载文件

HDFS Shell命令

很多命令和 Linux 是一致的

​

./hdfs dfs -help #查看帮助文档

hdfs dfs -ls / #查看根目录

hdfs dfs -du -h #查看文件大小

hdfs dfs -text #查看文件

hdfs dfs -mkdir /test #创建test目录

hdfs dfs -copyFromLocal /home/hadoop/mk.txt /test/

hdfs dfs -cat /test/mk.txt

hdfs dfs -copyToLocal /test/mk.txt /home/hadoop/mk2.txt

hdfs dfs -chmod 777 /test/mk.txt

lzop

//使用lzop命令解压并查看
lzop -cd xxx.lzo |more

//压缩命令：
lzop xxx.log （生成xxx.log.lzo）

//列出test.lzo中各个文件的压缩信息
lzop -l test.lzo

程序操作HDFS

通过相应的API，调用相应的方法即可

YARN

Hadoop 2.0 之后的资源管理器，移除了原有的JobTracker和TaskTracker,所有的MapReduce需要通过YARN进行调度。

• ResourceManager: 分配调度资源，启动监控ApplicationManager，监控NodeManager
• ApplicationManager: 为MR程序申请资源，分配任务，负责数据切片，监控任务执行，容错
• NodeManager: 管理单个节点的资源，处理ResourceManager，ApplicationManager的命令

MapReduce 编程模型

1. 输入一个大的模型，通过Split之后，将其切分为多个分片。
2. Map阶段每个文件分片由单独的机器处理（需要编码）
3. Shuffle阶段.Map输入到Reduce输出的过程，涉及到网络传输
4. Reduce阶段将各个机器的结果汇总，并得到最终的结果（需要编码）

Hadoop 1.x 默认block大小64M

Hadoop 2.x 默认block大小128M

可以在hdfs-site.xml中设置参数：dfs.block.size

map任务个数

map任务个数是受多条件制约的，一般一个DataNode的map任务数量控制在10到100比较合适

• 可增大mapred.map.tasks；减少map个数
• 可增大mapred.min.split.size
• 如果要减少map个数，但有很多小文件，可将小文件合并为大文件，再使用上条准则

数据经过Map端输出后会进行网络混洗，经Shuffle后进入Reduce，在大数据量的情况下可能会造成巨大的网络开销。故可以在本地先按照key进行一轮 排序 与合并，在进行网络混洗，这个过程就是Combine.

图

partition任务,reduce任务,输出文件三者的数量总是相等的。所有数据默认会按照key值升序排序。

reduce任务个数

reduce任务不像Map任务那样受多个因素制约,大数据情况下，reduce数量不宜过少

• 可通过调节参数 mapred.reduce.tasks
• 可在代码中调用 job.setNumReduceTasks(int n)

问题

1. 如何通过Hadoop存储小文件？

   Hadoop Archives 出现就是为了缓解大量小文件消耗NameNode内存的问题，HAR文件是通过在HDFS上构建一个分层文件系统来工作；使用SequenceFile，用文件名（filename）作为key，并且文件内容（file contents）作为value；可以通过Hbase开发一个对象存储服务;可以采用压缩、合并小文件的策略，例如设置文件输入类型为CombineFileInputFormat格式

2. 当有节点故障的时候，集群是如何继续提供服务的，如何读？如何写？
3. 哪些是影响MapReduce性能的因素？

HBase

• 高可靠，高性能，面向列，可伸缩，实时读写的 分布式数据库
• 利用HDFS作为其文件存储系统，支持MR程序读取数据
• 存储非结构化和半结构化数据
• RowKey: 数据唯一标识，按字典排序
• ColumnFamily: 列族，多个列的集合，最多不超过三个
• TimeStamp: 支持多版本数据同时存在

HIVE

show databases;

use database;

show tables;

desc table;

dfs -ls 目录；

! 操作系统命令

select * from gen_yp_show_log_det_hour limit 5;

source mysql.sql;

hive -S #进入静默模式

Spark

• 基于内存计算 的大数据并行计算框架
• Spark是MapReduce的替代方案，兼容HDFS，Hive等数据源
• 抽象出分布式内存存储数据结构，弹性分布式数据集RDD
• 基于事件驱动，通过线程池复用线程提高性能

MapReduce

既可以直接编写相应 JAVA 处理程序，也可用 Python 等其他语言编写Map,Reduce处理过程，利用Hadoop进行执行

1. 编写 WordCount.java 程序

2. 编译 WordCount.java

   javac -classpath -d

3. 打包

   jar -cvf wordcount.jar *.class

4. 提交作业
   hdfs dfs -copyFromLocal

   hadoop jar
   yarn logs -applicationId xxxx    //查看日志

   Hadoop分布式缓存

   在执行MapReduce时，可能Mapper之间需要共享一些信息，如果信息量不大，可以将其从HDFS加载到内存中，这就是Hadoop分布式缓存机制

加载到内存发生在Job执行之前，每个从节点各自都缓存一份相同的共享数据。如果数据量过大，可以将其分批缓存，重复执行作业。

1. 在main方法中加载共享文件的HDFS路径，路径可以是目录也可以是文件。可以在路径末尾追加“#”+别名，在map阶段可以使用该别名

   String cache = "hdfs://10.105.***.***:8020/cache/file";    //目录或文件
   cache = cache + "#myfile";     //myfile是文件别名
   job.addCacheFile(new Path(cache).toUri(),conf)    //添加到job设置

2. 在Mapper类或Reducer的setup方法中，用输入流获取分布式缓存中的文件

   //该方法只执行一次，在map方法循环之前

   protected void setup（Context context） throws IOException,InterruptedException{
       FileReader reader = new FileReader("myfile");
       BufferedReader br = new BufferedReader(reader);

           }
   
   ### Mac Eclipse Hadoop-plus配置
   将插件jar包放在Eclipse的 **dropins** 文件夹下
   - 查看端口
   > hdfs端口号在：```$HADOOP_HOME/conf/core-site.xml中
   

mr端口号在： $HADOOP_HOME/conf/mapred-site.xml中

dfs.datanode.ipc.address  JobTracker端口，默认50020
fs.defaultFS    HDFS端口，默认8020

Python MapReduce

cat 0318part-00001|python mapper.py 
cat 0318part-00001|python mapper.py |sort -k1,1|python reducer.py

yarn

//查看日志
yarn logs -applicationId appid
//查看状态
yarn application -status
//杀掉任务
yarn application -kill appid