Hadoop面试题汇总-20221031

2022-11-14 20:08:14 浏览数 (1)

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中进行全局配置,表示开启公平调度策略。
代码语言:javascript复制
<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,对公平调度进行详细配置。
代码语言:javascript复制
<?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容量调度。

答:

  • 现在要划分的队列如下
代码语言:javascript复制
root 
├── prod 
└── dev 
	├── spark 
	└── hdp
  • 首先在yarn-site.xml中进行全局配置,表示开启容量调度策略。
代码语言:javascript复制
<property>
	<name>yarn.resourcemanager.scheduler.class</name>
	<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>
  • 配置capacity-scheduler.xml,对容量调度进行详细配置。
代码语言:javascript复制
<?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进行逻辑与计算?

答:

  • 源码部分如下
代码语言:javascript复制
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方法。
代码语言:javascript复制
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数量。
代码语言:javascript复制
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大小相同),具体计算规则为:
代码语言:javascript复制
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规则。
代码语言:javascript复制
mapreduce.input.fileinputformat.split.maxsize
mapreduce.input.fileinputformat.split.minsize
  • 当剩余的文件大于splitSize的1.1倍时,才进行拆分,避免生成较小的切片。例如当前文件为129M,小于splitSize(128M)的1.1倍,所以不做拆分。
代码语言:javascript复制
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使用自定义切片方法,它能够解决小文件问题。生成切片的过程包括:虚拟存储过程和切片过程两部分。切分时,需要设置虚拟存储切片最大值。
代码语言:javascript复制
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类,并重写方法。
代码语言:javascript复制
// 初始化资源,一般用于打开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。
代码语言:javascript复制
job.setInputFormatClass();

16、自定义InputFormat时,如果不想让文件在读取时被切片,可以怎么做?

答:

  • 重写isSplitable方法,返回false。

17、如果没有自定义Map、Reduce,默认会执行什么操作?

答:

  • Mapper会读取数据,然后输出(key,value)。Reducer接收数据,遍历value后输出(key,value)。相当于是按照InputFormat将数据读取为(key,value)格式后原样输出。

18、如何实现自定义排序?

答:

  • 默认排序方式是按照字典序升序,使用快速排序实现。
  • 自定义排序,需要在自定义数据类中,实现WritableComparable接口(Writable、Comparable),重写compareTo方法。
代码语言:javascript复制
// 如果返回正数,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方法。
代码语言:javascript复制
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,实现方法,完成自定义输出。
代码语言:javascript复制
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方法。
代码语言:javascript复制
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?

答:

  • 在驱动函数中将数据加载到缓存
代码语言:javascript复制
job.addCacheFile(new URI("path"));
  • 在Mapper的setup阶段,将文件读取到缓存集合中。
代码语言:javascript复制
URI[] cacheFiles = context.getCacheFiles();
String path = cacheFiles[0].toString();
BufferReader bufferReader = new BufferReader(new FileReader(path));
String line;
while (StringUtils.isNotEmpty(line = bufferReader.readLine())){
	//进行数据处理
}

0 人点赞