Hadoop面试题汇总
HDFS部分
1、请描述HDFS的写流程。
答:
- 首先由客户端向 NameNode 发起文件上传请求,NameNode 检查文件要上传的目录,并鉴权。
- 如果上传用户对此目录有权限,则允许客户端进行上传操作。客户端接收到允许指令后,将要上传的文件切分为 Block,之后按照顺序依次上传 block1、block2…block N,不允许多线程并发写入。
- 按照顺序,开始上传 block1。Client向 NameNode 发起请求,NameNode 会按照 block 副本放置策略,为 block1 选择合适的 DataNode 节点,并按照与客户端的路由由近到远的顺序进行排序,之后将 DataNode 列表返回给客户端。
- 客户端接收到 DataNode 列表后,便按照列表顺序(由近到远),依次与 DataNode 建立连接管道Pipline。
- 连接建立后,将 block1 以packet包的形式发送到第一个 DataNode 中,当数据写入到 DataNode 内存后,在落盘的同时会将block通过连接管道发送到第二个 DataNode 中。第二个 DataNode 接收到数据后,也会将数据推送到第三个 DataNode。
- 当最后一个 DataNode 接收到数据,并将数据写入到磁盘后,便通过Pipline管道向上一个 DataNode(即第二个)返回成功信息。第二个 DataNode 接收到返回的成功信息,并且也已经完成了数据的落盘,此时会向 第一个 DataNode 返回成功信息。第一个 DataNode 类似,接收到成功信息,并且完成数据落盘后,向Client客户端返回写入成功信息。
- 客户端接收到成功信息后,便向 NameNode 报告 block1 写入成功,然后按照此步骤,依次将剩余的 block存储到DataNode中。
- 所有的 block 存储完成后,NameNode 会在内存中生成文件所对应的元数据,提供数据查询功能。
2、请描述HDFS的读流程。
答:
- 客户端首先向 NameNode 发起读取文件的请求,NameNode 鉴定用户权限。
- 如果用户对文件有读取权限,则查询文件的元数据信息,将文件的Block组成、以及Block对应的DataNode存储位置按照与客户端的路由距离由近到远排序后返回给客户端。
- 客户端接收到 NameNode 的返回后,依次与最近的 DataNode 进行连接,请求读取 block 数据。
- DataNode将Block数据以packet包的形式发送到客户端,发送前会完成数据校验。
- 所有的 block 读取完成后,客户端会将 block 组装成文件,返回给用户。
3、请描述HDFS副本选择策略(3副本)。
答:
- 在第一个副本放置时,会随机选择 DataNode 节点进行存放,优先选择最近、最空闲的 DataNode 节点。如果 Client 在 DataNode 节点(登录到 DataNode 节点,调用客户端进行文件上传),会直接存放在当前节点(因为网络拓扑距离最近)。
- 第二个副本会放在与第一个副本相同的机架中,选择最空闲的节点。
- 第三个副本为了保证数据的安全性,会存放在其它机架中。
- 在单机架的情况下,会随机选择空闲节点进行存储。
4、请描述HDFS的缓存机制。
答:
HDFS提供了一个高效的缓存加速机制—— Centralized Cache Management ,它允许用户指定要缓存的HDFS路径。NameNode会和保存着所需Block数据的所有DataNode通信,并指导它们把块数据缓存在堆外内存(off-heap)中进行缓存。DataNode会通过心跳机制向NameNode汇报缓存状态。
NameNode查询自身的缓存指令集来确定应该缓存哪个路径。缓存指令持久化存储在fsimage和edit日志中,可以通过Java和命令行API被添加、移除或修改。
在使用HDFS完成数据缓存时,首先要创建一个缓存池。缓存池是一个管理实体,用于管理缓存指令组。缓存池拥有类UNIX的权限,可以限制哪个用户和组可以访问该缓存池。写权限允许用户向缓存池添加、删除缓存指令 。读权限允许用户列出缓存池内的缓存指令,还有其他元数据。
缓存池也可以用于资源管理,可以设置一个最大限制值,用于限制缓存的数据量。
缓存池创建成功后,可以通过命令,将HDFS某个目录、文件缓存到缓存池中,从而完成数据缓存功能。
代码语言:javascript复制#创建缓存组,默认为cache_data
hdfs cacheadmin -addPool cache_data -mode 0777
#生成一个1GB大小的文件
dd if=/dev/zero of=/tmp/test.zero bs=1M count=1024
#将文件上传到HDFS
hdfs dfs -put /tmp/test.zero /data
#生成缓存指令
hdfs cacheadmin -addDirective -path /data -pool cache_data -ttl 1d
#显示缓存池的信息
hdfs cacheadmin -listPools -stats cache_data
#统计信息,显示EXP Date
hdfs cacheadmin -listDirectives -path /data
#删除缓存指令
hdfs cacheadmin -removeDirectives -path /data
#删除缓存池
hdfs cacheadmin -removePool cache_data
扩展阅读:官网文档、官网文档翻译
4、请说出几个常用的HDFS Shell命令。
答:
HDFS命令有两种风格,①hadoop fs ②hdfs dfs 。在使用上没有区别,hadoop fs内部会转换为hdfs dfs来使用。
代码语言:javascript复制# HDFS Shell帮助命令,help更加详细,usage简略些
hadoop fs -help [option]
hadoop fs -usage [option]
# 查看文件目录
hadoop fs -ls /
hadoop fs -ls -R /
# 创建文件
hadoop fs -touchz /edits.txt
# 创建目录
hadoop fs -mkdir /tmp_data
# 追加文件
hadoop fs -appendToFile edits.txt /edits.txt
# 查看文件内容
hadoop fs -cat /edits.txt
# 上传文件
hadoop fs -put /LocalPath /HDFSPath
hadoop fs -copyFromLocal /LocalPath /HDFSPath
hadoop fs -moveFromLocal /LocalPath /HDFSPath
# 下载文件
hadoop fs -get /HDFSPath /LocalPath
hadoop fs -copyToLocal /HDFSPath /LocalPath
# 删除文件
hadoop fs -rm /edits.txt
hadoop fs -rm -skipTrash /edits.txt
# 递归删除
hadoop fs -rm -r /shell
# 移动文件
hadoop fs -mv /originFile /newFile
# 拷贝文件
hadoop fs -copy /FilePath /dictPath
# 文件查找
hadoop fs -find / -name part-r-00000
5、HDFS文件的x权限是指?
答:
HDFS文件的x权限没有实际意义,在使用时可以忽略。目录的x权限表示是否可以访问这个目录。
6、HDFS企业级调优
答:
调整namenode处理客户端的线程数为dfs.namenode.handler.count=20 * log2(Cluster Size)。比如集群规模为8台时,此参数设置为60。
NameNode有一个工作线程池,用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。对于大集群或者有大量客户端的集群来说,通常需要增大参数dfs.namenode.handler.count的默认值10。设置该值的一般原则是将其设置为集群大小的自然对数乘以20,即20logN,N为集群大小。
Yarn部分
1、如何开启JobHistoryServer。
答:
先配置mapred-site.xml,开启JobHistoryServer。
代码语言:javascript复制 <!-- MapReduce HistoryServer地址 -->
<property>
<name>mapreduce.jobhistory.address</name>
<value>node01:10020</value>
</property>
<!-- HistoryServer Web地址 -->
<property>
<name>mapreduce.jobhistory.webapp.address</name>
<value>node01:19888</value>
</property>
<!-- MapReduce日志保存位置 -->
<property>
<name>mapreduce.jobhistory.intermediate-done-dir</name>
<value>/mr-history/log</value>
</property>
<!-- HistoryServer日志保存位置 -->
<property>
<name>mapreduce.jobhistory.done-dir</name>
<value>/mr-history/done</value>
</property>
然后配置yarn-site.xml,开启日志聚合。
代码语言:javascript复制 <!-- 聚合日志开启后,保存到HDFS中 -->
<property>
<name>yarn.log-aggregation-enable</name>
<value>true</value>
</property>
对于yarn-site.xml,还可以增加以下额外参数。
代码语言:javascript复制<!-- 聚合日志保存时间 -->
<property>
<name>yarn.log-aggregation.retain-seconds</name>
<value>2592000</value><!--30 day-->
</property>
<!-- 聚合日志压缩类型 -->
<property>
<name>yarn.nodemanager.log-aggregation.compression-type</name>
<value>gz</value>
</property>
<!-- 不启用聚合日志,日志在本地保存的时间 -->
<property>
<name>yarn.nodemanager.log.retain-seconds</name>
<value>604800</value><!-- 7 day -->
</property>
<!-- nodemanager 本地文件存储目录 -->
<property>
<name>yarn.nodemanager.local-dirs</name>
<value>/opt/app/hadoop-2.7.7/hadoopDatas/yarn/local</value>
</property>
<!-- resourceManager 最多保存已完成任务个数 -->
<property>
<name>yarn.resourcemanager.max-completed-applications</name>
<value>1000</value>
</property>
配置文件分发到各个节点后,重启yarn。在HisotryServer节点,启动服务。
代码语言:javascript复制$HADOOP_HOME/sbin/mr-jobhistory-daemon.sh start historyserver
2、如何配置Yarn公平调度。
答:
- 首先在yarn-site.xml中进行全局配置,表示开启公平调度策略。
<property>
<name>yarn.resourcemanager.scheduler.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value>
<description>开启公平调度策略</description>
</property>
<property>
<name>yarn.scheduler.fair.allocation.file</name>
<value>/opt/app/hadoop-2.7.7/etc/hadoop/fair-scheduler.xml</value>
<description>公平调度策略配置文件目录</description>
</property>
<property>
<name>yarn.scheduler.fair.preemption</name>
<value>true</value>
<description>开启资源抢占</description>
</property>
<property>
<name>yarn.scheduler.fair.preemption.cluster-utilization-threshold</name>
<value>0.8f</value>
<description>当集群的整体资源利用率超过80%,则开始抢占</description>
</property>
<property>
<name>yarn.scheduler.fair.user-as-default-queue</name>
<value>true</value>
<description>默认提交到default队列</description>
</property>
<property>
<name>yarn.scheduler.fair.allow-undeclared-pools</name>
<value>false</value>
<description>任务无法提交到现有队列,是否允许新建一个队列</description>
</property>
- 然后再编辑fair-scheduler.xml,对公平调度进行详细配置。
<?xml version="1.0"?>
<allocations>
<!-- 每个队列中的默认调度策略,默认值是fair -->
<defaultQueueSchedulingPolicy>fair</defaultQueueSchedulingPolicy>
<user name="hadoop">
<!-- 用户hadoop最多运行的任务数 -->
<maxRunningApps>30</maxRunningApps>
</user>
<!-- 默认运行任务数 -->
<userMaxAppsDefault>10</userMaxAppsDefault>
<!-- 队列划分 -->
<!-- 共有4个队列:default、hadoop、develop、test -->
<!-- 权重weight分别是:1、2、1、1.5 -->
<queue name="root">
<minResources>512mb,4vcores</minResources>
<maxResources>102400mb,100vcores</maxResources>
<maxRunningApps>100</maxRunningApps>
<weight>1.0</weight>
<schedulingMode>fair</schedulingMode>
<aclSubmitApps> </aclSubmitApps>
<aclAdministerApps> </aclAdministerApps>
<queue name="default">
<minResources>512mb,4vcores</minResources>
<maxResources>30720mb,30vcores</maxResources>
<maxRunningApps>100</maxRunningApps>
<schedulingMode>fair</schedulingMode>
<weight>1.0</weight>
<!-- 允许提交任务的用户和组 -->
<!-- 格式为:用户名1,用户名2 用户名1所属组,用户名2所属组 -->
<!-- * 表示接收所有用户的任务 -->
<!-- 默认情况下,当任务没有指定队列时,会提交到default中 -->
<aclSubmitApps>*</aclSubmitApps>
</queue>
<queue name="hadoop">
<minResources>512mb,4vcores</minResources>
<maxResources>20480mb,20vcores</maxResources>
<maxRunningApps>100</maxRunningApps>
<schedulingMode>fair</schedulingMode>
<weight>2.0</weight>
<!-- 允许提交任务的用户和组 -->
<aclSubmitApps>hadoop hadoop</aclSubmitApps>
<!-- 允许管理任务的用户和组 -->
<aclAdministerApps>hadoop hadoop</aclAdministerApps>
</queue>
<queue name="develop">
<minResources>512mb,4vcores</minResources>
<maxResources>20480mb,20vcores</maxResources>
<maxRunningApps>100</maxRunningApps>
<schedulingMode>fair</schedulingMode>
<weight>1</weight>
<aclSubmitApps>develop develop</aclSubmitApps>
<aclAdministerApps>develop develop</aclAdministerApps>
</queue>
<queue name="test">
<minResources>512mb,4vcores</minResources>
<maxResources>20480mb,20vcores</maxResources>
<maxRunningApps>100</maxRunningApps>
<schedulingMode>fair</schedulingMode>
<weight>1.5</weight>
<aclSubmitApps>test,hadoop,develop test</aclSubmitApps>
<aclAdministerApps>test group_businessC,supergroup</aclAdministerApps>
</queue>
</queue>
<!-- 任务提交规则,由scheduler调度器决定提交的任务进入指定队列 -->
<!-- 包含多个rule标签,rule标签默认情况下create参数=true,表示当前rule可以创建一个新队列来存放任务 -->
<queuePlacementPolicy>
<!-- 任务被提交到指定的队列;如果队列不存在,则不创建 -->
<rule name="specified" create="false"/>
<!-- 任务被提交到,以提交用户所属组名称所命名的队列;如果队列不存在,则不创建 -->
<rule name="primaryGroup" create="false" />
<!-- 当不满足以上规则,则提交到root.default队列(默认值) -->
<rule name="default" queue="root.default"/>
</queuePlacementPolicy>
</allocations>
3、如何配置Yarn容量调度。
答:
- 现在要划分的队列如下
root
├── prod
└── dev
├── spark
└── hdp
- 首先在yarn-site.xml中进行全局配置,表示开启容量调度策略。
<property>
<name>yarn.resourcemanager.scheduler.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>
- 配置capacity-scheduler.xml,对容量调度进行详细配置。
<?xml version="1.0" encoding="utf-8"?>
<configuration>
<property>
<name>yarn.scheduler.capacity.root.queues</name>
<value>prod,dev</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.dev.queues</name>
<value>hdp,spark</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.prod.capacity</name>
<value>40</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.dev.capacity</name>
<value>60</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.dev.maximum-capacity</name>
<value>75</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.dev.hdp.capacity</name>
<value>50</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.dev.spark.capacity</name>
<value>50</value>
</property>
</configuration>
4、可以动态更新调度器策略吗?
答:
可以,在集群不重启的情况下,在主节点执行以下命令即可。
代码语言:javascript复制yarn rmadmin -refreshQueues
但动态更新,只能新增、关闭队列,不能删除队列。
5、作业提交时,如何指定到特定队列。
答:
在MR中,使用mapreduce.job.queuename参数来指定。
代码语言:javascript复制Configuration conf = new Configuration();
conf.set("mapreduce.job.queuename", "hadoop");
6、Yarn参数调优。
答:
根据实际调整每个节点和单个任务申请内存值
- yarn.nodemanager.resource.memory-mb 表示该节点上YARN可使用的物理内存总量,默认是8192(MB),注意,如果你的节点内存资源不够8GB,则需要调减小这个值,YARN不会智能的探测节点的物理内存总量,可能会导致系统宕机
- yarn.scheduler.maximum-allocation-mb 单个任务可申请的最多物理内存量,默认是8192(MB)
MapReduce
1、MapReduce调优。
答:
- 采用CombineTextInputFormat来作为输入,解决输入端大量小文件场景。大量的小文件会产生大量的map任务,任务都需要初始化,从而导致mr运行缓慢
- 减少spill溢写次数:通过调整mapreduce.task.io.sort.mb及mapreduce.map.sort.spill.percent参数的值,增大触发spill的内存上限,减少spill次数,从而减少磁盘io的次数。
- 减少merge合并次数:调整mapreduce.task.io.sort.factor参数(默认10),增大merge的文件数,减少merge的次数,从而缩短mr处理时间
- 在map之后,不影响业务逻辑的情况下,先进行combine处理,减少I/O。
- 设置合理的map、reduce个数
- map进行到一定阶段时,开始分配reduce资源。由mapreduce.job.reduce.slowstart.completedmaps参数配置(默认0.05,即map进行到5%时,开始为reduce分配资源并运行),可以适当调大,如0.8,减少reduce的等待时间。
- 尽量避免使用reduce
- 合理设置reduce端的buffer:默认情况下,数据达到一定阈值的时候,Buffer中的数据会写入磁盘,然后reduce会从磁盘中获得所有的数据。即Buffer与reduce没有关联的,中间多次写磁盘、读磁盘的过程。那么可以通过调整参数,使得Buffer中的数据可以直接输送到reduce,从而减少I/O开销;mapreduce.reduce.input.buffer.percent默认为0.0,当值大于0的时候,会保留指定比例的内存读Buffer中的数值直接拿给Reducer使用。这样一来,设置Buffer需要内存,读取数据需要内存,Reduce计算也需要内存,所以要根据作业的运行情况进行调整。
- 设置数据压缩:在map端可以设置数据压缩,减少shuffle开销。reduce处理结果落盘时,也可以进行数据压缩,减少数据量。
2、常见调优参数。
- 资源相关参数,在MR程序中设置即可生效。
配置参数 | 参数说明 |
---|---|
mapreduce.map.memory.mb | 一个MapTask可使用的资源上限(单位:MB),默认为1024。如果MapTask实际使用的资源量超过该值,则会被强制杀死。 |
mapreduce.reduce.memory.mb | 一个ReduceTask可使用的资源上限(单位:MB),默认为1024。如果ReduceTask实际使用的资源量超过该值,则会被强制杀死。 |
mapreduce.map.cpu.vcores | 每个MapTask可使用的最多cpu core数目,默认值: 1 |
mapreduce.reduce.cpu.vcores | 每个ReduceTask可使用的最多cpu core数目,默认值: 1 |
mapreduce.reduce.shuffle.parallelcopies | 每个Reduce去Map中取数据的并行数。默认值是5 |
mapreduce.reduce.shuffle.merge.percent | Buffer中的数据达到多少比例开始写入磁盘。默认值0.66 |
mapreduce.reduce.merge.inmem.threshold | Buffer中的数据文件达到多少后开始写入磁盘,默认1000 |
mapreduce.reduce.shuffle.input.buffer.percent | Buffer大小占Reduce可用内存的比例。默认值0.7 |
mapreduce.reduce.input.buffer.percent | 指定多少比例的内存用来存放Buffer中的数据,默认值是0.0 |
- 资源相关参数,需要重启yarn才会生效(yarn-site.xml)
配置参数 | 参数说明 |
---|---|
yarn.scheduler.minimum-allocation-mb | 给应用程序Container分配的最小内存,默认值:1024 |
yarn.scheduler.maximum-allocation-mb | 给应用程序Container分配的最大内存,默认值:8192 |
yarn.scheduler.minimum-allocation-vcores | 每个Container申请的最小CPU核数,默认值:1 |
yarn.scheduler.maximum-allocation-vcores | 每个Container申请的最大CPU核数,默认值:32 |
yarn.nodemanager.resource.memory-mb | 给Containers分配的最大物理内存,默认值:8192 |
- Shuffle性能优化的关键参数,重启yarn后生效(mapred-site.xml)
配置参数 | 参数说明 |
---|---|
mapreduce.task.io.sort.mb | Shuffle的环形缓冲区大小,默认100m |
mapreduce.map.sort.spill.percent | 环形缓冲区溢出的阈值,默认80% |
- Shuffle性能优化的关键参数,重启yarn后生效(mapred-site.xml)
配置参数 | 参数说明 |
---|---|
mapreduce.map.maxattempts | 每个Map Task最大重试次数,一旦重试参数超过该值,则认为Map Task运行失败,默认值:4。 |
mapreduce.reduce.maxattempts | 每个Reduce Task最大重试次数,一旦重试参数超过该值,则认为Map Task运行失败,默认值:4。 |
mapreduce.task.timeout | Task超时时间,经常需要设置的一个参数,该参数表达的意思为:如果一个Task在一定时间内没有任何进入,即不会读取新的数据,也没有输出数据,则认为该Task处于Block状态,可能是卡住了,也许永远会卡住,为了防止因为用户程序永远Block住不退出,则强制设置了一个该超时时间(单位毫秒),默认是600000。如果你的程序对每条输入数据的处理时间过长(比如会访问数据库,通过网络拉取数据等),建议将该参数调大,该参数过小常出现的错误提示是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。 |
3、MapReduce过程中,如何解决数据倾斜问题
答:
- 注意空数据的情况,否则空数据都会被分发到reduce-0中进行运算。可以提前对空数据进行过滤,或者将空数据均匀赋值。
- 使用Combiner进行预合并,精简数据,减少数据倾斜的可能。
- 可以进行自定义分区Partitioner,决定数据的分发过程。
- 采用Map Join,尽量避免Reduce Join
4、MapReduce过程中,如何避免小文件问题
答:
- 数据上传HDFS时,提前进行数据合并。可以物理合并,或者打包成sequence file。
- 存在HDFS上的小文件,可以使用hadoop archive进行归档。它能将多个小文件打包成一个HAR文件。
- 在MapReduce进行处理时,使用CombineFileInputFormat将多个文件合并成一个单独的split;或者开启jvm重用,设置mapreduce.job.jvm.numtasks值在10到20之间,一个map运行在一个jvm上,开启重用的话,该map在jvm上运行完毕后,此jvm中会继续运行此job的其他map任务。
5、请描述Reduce过程。
答:
- Copy阶段:ReduceTask从各个MapTask上远程拷贝数据,并针对某一块数据,如果其大小超过一定阈值(内存缓存*25%),则写到磁盘上,否则直接放到内存中(jvm*70%)。
- Merge阶段:在远程拷贝数据的同时,ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并,一次合并10个,合并成有序的大文件,以防止内存使用过多或磁盘上文件过多。
- Sort阶段:当所有map task的分区数据全部拷贝完,会对数据进行分组操作,将key相同的数据聚在一起。Hadoop采用了基于排序的策略,由于各个MapTask已经实现对自己的处理结果进行了局部排序,因此,ReduceTask只需对所有数据进行一次归并排序即可。
- Reduce阶段:数据交给reduce任务进行运算,将计算结果写到HDFS上。
6、MapReduce在进行HashPartitoner时,会获取key的hashCode,之后为什么要与Integer.MAX_VALUE进行逻辑与计算?
答:
- 源码部分如下
public class HashPartitioner<K, V> extends Partitioner<K, V> {
/** Use {@link Object#hashCode()} to partition. */
public int getPartition(K key, V value,
int numReduceTasks) {
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}
- (key.hashCode() & Integer.MAX_VALUE),因为Integer.MAX_VALUE最高位是符号位0,表示正数。如果hashCode得到的值为负数,符号位为1,会影响哈希取模的结果(Partition不能为负),此时进行逻辑与计算后,会变成一个正整数,从而保证取模结果正确。
7、如何实现自定义Partitioner?
答:
- 创建自定义类,继承Partitioner,重写getPartition方法。
public class MyPartitioner extends Partitioner<Text, IntWritable> {
@Override
public int getPartition(Text key, IntWritable value, int numPartitions) {
// TODO Auto-generated method stub
return 0;
}
}
- 在Job中设置setPartitionerClass,并配置ReduceTask数量。
job.setNumReduceTasks(5);
job.setPartitionerClass(MyPartitioner.class);
8、Hadoop为什么不用Java内置的序列化方法
答:
Java的序列化框架(Serializable)比较重,对象被序列化后,会携带很多额外信息(校验值、Header、继承体系等),不便于在网络中高效传输。所以Hadoop开发了自己的序列化机制。
9、Hadoop数据类型Writable,与Java基本数据类型有什么区别?
答:
代码语言:javascript复制public interface WritableComparable<T> extends Writable, Comparable<T>
Writable实现了WritableComparable接口,间接继承了Writable, Comparable类,实现了序列化、排序的功能。而这两个功能,在MapReduce中非常重要,排序是MapTask、ReduceTask默认操作,在集群中进行数据传输时要进行序列化。
10、Hadoop中如何实现自定义序列化
答:
在自定义类中,实现org.apache.hadoop.io.Writable接口,并实现write、readFields方法,完成序列化操作。
代码语言:javascript复制public class Person implements Writable {
private String name;
private int age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public String toString() {
return "My name is " this.name ", and i'm " this.age " years old.";
}
// 序列化
@Override
public void write(DataOutput out) throws IOException {
out.write(this.name.getBytes());
out.writeInt(this.age);
}
// 反序列化
@Override
public void readFields(DataInput in) throws IOException {
this.setName(in.readLine());
this.setAge(in.readInt());
}
}
11、在MapReduce进行数据处理时,会进行split数据切片,它的默认拆分规则是?如果不按照默认规则进行拆分,会发生什么现象?
答:
- MapReduce默认按照128M进行数据split切分(与HDFS Block大小相同),具体计算规则为:
Math.max(minSize, Math.min(maxSize, blockSize));
mapreduce.input.fileinputformat.split.minsize=1 默认值为1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue
blockSize为128M
- 可以通过改变minsize、maxSize参数,来调整split规则。
mapreduce.input.fileinputformat.split.maxsize
mapreduce.input.fileinputformat.split.minsize
- 当剩余的文件大于splitSize的1.1倍时,才进行拆分,避免生成较小的切片。例如当前文件为129M,小于splitSize(128M)的1.1倍,所以不做拆分。
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
...
bytesRemaining -= splitSize;
}
- 如果数据存储在HDFS中,按128M进行拆分。调整了split大小后,会导致数据移动。假设需要处理的文件大小为300M,存储到HDFS中后被拆分为3个Block块(128M、128M、44M)。此时MapReduce按照100M进行split,那么Block1拆出100M后剩余的28M会通过网络传输到Block2节点,Block2拆分100M后剩余56M通过网络传输到Block3节点,Block3此时正好拥有100M数据,完成切分。这个过程,会增加额外开销。
12、在MapReduce中InputFormat做了什么事情?
答:
- InputFormat抽象类拥有两个方法,getSplits、createRecordReader。
- 其中,getSplits完成数据的切片操作。createRecordReader为切片后的数据创建RecordReader。
- RecordReader可以将数据拆分为key/value形式。
13、请描述MapReduce所有的FileInputFormat,并分别说明它们的用法(除CombineTextInputFormat)。
答:
- TextInputFormat、KeyValueInputFormat、NLineInputFormat、CombineTextInputFormat、FixedLengthInputFormat、SequenceFileInputFormat。
- TextInputFormat使用默认InputFormat的切片方法,在进行数据读取时,createRecordReader会返回LineRecordReader,它会将每一行数据的行号作为key,这行数据作为value。
- NLineInputFormat使用自定义切片方法,每N行会拆分为一片。在进行数据读取时,createRecordReader会返回LineRecordReader,它会将每一行数据的行号作为key,这行数据作为value。
- KeyValueInputFormat使用默认InputFormat的切片方法,在进行数据读取时,createRecordReader会返回KeyValueLineRecordReader,它会将每一行数据默认按照t进行切分,拆分后赋值给key、value。可以通过配置conf.set(KeyValueLineRecordReaderKEY_VALUE_SEPERATOR, “t”)来设定分隔符。
- FixedLengthInputFormat使用默认InputFormat的切片方法。在进行数据读取时,createRecordReader会返回FixedLengthRecordReader,它会将数据的偏移量作为key,数据作为value。
- SequenceFileInputFormat使用默认InputFormat的切片方法。在进行数据读取时,createRecordReader会返回SequenceFileRecordReader,它会读取Sequence文件。
14、请详细描述下CombineTextInputFormat。
答:
- CombineTextInputFormat使用自定义切片方法,它能够解决小文件问题。生成切片的过程包括:虚拟存储过程和切片过程两部分。切分时,需要设置虚拟存储切片最大值。
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
- 虚拟存储过程:将输入目录下所有文件大小,依次和设置的setMaxInputSplitSize值比较,如果不大于设置的最大值,逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍,那么以最大值切割一块;当剩余数据大小超过设置的最大值且不大于最大值2倍,此时将文件均分成2个虚拟存储块(防止出现太小切片)。 例如setMaxInputSplitSize值为4M,输入文件大小为8.02M,则先逻辑上分成一个4M。剩余的大小为4.02M,如果按照4M逻辑划分,就会出现0.02M的小的虚拟存储文件,所以将剩余的4.02M文件切分成(2.01M和2.01M)两个文件。
- 切片过程: (a)判断虚拟存储的文件大小是否大于setMaxInputSplitSize值,大于等于则单独形成一个切片。 (b)如果不大于则跟下一个虚拟存储文件进行合并,共同形成一个切片。 (c)测试举例:有4个小文件大小分别为1.7M、5.1M、3.4M以及6.8M这四个小文件,则虚拟存储之后形成6个文件块,大小分别为: 1.7M,(2.55M、2.55M),3.4M以及(3.4M、3.4M) 最终会形成3个切片,大小分别为: (1.7 2.55)M,(2.55 3.4)M,(3.4 3.4)M
15、如何自定义InputFormat?
答:
- 创建自定义类,继承需要的InputFormat,如FileInputFormat。重写createRecordReader方法,返回自定义RecordReader。
- 创建自定义RecordReader,继承RecordReader类,并重写方法。
// 初始化资源,一般用于打开IO流
// 常用IO流为FSDataInputStream,默认会定义在成员变量中:inputStream
public void initialize(InputSplit split,
TaskAttemptContext context
) throws IOException, InterruptedException
{
FileSplit fs = (FileSplit) split;
Path path = fs.getPath();
FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
inputStream = fileSystem.open(path);
}
// 关闭资源,一般用于关闭IO流
public void close() throws IOException {
IOUtils.closeStream(inputStream);
}
// 类似于指针,如果要读取的数据存在,返回true,否则返回false
public boolean nextKeyValue() throws IOException, InterruptedException {
}
// 获取当前行的key
public KEYIN getCurrentKey() throws IOException, InterruptedException;
// 获取当前行的value
public VALUEIN getCurrentValue() throws IOException, InterruptedException;
// 返回数据读取进度,0-1
public float getProgress() throws IOException, InterruptedException;
- 为Job指定InputFormat。
job.setInputFormatClass();
16、自定义InputFormat时,如果不想让文件在读取时被切片,可以怎么做?
答:
- 重写isSplitable方法,返回false。
17、如果没有自定义Map、Reduce,默认会执行什么操作?
答:
- Mapper会读取数据,然后输出(key,value)。Reducer接收数据,遍历value后输出(key,value)。相当于是按照InputFormat将数据读取为(key,value)格式后原样输出。
18、如何实现自定义排序?
答:
- 默认排序方式是按照字典序升序,使用快速排序实现。
- 自定义排序,需要在自定义数据类中,实现WritableComparable接口(Writable、Comparable),重写compareTo方法。
// 如果返回正数,o在前面;返回负数,表示o在后面;相等,返回0
// 意味着如果降序,则o>this时,返回正数;o<this时返回负数。升序时,this>o返回正数;this<o返回负数
public int compareTo(T o) {
return Long.compare(o.value,this.value);
};
19、什么是Combiner?如何设置?
答:
- Combinner是在Map端提前进行的数据聚合操作,对Key相同的数据进行合并。以减少在shuffle阶段传输的数据量。
- 在Job中使用setCombinerClass进行配置。
20、如果Map输出时value没有意义(为空),应该如何处理?
答:
- 在使用Context写出时,value定义为NullWritable.get(),表示数据为空。
21、如何实现自定义分组
答:
- 自定义类,继承WritableComparator类。实现OrderComparator、comparable方法。
public class OrderGroupingComparator extends WritableComparator {
// 初始化空对象用于数据接收
protected OrderGroupingComparator() {
super(OrderBean.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
OrderBean aBean = (OrderBean) a;
OrderBean bBean = (OrderBean) b;
int result;
if (aBean.getOrder_id() > bBean.getOrder_id()) {
result = 1;
} else if (aBean.getOrder_id() < bBean.getOrder_id()) {
result = -1;
} else {
result = 0;
}
return result;
}
}
22、Reduce如何实现分组操作?
答:
- Reduce分组的结果是,把key相同的数据,value值放置在一个迭代器中,即(key,Iterable(values))。
- 从Map端拿到所需文件后,合并成一个有序的大文件。然后指针会指向文件的第一个数据,将key,value进行反序列化,处理后,指针会指向下一条数据。此时会判断当前数据是否和上一条数据为一组,如果是一组则返回true,否则返回false。
23、如何实现自定义输出?
答:
- 自定义MyRecordWriter,继承RecordWriter,实现方法,完成自定义输出。
public class MyRecordWriter extends RecordWriter<Text, IntWritable> {
private FSDataOutputStream file;
// 自定义方法,开流
public void initialize(TaskAttemptContext job) throws IOException {
String outdir = job.getConfiguration().get(FileOutputFormat.OUTDIR);
FileSystem fileSystem = FileSystem.get(job.getConfiguration());
file = fileSystem.create(new Path(outdir));
}
// 将数据写出
@Override
public void write(Text key, IntWritable value) throws IOException, InterruptedException {
// TODO Auto-generated method stub
file.write(value.hashCode());
}
// 关闭资源
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
IOUtils.closeStream(file);
}
}
- 自定义OutputFormat,继承FileOutputFormat,重写getRecordWriter方法。
public class MyOutputFormat extends FileOutputFormat<Text, IntWritable> {
@Override
public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext job)
throws IOException, InterruptedException {
MyRecordWriter mRecordWriter = new MyRecordWriter();
mRecordWriter.initialize(job);
return mRecordWriter;
}
}
24、如何在MapReduce中实现MapJoin?
答:
- 在驱动函数中将数据加载到缓存
job.addCacheFile(new URI("path"));
- 在Mapper的setup阶段,将文件读取到缓存集合中。
URI[] cacheFiles = context.getCacheFiles();
String path = cacheFiles[0].toString();
BufferReader bufferReader = new BufferReader(new FileReader(path));
String line;
while (StringUtils.isNotEmpty(line = bufferReader.readLine())){
//进行数据处理
}