mr 将得到的split 分配对应的 task，每个任务处理相对应的 split，将 split 以 line 方式读取每一行数据，将数据依次读取到100M（maprdeuce.task.io.sort.mb）的环形缓冲区读取过程中一旦到达阈值（mapreduce.map.sort.spill.percent）80M进行溢写操作，spiller线程溢写到磁盘（mapreduce.cluster.local.dir）目录中，期间会进行kv分区（分区数由reduce数来决定）默认使用hashpartition，再将分区中数据进行key的排序（默认排序规则是字典和升序），如果设置了setCombinerClass 则会对每个分区中的数据进行 combiner 操作，如果设置了output.compress压缩格式会对溢写的数据进行压缩，最后merge根据分区规则将数据归并到同一个文件中等待reduce的拉取，nodemanger将启动一个mapreduce_shuffle服务将数据以http方式拉取到reduce端，reduce处理阶段当达到阈值（默认0.66）或map输出数的阈值（默认100）会进行merge（同一分区的一组数据会先进行归并）|sort（将归并好的数据进行排序）|group（判断迭代器中的元素是否可以迭代）,处理完成mr将同一个分区内的数据，在hdfs中以文件形式体现出来，几个分区就会创建几个文件。 其中reduce端的merge达到阈值会触发，而sort和group是设置后才会触发。 有效的理解mr工作流程可大大提升程序运行效率，因此mr 的 shuffle 也被称为奇迹开始的地方 spilit是在mr 处理的map端之前产生的概念，split切片大小，默认等于block*1.1，在FileInputFormat中计算切片大小的逻辑： blocksize：默认是 128M，可通过 dfs.blocksize 修改 minSize：默认是 1，可通过 mapreduce.input.fileinputformat.split.minsize 修改 maxsize：默认是 Long.MaxValue，可通过 mapreduce.input.fileinputformat.split.maxsize 修改 Hadoop FileInputFormat 源码： mr 是通过split切片来决定 task 的并行度。 为什么split不是与block 一一对应的？ 大量小文件场景，map进程造成资源严重浪费。 针对大小文件场景可以手动配置。 namenode，在基于主从架构的hdfs文件系统中是主节点，其主要职责就是对hdfs中的文件的元信息，副本数，文件目录树，block 数据节点信息； datanode，它是从节点也是数据节点，基于本地磁盘存储 block（文件的形式），有相关数据块的长度、效验和、时间戳，与namnode保持心跳，汇报 block 状态。 secondaryNameNode，检查点节点，namenode 日志高可用的关键，其主要作用就是将namenode的元数据日志信息合并后备份，防止元数据丢失。 元信息：是数据文件的block大小，文件副本存储位置，副本数量，block 数量，主要体现在edits文件和fsimage文件。 副本数：hdfs 中同一个文件在多个节点中所存储的总数量，也是实现持久化和保证安全性的关键。 文件目录树：hdfs提供了一个可以维护的文件目录，该文件目录下存储着有关所有hdfs的文件。 block 数据节点信息：如a文件在01和02节点中存储，该信息称为数据节点信息。 edits：记录 它分布在磁盘上的多个文件，名称由前缀 fsimage：记录的是数据块的位置信息、数据块的冗余信息（二进制文件） 由于 没有体现 每个 为什么引入 secondaryNameNode？ 由于只有在重启时 但 editlog 日志大小会随着时间变的越来越大，导致系统重启，根据日志恢复元数据的时间会越来越长； 为了避免这种情况，引入检查点机制checkpoint，命名空间镜像 fsimage 就是 HDFS 元数据的持久性检查点，即将内存中的元数据落磁盘生成的文件； 了解详细可以访问我另一个博客：hdfs详细 https://blog.csdn.net/qq_43259670/article/details/105882983 1、向client端提交MapReduce job. 2、随后yarn的ResourceManager进行资源的分配. 3、由NodeManager进行加载与监控containers. 4、通过applicationMaster与ResourceManager进行资源的申请及状态的交互，由NodeManagers进行MapReduce运行时job的管理. 5、通过hdfs进行job配置文件、jar包的各节点分发。 edits：记录 它分布在磁盘上的多个文件，名称由前缀 fsimage：记录的是数据块的位置信息、数据块的冗余信息（二进制文件） 由于 没有体现 每个 为什么使用？ 滚动将增量的 一般就是读写的工作流程，因为hdfs 主要还是对文件存储与读写。 读流程： client端创建一个代理对象与namenode进行rpc通信，拿到namenode对象后请求获取文件的元信息，namenode效验无误后将元信息返回，client获取到元信息之后根据元信息读取相应datanode的block块，将block合并成一个文件进行返回。 写流程： client端创建一个代理对象与namenode进行rpc通信，拿到namenode对象后请求创建文件的元信息，namenode触发副本放置策略，返回元数据信息，client和datanode建立piepline连接，client将packet放入一个队列中，并向第一个datanode发送packet这一过程中上游节点同时发送下一个packet，当 1.x 可能产生的问题是前两个副本在同一机架当机架出现问题时会丢失两个副本 2.x 了解详细可以访问我另一个博客：hdfs详细 https://blog.csdn.net/qq_43259670/article/details/105882983 1、执行查询：Hive接口，命令行或 web UI发送查询驱动程序 2、get Plan：驱动程序查询编译器 3、词法分析/语法分析 4、语义分析 5、逻辑计划产生 6、逻辑计划优化 7、物理计划生成 8、物理计划优化 9、物理计划执行 10、查询结果返回 提示：以上是hive的大致工作原理流程，一般面试问到这里就算比较深入了 创建表时：创建内部表时，会将数据移动到数据仓库指向的路径；若创建外部表，仅记录数据所在的路径， 不对数据的位置做任何改变。 删除表时：在删除表的时候，内部表的元数据和数据会被一起删除， 而外部表只删除元数据，不删除数据。这样外部表相对来说更加安全些，数据组织也更加灵活，方便共享源数据。 提示：内部表与外部表的区别一定要掌握，通常情况下我们都会使用外部表保证数据安全性，但是像中间表，结果表这种我们就会考虑使用内部表（管理表） 分区 是指按照数据表的某列或某些列分为多个区，区从形式上可以理解为文件夹，比如我们要收集某个大型网站的日志数据，一个网站每天的日志数据存在同一张表上，由于每天会生成大量的日志，导致数据表的内容巨大，在查询时进行全表扫描耗费的资源非常多。 那其实这个情况下，我们可以按照日期对数据表进行分区，不同日期的数据存放在不同的分区，在查询时只要指定分区字段的值就可以直接从该分区查找 分桶 分桶将整个数据内容安装某列属性值得hash值进行区分，如要按照name属性分为3个桶，就是对name属性值的hash值对3取摸，按照取模结果对数据分桶。 如取模结果为0的数据记录存放到一个文件，取模为1的数据存放到一个文件，取模为2的数据存放到一个文件 总结：分区就是在hdfs上分目录（文件夹），分桶就是分文件。 方案一：上传数据后修复表 提示：这里我们如果直接将新的分区文件上传到hdfs上，因为hive没有对应的元数据所以是无法查询到数据的，所以我们要进行表修复或者添加分区。 不可以，因为load数据的话hdfs下只会有一个文件无法完成分桶的效果，分桶和mapredue中的分区是一样的道理，所以我们要借助中间表导入数据。 order by会对所给的全部数据进行全局排序 ，不管来多少数据，都只启动一个reducer来处理 。 sort by是局部排序 ，sort by会根据数据量的大小启动一到多个reducer来干活，并且，它会在进入reduce之前为每个reducer都产生一个排序文件 。 distribute by 控制map结果的分发，它会将具有相同字段的map输出分发到一个reduce节点上做处理 。 cluster by 可以理解为一个特殊的distribute by和sort by的结合，当distribute by和sort by后面所跟的列名相同时，就等同于直接使用cluster by 跟上该列名。但是被cluster by指定的列最终的排序结果只能是降序，而且无法指定asc和desc。 提示：这个问题面试问的频率很高，大家一定要注意区分以下。 不可以。 原因：执行顺序！！！order by的执行顺序在select之后，所以需使用重新定义的列名进行排序。 提示：理解sql的执行顺序更加有利于大家写sql 数据倾斜就是数据的分布不平衡，某些地方特别多，某些地方又特别少，导致的在处理数据的时候，有些很快就处理完了，而有些又迟迟未能处理完，导致整体任务最终迟迟无法完成，这种现象就是数据倾斜。 针对mapreduce的过程来说就是，有多个reduce，其中有一个或者若干个reduce要处理的数据量特别大，而其他的reduce处理的数据量则比较小，那么这些数据量小的reduce很快就可以完成，而数据量大的则需要很多时间，导致整个任务一直在等它而迟迟无法完成。 比如某些业务数据作为key的字段本就很集中，那么结果肯定会导致数据倾斜啊。 既然有数据倾斜这种现象，就必须要有数据倾斜对应的处理方案啊。 1） 通常情况下，作业会通过input的目录产生一个或者多个map任务。 2） 是不是map数越多越好？ 3） 是不是保证每个map处理接近128m的文件块，就高枕无忧了？ 在map执行前合并小文件，减少map数： CombineHiveInputFormat 具有对小文件进行合并的功能（系统默认的格式） 这个参数表示执行前进行小文件合并，前面三个参数确定合并文件块的大小，大于文件块大小128m的，按照128m来分隔，小于128m，大于100m的，按照100m来分隔，把那些小于100m的（包括小文件和分隔大文件剩下的），进行合并。 当input的文件都很大，任务逻辑复杂，map执行非常慢的时候，可以考虑增加Map数，来使得每个map处理的数据量减少，从而提高任务的执行效率。 增加map的方法为 1、调整reduce个数方法一 1）每个Reduce处理的数据量默认是256MB 2）每个任务最大的reduce数，默认为1009 3）计算reducer数的公式 2、调整reduce个数方法二 3、reduce个数并不是越多越好 过多的启动和初始化reduce也会消耗时间和资源； 同时过多的reduce会生成很多个文件，也有可能出现小文件问题 总的来说就是，数据倾斜的根源是key分布不均匀，所以应对方案要么是从源头解决（不让数据分区，直接在map端搞定），要么就是在分区时将这些集中却无效的key过滤（清洗）掉，或者是想办法将这些key打乱（给key加上标签），让它们进入到不同的reduce中。 order by sort by distribute by distribute by length(name); Cluster By 我们发现其实桶的概念就是MapReduce的分区的概念，两者完全相同。物理上每个桶就是目录里的一个文件，一个作业产生的桶（输出文件）数量和reduce任务个数相同。 而分区表的概念，则是新的概念。分区代表了数据的仓库，也就是文件夹目录。每个文件夹下面可以放不同的数据文件。通过文件夹可以查询里面存放的文件。但文件夹本身和数据的内容毫无关系。 桶则是按照数据内容的某个值进行分桶，把一个大文件散列称为一个个小文件。 这些小文件可以单独排序。如果另外一个表也按照同样的规则分成了一个个小文件。两个表join的时候，就不必要扫描整个表，只需要匹配相同分桶的数据即可。效率当然大大提升。 同样，对数据抽样的时候，也不需要扫描整个文件。只需要对每个分区按照相同规则抽取一部分数 据即可。 • 分区表 1.hive如果有过多的分区，由于底层是存储在HDFS上，HDFS上只用于存储大文件 而非小文件，因为过多的分区会增加namenode的负担。 2.hive会转化为mapreduce，mapreduce会转化为多个task。过多小文件的话，每个文件一个task，每个task一个JVM实例，JVM的开启与销毁会降低系统效率。 注意：合理的分区不应该有过多的分区和文件目录，并且每个目录下的文件应该足够大 hive cli 中设置调优参数 压缩模式评价 常见压缩格式 https://google.github.io/snappy/ On a single core of a Core i7 processor in 64-bit mode, Snappy compresses at about 250 MB/sec or more and decompresses at about 500 MB/sec or more. 要在Hadoop中启用压缩，可以配置如下参数（mapred-site.xml文件中）： Hive支持的存储数的格式主要有：TEXTFILE（行式存储） 、SEQUENCEFILE(行式存储)、ORC（列式存储）、PARQUET（列式存储）。 上图左边为逻辑表，右边第一个为行式存储，第二个为列式存储。 行存储的特点： 查询满足条件的一整行数据的时候，列存储则需要去每个聚集的字段找到对应的每个列的值，行存储只需要找到其中一个值，其余的值都在相邻地方，所以此时行存储查询的速度更快。select * 列存储的特点： 因为每个字段的数据聚集存储，在查询只需要少数几个字段的时候，能大大减少读取的数据量；每个字段的数据类型一定是相同的，列式存储可以针对性的设计更好的设计压缩算法。 select 某些字段效率更高 TEXTFILE和SEQUENCEFILE的存储格式都是基于行存储的； ORC和PARQUET是基于列式存储的。 默认格式，数据不做压缩，磁盘开销大，数据解析开销大。可结合Gzip、Bzip2使用(系统自动检查，执行查询时自动解压)，但使用这种方式，hive不会对数据进行切分，从而无法对数据进行并行操作。 Orc (Optimized Row Columnar)是hive 0.11版里引入的新的存储格式。 可以看到每个Orc文件由1个或多个stripe组成，每个stripe250MB大小，这个Stripe实际相当于RowGroup概念，不过大小由4MB->250MB，这样能提升顺序读的吞吐率。每个Stripe里有三部分组成，分别是Index Data,Row Data,Stripe Footer： 一个orc文件可以分为若干个Stripe 一个stripe可以分为三个部分 indexData：某些列的索引数据 rowData :真正的数据存储 StripFooter：stripe的元数据信息 1）Index Data：一个轻量级的index，默认是每隔1W行做一个索引。这里做的索引只是记录某行的各字段在Row Data中的offset。 2）Row Data：存的是具体的数据，先取部分行，然后对这些行按列进行存储。对每个列进行了编码，分成多个Stream来存储。 3）Stripe Footer：存的是各个stripe的元数据信息 每个文件有一个File Footer，这里面存的是每个Stripe的行数，每个Column的数据类型信息等；每个文件的尾部是一个PostScript，这里面记录了整个文件的压缩类型以及FileFooter的长度信息等。在读取文件时，会seek到文件尾部读PostScript，从里面解析到File Footer长度，再读FileFooter，从里面解析到各个Stripe信息，再读各个Stripe，即从后往前读。 Parquet是面向分析型业务的列式存储格式，由Twitter和Cloudera合作开发，2015年5月从Apache的孵化器里毕业成为Apache顶级项目。 Parquet文件是以二进制方式存储的，所以是不可以直接读取的，文件中包括该文件的数据和元数据，因此Parquet格式文件是自解析的。 通常情况下，在存储Parquet数据的时候会按照Block大小设置行组的大小，由于一般情况下每一个Mapper任务处理数据的最小单位是一个Block，这样可以把每一个行组由一个Mapper任务处理，增大任务执行并行度。Parquet文件的格式如下图所示。 上图展示了一个Parquet文件的内容，一个文件中可以存储多个行组，文件的首位都是该文件的Magic Code，用于校验它是否是一个Parquet文件，Footer length记录了文件元数据的大小，通过该值和文件长度可以计算出元数据的偏移量，文件的元数据中包括每一个行组的元数据信息和该文件存储数据的Schema信息。除了文件中每一个行组的元数据，每一页的开始都会存储该页的元数据，在Parquet中，有三种类型的页：数据页、字典页和索引页。数据页用于存储当前行组中该列的值，字典页存储该列值的编码字典，每一个列块中最多包含一个字典页，索引页用来存储当前行组下该列的索引，目前Parquet中还不支持索引页。 存储文件的查询速度测试： map,filter,flatmap,groupbykey,repartition reduce,collect,count,take,saveAsTextFile rdd即弹性分布式数据集，是spark中最基本的数据抽象，它代表一个不可变的、可分区的内部元素可并行计算 的集合。 使用makeRDD通过集合创建。由本地核数来决定分区数量 使用外部数据源创建如hdfs。由block的数量来决定的，通常默认为2个分区最低也是2个。 由另一个rdd得出的结果创建，即转换时创建。根据父rdd的 reduceTask数量 简单的说，有效分别各个算子之间的关系有利于生成dag图形，以及程序运行过程中产生的多次依赖变化的监察，窄依赖就是将依赖较为单一依赖视为一种方式，如果将宽窄依赖混为一谈，处理、区分、运行都会导致效率的分配不均。 即 stage 下的一个任务执行单元，一般来说，一个 rdd 有多少个 partition，就 会有多少个 task，因为每一个 task 只是处理一个partition 上的数据。 一个Job会被拆分为多组Task，每组任务被称为一个stage，每一次数据的shuffle都会产生一个stage。 每触发一次action操作就会生成一个job， 由于划分完 stage 之后，在同一个stage中只有窄依赖，没有宽依赖，可以实现流水线计算，stage中的每一个分区对应一个task，在同一个stage中就有很多可以并行运行的task。 提交任务的脚本: 设置task的数量 spark.defalut.parallelism 默认是没有值的，如果设置了值为10，它会在shuffle的过程才会起作用。 比如 val rdd2 = rdd1.reduceByKey(+) 此时rdd2的分区数就是10 可以通过在构建SparkConf对象的时候设置，例如： 给RDD重新设置partition的数量 使用rdd.repartition 来重新分区，该方法会生成一个新的rdd，使其分区数 变大。 此时由于一个partition对应一个task，那么对应的task个数越多，通过这 种方式也可以提高并行度。 通过设置参数sql.shuffle.partitions=500 默认为200； 可以适当增大，来提高并行度。 比如设置为 spark.sql.shuffle.partitions=500 持久化的双副本机制，持久化后的一个副本，因为机器宕机了，副本丢了，就还是得重新计算一次； 持久化的每个数据单元，存储一份副本，放在其他节点上面，从而进行容错；一个副本丢了，不用重新计算，还可以使用另外一份副本。这种方式，仅仅针对你的内存资源极度充足。 比如: StorageLevel.MEMORY_ONLY_2 比如一个任务需要50个executor，1000个task，共享数据为100M。 总结： 不使用广播变量的内存开销为100G，使用后的内存开销5G，这里就相差了20 倍左右的网络传输性能损耗和内存开销，使用广播变量后对于性能的提升和影响，还是很可观的。 广播变量的使用不一定会对性能产生决定性的作用。比如运行30分钟的spark作业，可能做了广播变量以后，速度快了2分钟，或者5分钟。但是一点一滴的调优，积少成多。最后还是会有效果的。 注意事项： 有效的避免shuffle可以减少网络间io和各个分区之间传输。 如何避免? 案例: 如果因为业务需要，一定要使用shuffle操作，无法用map类的算子来替代，那么尽量使用可以map-side预聚合的算子。 所谓的map-side预聚合，说的是在每个节点本地对相同的key进行一次聚合操作，类似于MapReduce中的本地combiner。 map-side预聚合之后，每个节点本地就只会有一条相同的key，因为多条相同的key都被聚合起来了。其他节点在拉取所有节点上的相同key时，就会大大减少需 要拉取的数据数量，从而也就减少了磁盘IO以及网络传输开销。 通常来说，在可能的情况下，建议使用reduceByKey或者aggregateByKey算子来 替代掉groupByKey算子。因为reduceByKey和aggregateByKey算子都会使用用户 自定义的函数对每个节点本地的相同key进行预聚合。 而groupByKey算子是不会进行预聚合的，全量的数据会在集群的各个节点之间分 发和传输，性能相对来说比较差。 比如如下两幅图，就是典型的例子，分别基于reduceByKey和groupByKey进行单 词计数。其中第一张图是groupByKey的原理图，可以看到，没有进行任何本地聚 合时，所有数据都会在集群节点之间传输；第二张图是reduceByKey的原理图，可以看到，每个节点本地的相同key数据，都进行了预聚合，然后才传输到其他节点上进行全局聚合。 groupByKey进行单词计数原理： reduceByKey单词计数原理： reduceByKey/aggregateByKey 可以进行预聚合操作，减少数据的传输量，提升性能 groupByKey 不会进行预聚合操作，进行数据的全量拉取，性能比较低 mapPartitions类的算子，一次函数调用会处理一个partition所有的数据，而不 是一次函数调用处理一条，性能相对来说会高一些。 但是有的时候，使用mapPartitions会出现OOM（内存溢出）的问题。因为单次函 数调用就要处理掉一个partition所有的数据，如果内存不够，垃圾回收时是无法 回收掉太多对象的，很可能出现OOM异常。所以使用这类操作时要慎重！ 原理类似于“使用mapPartitions替代map”，也是一次函数调用处理一个partition的所有数据，而不是一次函数调用处理一条数据。在实践中发现，foreachPartitions类的算子，对性能的提升还是很有帮助的。比 如在foreach函数中，将RDD中所有数据写MySQL，那么如果是普通的foreach算子，就会一条数据一条数据地写，每次函数调用可能就会创建一个数据库连接，此时就势必会频繁地创建和销毁数据库连接，性能是非常低下；但是如果用foreachPartitions算子一次性处理一个partition的数据，那么对于每个partition，只要创建一个数据库连接即可，然后执行批量插入操作，此时性能是比较高的。实践中发现，对于1万条左右的数据量写MySQL，性能可以提升30%以上。 通常对一个RDD执行filter算子过滤掉RDD中较多数据后（比如30%以上的数据），建议使用coalesce算子，手动减少RDD的partition数量，将RDD中的数据压缩到更少的partition中去。 因为filter之后，RDD的每个partition中都会有很多数据被过滤掉，此时如果照常进行后续的计算，其实每个task处理的partition中的数据量并不是很多，有一点资源浪费，而且此时处理的task越多，可能速度反而越慢。 因此用coalesce减少partition数量，将RDD中的数据压缩到更少的partition 之后，只要使用更少的task即可处理完所有的partition。在某些场景下，对于 性能的提升会有一定的帮助。 repartitionAndSortWithinPartitions是Spark官网推荐的一个算子，官方建议，如果需 要在repartition重分区之后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子。 因为该算子可以一边进行重分区的shuffle操作，一边进行排序。shuffle与sort两个操作同时进行，比先shuffle再sort来说，性能可能是要高的。 fastutil是扩展了Java标准集合框架（Map、List、Set；HashMap、ArrayList、 HashSet）的类库，提供了特殊类型的map、set、list和queue； fastutil能够提供更小的内存占用，更快的存取速度；我们使用fastutil提供的集合类，来替代自己平时使用的JDK的原生的Map、List、Set. Spark中应用fastutil的场景和使用 算子函数使用了外部变量 1.你可以使用Broadcast广播变量优化； 2.可以使用Kryo序列化类库，提升序列化性能和效率； 3.如果外部变量是某种比较大的集合，那么可以考虑使用fastutil改写外部 变量； 首先从源头上就减少内存的占用(fastutil)，通过广播变量进一步减少内存占用，再通过Kryo序列化类库进一步减少内存占用。 算子函数里使用了比较大的集合Map/List 在你的算子函数里，也就是task要执行的计算逻辑里面，如果有逻辑中，出现，要创建比较大的Map、List等集合， 可能会占用较大的内存空间，而且可能涉及到消耗性能的遍历、存取等集合操作；那么此时，可以考虑将这些集合类型使用fastutil类库重写，使用了fastutil集合类以后，就可以在一定程度上，减少task创建出来的集合类型的内存占用。 避免executor内存频繁占满，频繁唤起GC，导致性能下降。 fastutil的使用 第一步：在pom.xml中引用fastutil的包 第二步：平时使用List （Integer）的替换成IntList即可。 List的list对应的到fastutil就是IntList类型 使用说明： 基本都是类似于IntList的格式，前缀就是集合的元素类型； 特殊的就是Map，Int2IntMap，代表了key-value映射的元素类型。 ——————————————-end—————————————————-
Hadoop 常见面试题
1.mr 工作原理
2. split 机制
public static final String SPLIT_MAXSIZE = "mapreduce.input.fileinputformat.split.maxsize"; public static final String SPLIT_MINSIZE = "mapreduce.input.fileinputformat.split.minsize"; protected long computeSplitSize(long blockSize, long minSize, long maxSize) { return Math.max(minSize, Math.min(maxSize, blockSize)); } 3. namenode,datanode,secondaryNameNode分别是干什么的？
client 执行创建,移动,修改文件的信息，同时体现了 HDFS 的最新的状态(二进制文件)。edits 及后缀组成.后缀值是该文件包含的事务 ID,同一时刻只有一个文件处于可读写状态.为避免数据丢失,事务完成后 client 端在执行成功前,文件会进行更新和同步,当 NN 向多个目录写数据时,只有在所有操作更新并同步到每个副本之后执行才成功。edits 文件记录了最新状态信息，并且随着操作越多，edits 文件就会越大，把 edits 文件中最新的信息写到 fsimage 文件中就解决了 edits 文件数量多不方便管理的情况。HDFS 的最新状态。fsimage 文件都是文件系统元数据的一个完整的永久性的检查点。
fsimage 和 edits 才会进行合并，得到一个新的 fsimage 文件，但是在实际生产环境中很少会重启集群，NN 的重启需要花费很长时间,因为会有很多改动需要合并到 fsimage 文件上，如果 NN 挂掉，fsimage 文件没有更新内容，从而丢失很多改动。4. mr on yarn 工作原理
5.fsimage 和 edits 是干什么的？为什么要使用？
client 执行创建,移动,修改文件的信息，同时体现了 HDFS 的最新的状态(二进制文件)。edits 及后缀组成.后缀值是该文件包含的事务 ID,同一时刻只有一个文件处于可读写状态.为避免数据丢失,事务完成后 client 端在执行成功前,文件会进行更新和同步,当 NN 向多个目录写数据时,只有在所有操作更新并同步到每个副本之后执行才成功。edits 文件记录了最新状态信息，并且随着操作越多，edits 文件就会越大，把 edits 文件中最新的信息写到 fsimage 文件中就解决了 edits 文件数量多不方便管理的情况。HDFS 的最新状态。fsimage 文件都是文件系统元数据的一个完整的永久性的检查点。NN 使用了 FsImage+EditLog 整合的方案；EditLog 更新到 FsImage，以保证更近时点的 FsImage 和更小的 EditLog 体积6. hdfs 工作原理
block 传输完成，DN 们各自向 NN 汇报，同时 Client 继续传输下一个 block所以，Client 的传输和 block 的汇报也是并行的7. block 副本放置策略
DataNode；如果时集群外提交，则随机挑选一台磁盘不太满，CPU 不太忙的节点。
DataNode；如果时集群外提交，则随机挑选一台磁盘不太满，CPU 不太忙的节点。Hive 常见面试题
1.简述Hive工作原理
2.hive 内部表和外部表区别
3.分区和分桶的区别
分桶是相对分区进行更细粒度的划分。4.将数据直接上传到分区目录（hdfs）上，让分区表和数据产生关联有哪些方式？
dfs -mkdir -p 分区目录
dfs -put 分区目录
msck repair table 表名
方案二：上传数据后添加分区
dfs -mkdir -p 分区目录
dfs -put 分区目录
alter table 表名 add partition();5.桶表是否可以通过直接load将数据导入？
6.order by,sort by,distribute by,cluster by的区别？
7.聚合函数是否可以写在order by后面，为什么？
(1)from (2)join (3)on (4)where (5)select (6)group by (7)having (8)order by (9)limit 8.导致数据倾斜的原因有哪些，有什么解决的方案？
什么是数据倾斜？
跑mr任务时常见的reduce的进度总是卡在99%，这种现象很大可能就是数据倾斜造成的。造成数据倾斜的原因
还有其他的一些原因，但是，根本原因还是key的分布不均匀，而其他的原因就是会造成key不均匀，进而导致数据倾斜的后果，所以说根本原因是key的分布不均匀。
简单地说数据倾斜这种现象导致的任务迟迟不能完成，耗费了太多时间，极大地影响了性能，所以我们数据倾斜的解决方案设计思路就是往如何提高性能，即如何缩短任务的处理时间这方面考虑的，而要提高性能，就要让key分布相对均衡，所以我们的终极目标就是考虑如何预处理数据才能够使得它的key分布均匀。解决办法
1 合理设置Map数
主要的决定因素有：input的文件总个数，input的文件大小，集群设置的文件块大小。 举例： a) 假设input目录下有1个文件a，大小为780M，那么hadoop会将该文件a分隔成7个块（6个128m的块和1个12m的块），从而产生7个map数。 b) 假设input目录下有3个文件a，b，c大小分别为10m，20m，150m，那么hadoop会分隔成4个块（10m，20m，128m，22m），从而产生4个map数。即，如果文件大于块大小(128m)，那么会拆分，如果小于块大小，则把该文件当成一个块。 答案是否定的。如果一个任务有很多小文件（远远小于块大小128m），则每个小文件也会被当做一个块，用一个map任务来完成，而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的map数是受限的。 答案也是不一定。比如有一个127m的文件，正常会用一个map去完成，但这个文件只有一个或者两个小字段，却有几千万的记录，如果map处理的逻辑比较复杂，用一个map任务去做，肯定也比较耗时。 针对上面的问题2和3，我们需要采取两种方式来解决：即减少map数和增加map数； 2 小文件合并
set mapred.max.split.size=112345600; set mapred.min.split.size.per.node=112345600; set mapred.min.split.size.per.rack=112345600; set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 3 复杂文件增加Map数
mapreduce.input.fileinputformat.split.minsize=1 默认值为1 mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue 默认值Long.MAXValue因此，默认情况下，切片大小=blocksize maxsize（切片最大值): 参数如果调到比blocksize小，则会让切片变小，而且就等于配置的这个参数的值。 minsize(切片最小值): 参数调的比blockSize大，则可以让切片变得比blocksize还大。
--设置maxsize大小为10M，也就是说一个fileSplit的大小为10M set mapreduce.input.fileinputformat.split.maxsize=10485760; 4 合理设置Reduce数
set hive.exec.reducers.bytes.per.reducer=256000000; set hive.exec.reducers.max=1009; N=min(参数2，总输入数据量/参数1) --设置每一个job中reduce个数 set mapreduce.job.reduces=3;
9. Hive的四种排序
order by 会对输入做全局排序，因此只有一个reducer（多个reducer无法保证全局有序）
只有一个reducer，会导致当输入规模比较大时，需要较长的时间。
set hive.mapred.mode=nonstrict; (default value / 默认值)
set hive.mapred.mode=strict;
order by 和数据库中的order by功能一致按照某一项&几项排序输出。
与数据库中order by的区别在于hive.mapred.mode = strict模式下 必须指定limit否则执行会报错
原因：在order by状态下所有的数据都会到一台服务器进行reduce操作也就是只有一个reduce， 如果在数据量大的情况下会出现无果的情况，如果进行limit n，那只有n * map
number 条记录而已。只有一个reduce也可以出来里过来
sort by不是全局排序，其在数据进入reducer前完成排序
因此，如果用sort by进行排序，并且设置mapred.reduce.tasks>1，则sort by只保证每个reducer 的输出有序，不保证全局有序。
sort by 不受hive.mapred.mode是否为strict，nostrict的影响。
sort by的数据只能保证在同一个reduce中的数据可以按指定字段排序。
使用sort by你可以指定执行的reduce个数（set mapred.reduce.tasks=），对输出的 数据在执行归并排序，即可以得到全部结果。
注意：可以用limit子句大大减少数据量。使用limit n后，传输到reduce端（单机）的数据记录就 减少到n*（map个数）。否则由于数据过大可能出不了结果。
按照指定的字段对数据进行划分到不同的输出reduce / 文件中。
insert overwrite local directory ‘/home/hadoop/out’ select * from test order by name
此方法会根据name的长度划分到不同的reduce中，最终输出到不同的文件中。
length 是内建函数，也可以指定其他的函数或这使用自定义函数。
cluster by 除了具有 distribute by 的功能外还兼具 sort by 的功能。
但是排序只能是倒序排序，不能指定排序规则为asc 或者desc。10. Hive的分区分桶
如果在建表时使用了 PARTITIONED BY，表即为分区表。分区表下的数据按分区键的值（或值的范围）放在HDFS下的不同目录中，可以有效减少查询时扫描的数据量，提升查询效率。
• 非分区表
非分区表即除分区表之外的表。
按表是否分桶分类
按表是否分桶可以将表分为两类：分桶表和非分桶表。
• 分桶表
如果在建表时使用了 CLUSTERED BY … INTO … BUCKETS，表即为分桶表。分桶表下的数据按
分桶键的哈希值放在HDFS下的不同目录中，可以有效减少查询时扫描的数据量，提升查询效率。
• 非分桶表
非分桶表即除分桶表之外的表11. hive中分区可以提高查询效率，分区是否越多越好，为什么？
12. hive 调优
12.1 hive-site.xml参考
<configuration> <property> <name>javax.jdo.option.ConnectionURL</name> <value>jdbc:mysql://bd01:3306/hive?createDatabaseIfNotExist=true</value> <description>JDBC connect string for a JDBC metastore</description> </property> <!--开启 spark 引擎--> <property> <name>hive.execution.engine</name> <value>spark</value> </property> <!-- 合并 block 减少 task 数量--> <property> <name>ngmr.partition.automerge</name> <value>true</value> </property> <!-- 表示将 n 个 block 安排给单个线程处理。 --> <property> <name>ngmr.partition.mergesize.mb</name> <value>3</value> </property> <!-- 开启小文件合并--> <property> <name>hive.merge.sparkfiles</name> <value>true</value> </property> <!-- 开启小文件合并--> <property> <name>hive.map.agg</name> <value>true</value> </property> <!-- 使用向量化查询--> <property> <name>hive.vectorized.execution.enabled</name> <value>true</value> </property> <!-- cbo可以优化hive的每次查询--> <property> <name>hive.cbo.enable</name> <value>true</value> </property> <property> <name>hive.stats.fetch.column.stats</name> <value>true</value> </property> <property> <name>hive.stats.fetch.partition.stats</name> <value>true</value> </property> <property> <name>hive.compute.query.using.stats</name> <value>true</value> </property> <!-- 开启数据压缩--> <property> <name>hive.exec.compress.intermediate</name> <value>true</value> </property> <property> <name>hive.exec.compress.output</name> <value>true</value> </property> <!-- 使简单的sql语句不执行spark引擎--> <property> <name>hive.fetch.task.conversion</name> <value>more</value> </property> <!-- 有数据倾斜的时候进行负载均衡 group by操作是否允许数据倾斜，默认是false，当设置为true时，执行计划会生成两个map/reduce作业，第一个MR中会将map的结果随机分布到reduce中，达到负载均衡的目的来解决数据倾斜， --> <property> <name>hive.groupby.skewindata</name> <value>true</value> </property> <!-- 列裁剪，默认开启true，在做查询时只读取用到的列，这个是个有用的优化；--> <property> <name>hive.optimize.cp</name> <value>true</value> </property> <!--jvm重用--> <property> <name>mapreduce.job.jvm.numtasks</name> <value>10</value> <description>How many tasks to run per jvm. If set to -1, there is no limit.</description> </property> <property> <name>javax.jdo.option.ConnectionDriverName</name> <value>com.mysql.jdbc.Driver</value> <description>Driver class name for a JDBC metastore</description> </property> <property> <name>javax.jdo.option.ConnectionUserName</name> <value>root</value> <description>username to use against metastore database</description> </property> <property> <name>javax.jdo.option.ConnectionPassword</name> <value>root</value> <description>password to use against metastore database</description> </property> </configuration> // 合并 block 减少 task 数量 set ngmr.partition.automerge = true; // jvm 重用 set mapreduce.job.jvm.numtasks=10; // 表示将 n 个 block 安排给单个线程处理。 set ngmr.partition.mergesize.mb =3; // 开启小文件合并 set hive.merge.sparkfiles = true; // 开启小文件合并 set hive.map.agg = true; // 使用向量化查询 set hive.vectorized.execution.enabled = true; // cbo可以优化hive的每次查询 set hive.cbo.enable = true; set hive.stats.fetch.column.stats = true; set hive.stats.fetch.partition.stats = true; set hive.compute.query.using.stats = true; // 开启数据压缩 set hive.exec.compress.intermediate = true; set hive.exec.compress.output = true; // 有数据倾斜的时候进行负载均衡group by操作是否允许数据倾斜，默认是false，当设置为true时，执行计划会生成两个map/reduce作业，第一个MR中会将map的结果随机分布到reduce中，达到负载均衡的目的来解决数据倾斜， set hive.groupby.skewindata = true; // 列裁剪，默认开启true，在做查询时只读取用到的列，这个是个有用的优化； set hive.optimize.cp = true; 12.2 数据压缩
1、数据的压缩说明
压缩方式
压缩比
压缩速度
解压缩速度
是否可分割
gzip
13.4%
21 MB/s
118 MB/s
否
bzip2
13.2%
2.4MB/s
9.5MB/s
是
lzo
20.5%
135 MB/s
410 MB/s
是
snappy
22.2%
172 MB/s
409 MB/s
否
压缩格式
对应的编码/解码器
DEFLATE
org.apache.hadoop.io.compress.DefaultCodec
Gzip
org.apache.hadoop.io.compress.GzipCodec
BZip2
org.apache.hadoop.io.compress.BZip2Codec
LZO
com.hadoop.compress.lzo.LzopCodec
Snappy
org.apache.hadoop.io.compress.SnappyCodec
压缩性能的比较
压缩算法
原始文件大小
压缩文件大小
压缩速度
解压速度
gzip
8.3GB
1.8GB
17.5MB/s
58MB/s
bzip2
8.3GB
1.1GB
2.4MB/s
9.5MB/s
LZO
8.3GB
2.9GB
49.3MB/s
74.6MB/s
2、压缩配置参数
参数
默认值
阶段
建议
io.compression.codecs （在core-site.xml中配置）
org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec, org.apache.hadoop.io.compress.Lz4Codec
输入压缩
Hadoop使用文件扩展名判断是否支持某种编解码器
mapreduce.map.output.compress
false
mapper输出
这个参数设为true启用压缩
mapreduce.map.output.compress.codec
org.apache.hadoop.io.compress.DefaultCodec
mapper输出
使用LZO、LZ4或snappy编解码器在此阶段压缩数据
mapreduce.output.fileoutputformat.compress
false
reducer输出
这个参数设为true启用压缩
mapreduce.output.fileoutputformat.compress.codec
org.apache.hadoop.io.compress. DefaultCodec
reducer输出
使用标准工具或者编解码器，如gzip和bzip2
mapreduce.output.fileoutputformat.compress.type
RECORD
12.3 文件压缩
1、列式存储和行式存储
2 、TEXTFILE格式
3 、ORC格式
4 、PARQUET格式
1）TextFile hive (default)> select count(*) from log_text; _c0 100000 Time taken: 21.54 seconds, Fetched: 1 row(s) 2）ORC hive (default)> select count(*) from log_orc; _c0 100000 Time taken: 20.867 seconds, Fetched: 1 row(s) 3）Parquet hive (default)> select count(*) from log_parquet; _c0 100000 Time taken: 22.922 seconds, Fetched: 1 row(s) 存储文件的查询速度总结： ORC > TextFile > Parquet Spark 常见面试题
1. 说出几个常见的rdd算子？
1.1 transformation算子
1.2 action 算子
2. rdd是什么？rdd的创建方式有几种？分别是怎么分区的？
2.1 rdd是什么？
2.2 rdd的创建方式
3. RDD的依赖关系
3.1 宽依赖：多个子RDD的分区依赖同一个父RDD的Partition
3.2 窄依赖：每一个父RDD的Partition最多被子RDD的 一个Partition使用
3.3 为什么要划分依赖关系？
4. task，stage，job分别是什么？
4.1 task
4.2 stage
4.3 job
4.4 为什么要划分 stage？
5. RDD的缓存持久化机制？
主要通过cache，persist，checkpoint来实现RDD 的缓存持久化机制
5.1 cache与persist：
5.2 checkpoint：
6. spark常见调优
6.1 修改序列化机制有效压缩数据量，通过使用Kryo优化序列化性能
conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer ") 6.2 在实际的生产环境中，提交spark任务时，使用spark-submit shell脚本，在里面调整对应的参数。
spark-submit --master spark://node1:7077 --class com.kaikeba.WordCount --num-executors 3 配置executor的数量 --driver-memory 1g 配置driver的内存（影响不大） --executor-memory 1g 配置每一个executor的内存大小 --executor-cores 3 配置每一个executor的cpu个数 /export/servers/wordcount.jar 6.3 提高并行度
new SparkConf().set("spark.defalut.parallelism","500") 6.4 RDD 的重用和持久化
6.5 适当将那些多次使用变量广播出去
1.通过sparkContext的broadcast方法把数据转换成广播变量，类型为Broadcast， val broadcastArray: Broadcast[Array[Int]] = sc.broadcast(Array(1,2,3,4,5,6)) sc.broadcast(Array(1,2,3,4,5,6)) 2.然后executor上的BlockManager就可以拉取该广播变量的副本获取具体的数据。 获取广播变量中的值可以通过调用其value方法 val array: Array[Int] = broadcastArray.value
6.6 尽量避免shuffle
//错误的做法： // 传统的join操作会导致shuffle操作。 // 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。 val rdd3 = rdd1.join(rdd2) //正确的做法： // Broadcast+map的join操作，不会导致shuffle操作。 // 使用Broadcast将一个数据量较小的RDD作为广播变量。 val rdd2Data = rdd2.collect() val rdd2DataBroadcast = sc.broadcast(rdd2Data) // 在rdd1.map算子中，可以从rdd2DataBroadcast中，获取rdd2的所有数据。 // 然后进行遍历，如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，那么就判定可以进行join。 // 此时就可以根据自己需要的方式，将rdd1当前数据与rdd2中可以连接的数据，拼接在一起（String或Tuple）。 val rdd3 = rdd1.map(rdd2DataBroadcast...) // 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。 // 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。 6.7 使用map-side预聚合的shuffle操作
6.8 使用高性能的算子
6.8.1 使用reduceByKey/aggregateByKey替代groupByKey
6.8.2 使用mapPartitions替代普通map
6.8.3 使用foreachPartitions替代foreach
6.8.4 使用filter之后进行coalesce操作
6.8.5 使用repartitionAndSortWithinPartitions替代repartition与sort类操作
6.8.6 使用fastutil优化数据格式
<dependency> <groupId>fastutil</groupId> <artifactId>fastutil</artifactId> <version>5.0.9</version> </dependency>
reduce,collect,count,take,saveAsTextFile
文章部分引用
开课吧大数据开发工程师课程如侵权请联系删除。
https://mkt.kaikeba.com/vipcourse/bde
本网页所有视频内容由 imoviebox边看边下-网页视频下载, iurlBox网页地址收藏管理器 下载并得到。
ImovieBox网页视频下载器 下载地址: ImovieBox网页视频下载器-最新版本下载
本文章由: imapbox邮箱云存储,邮箱网盘,ImageBox 图片批量下载器,网页图片批量下载专家,网页图片批量下载器,获取到文章图片,imoviebox网页视频批量下载器,下载视频内容,为您提供.
阅读和此文章类似的: 全球云计算
官方软件产品操作指南 (170)
