这些都是一些笔者不太熟悉的面试题,有问题欢迎评论区留言!看到就会回复,并把题目整理到文章里!

文章目录


1、MapReduce如何选择垃圾回收器?

分别对map任务和reduce任务进行minor GC。
发现map任务的老年代使用内存不会上升,但是Eden代的使用内存明显下降,这说明map任务中的对象大多数都是临时对象,可以使用Serial、ParNew、Parallel Scavenge来进行并发的垃圾回收。这两个垃圾回收器用的都是复制-清除算法,适用于新生代的回收。
reduce任务中,发现新生代的内存使用减少,老年代的内存使用明显上升,这里可以判断reduce任务中,存在较多长期存活的对象(或者大对象)。因此针对reduce任务,则主要对老年代进行垃圾回收,这里可以选择G1或者CMS进行并发的回收。G1的好处是回收可以满足用户期望时间,CMS则是吞追求最短用户停顿
至于这些对象到底是什么,我猜测是内存中的缓存数据对象。
CMS:基于标记-清除算法实现。并发收集、低停顿。
G1:
并行与并发:G1能充分利用多CPU、多核环境下的硬件优势,使用多个CPU来缩短Stop-The-World停顿时间。部分收集器原本需要停顿Java线程来执行GC动作,G1收集器仍然可以通过并发的方式让Java程序继续运行。
分代收集:G1能够独自管理整个Java堆,并且采用不同的方式去处理新创建的对象和已经存活了一段时间、熬过多次GC的旧对象以获取更好的收集效果。
空间整合:G1运作期间不会产生空间碎片,收集后能提供规整的可用内存。
可预测的停顿:G1除了追求低停顿外,还能建立可预测的停顿时间模型。能让使用者明确指定在一个长度为M毫秒的时间段内,消耗在垃圾收集上的时间不得超过N毫秒。


2、如何配置hdfs集群?

  1. 配置SSH免密登录,非对称加密:发送登录请求,服务器收到请求后发送公钥,客户端根据公钥对密码进行加密后传输到服务器。服务器使用私钥进行解密再进行身份验证。
  2. 配置jdk和hadoop环境变量
  3. 修改hdfs-site.xml文件,关键字段有集群节点名称,端口号等
  4. 修改core-site.xml文件,配置zk信息。
  5. 修改slaves文件
  6. 统一分发配置信息
  7. 格式化NameNode,接下来就是正常启动流程,如创建镜像文件、编辑日志、DataNode注册等等。

3、如何搭建yarn集群?

  1. 修改mapred-site.xml中的框架模式为yarn
  2. 在yarn-site中配置RM的节点以及配置ZK信息

4、hdfs存储结构?

  • NameNode存储数据元信息、管理命名空间、块映射、副本机制以及处理客户端请求(块映射信息存储在一个map中,key为blockInfo,value为blockInfo以及DataNode信息,如ip地址等)
  • DataNode存储具体信息,执行来自客户端的读写请求以及向NameNode注册并上传块信
  • Client进行文件块划分、发起读写请求以及通过命令管理hdfs
  • SecondaryNameNode协助NameNode完成工作,比如合并镜像文件和编辑日志以及座位NameNode的冷备份

5、hdfs的常见存储格式?

存储格式:行式存储以及列式存储

  • 行式存储,适合DQL操作,但是选择时即使只涉及部分列,也会将所有列数据全部读取一遍。
  • 列式存储,读取快,只需要按顺序读取即可

文件格式:TextFile、SequenceFile、Avro、Parquet、RC & ORC

  • SequenceFile:占地小,节省磁盘空间;本身是为了MR的k、v对设计。二进制文件,可读性低。
  • TextFile:常采用csv、json等固定长度的纯文本格式,可读性高但不支持块压缩
    具体可看hdfs存储格式和文件格式

6、hdfs小文件的危害以及如何处理?

危害:namenode在内存中存储文件的元信息,如果小文件过多就会导致内存中的元信息过多,假设一个元信息需要固定的150字节去存储,那么易得文件越大,namenode存储的元信息性价比越高。太多的小文件容易压垮namenode的内存。
解决方法:两个层面

  • MR层面
    • 最简单的方法就是减少map任务或者reduce任务
    • 多少个Map就会产生多少个中间文件,多少个reduce就会产生多少个结果文件,可以适当的进行合并。可以写一个专门的MR进行文件合并。
    • 使用CombineFileInputFormat进行小文件合并。它会将小文件组成一个map任务的输入进行小文件合并
  • NM层面
    • Hadoop Archives (HAR files)是在 0.18.0 版本中引入到 HDFS 中的,它的出现就是为了缓解大量小文件消耗 NameNode 内存的问题。它将hdfs中的小文件合并成大文件,即打包成一个HAR文件,相当于一个目录。但是对于MR来说,治标不治本,HAR只是相当于一个目录。

7、数据倾斜如何处理?

这里分为spark数据倾斜和MR数据倾斜

  • MR数据倾斜是由于reduce之间的数据分布不均匀导致的,即某一个分区的数据量过大。比如有如下key:aaa,bbb,ccc。100g的数据中,aaa有80g,那么一个reduce中的数据就达到80g了,导致reduce运行缓慢。
    • 在环形缓冲区进行预聚合可以有效缓解数据倾斜(hive.map.aggr=true),这样的话map就需要更多内存。空间换时间
    • 额外增加一个MR任务进行聚合。和上面是一样的思路,第一个MR中,map将数据随机放入reduce进行聚合,聚合的结果作为下一个job的输入。(set hive.groupby.skewindata=true)但是这样会比较慢,因为中间结果文件增加。时间换空间
    • 优化key,map输出的时候,按序生成aaa1,aaa2,aaa3…的key,combine时候做聚合后,再恢复key。或者reduce先进行一次aaa1,aaa2的聚合,再在下个MR进行最终聚合。1、2的升级,主要是防止导致数据倾斜的key在不同的map上。
  • spark上的数据倾斜则是shuffle造成的,导致不同分区之间,某些分区的数据量特别大。表现为某些task特别慢
    • 增加分区数来打散数据。
    • 特殊情况特殊处理,通过group by+count分析是哪些key导致了数据倾斜,再对这些特殊key进行特殊处理。比如对key加前缀后缀等
    • 自定义分区。手动打散数据

8、Reduce Join 和Map join

reduce join的任务是将map阶段生成的数据根据key进行连接。
这样可能会有数据倾斜的问题(map端数据处理较少,reduce端处理数据较多),可以通过在map端实现合并,就是接下来将的Map join
Map join比较适合小表去join大表,将小表读取放入内存后去join。提高效率并减少reduce的操作量

9、MR的压缩

由于MR中设计到大量的网络IO,所以对数据进行压缩,可以减少IO数量,有效地提高MR的效率。
压缩使用于IO操作比较多的MR,计算较多的则不合适。因为IO较少的话,压缩和解压的时间可能和不采用压缩的时间相差不大,性价比低。

10、spark中repartition和coalesce的区别

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
  coalesce(numPartitions, shuffle = true)
}

从源码中可以看出,repartition本质上就是shuffle设为true的coalesce,代表一定会发生shuffle操作。
那么shuffle这个参数又在coalesce中做了什么?

def coalesce(numPartitions: Int, shuffle: Boolean = false,
             partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
            (implicit ord: Ordering[T] = null)
    : RDD[T] = withScope {
  require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")
  if (shuffle) {
    /** Distributes elements evenly across output partitions, starting from a random partition. */
    val distributePartition = (index: Int, items: Iterator[T]) => {
    //随机生成0-新分区数的一个数,作为新分区
      var position = new Random(hashing.byteswap32(index)).nextInt(numPartitions)
      items.map { t =>
        // Note that the hash code of the key will just be the key itself. The HashPartitioner
        // will mod it with the number of total partitions.
        position = position + 1
        (position, t)
      }
    } : Iterator[(Int, T)]

    // include a shuffle step so that our upstream tasks are still distributed
    new CoalescedRDD(
      new ShuffledRDD[Int, T, T](
        mapPartitionsWithIndexInternal(distributePartition, isOrderSensitive = true),
        new HashPartitioner(numPartitions)),
      numPartitions,
      partitionCoalescer).values
  } else {
    new CoalescedRDD(this, numPartitions, partitionCoalescer)
  }
}

如果shuffle为true的话,则产生0-新分区数的随机数作为当前数据的分区,返回一个经过shuffle操作的CoalescedRDD,如果为false的话,则直接返回一个CoalescedRDD,这里适用于新分区数小于旧分区数,即直接合并。
Coalesced一般用于大数据过滤后,减少小数据分区数,并不打乱。有可能数据倾斜。这时候就要用reparation了
reparation一般用于增大分区或者新分区数远小于旧分区数。

11、spark 四个byKey的区别

  • reduceByKey: 相同 key 的第一个数据,即初始值,不进行任何计算,分区内和分区间计算规则相同
  • FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则相
  • AggregateByKey:相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则可以不相同
  • CombineByKey:当计算时,发现数据结构不满足要求时,可以让第一个数据转换结构。分区内和分区间计算规则不相同。(就是将相同key里面的第一个数据作为初始值)

底层都是通过combineByKeyWithClassTag[V]实现的,第一个是初始值,第二个是分区内计算规则,第三个是分区间计算规则。

其他的RDD比较简单,就不多说了,有问题可以直接问。

12、flume如何监听文件夹下的新文件?

这里首先介绍source的种类

  • taildir source:1.7版本之后 taildir source可以做到监听多个实时追加文件,实现断点续传(flume传输数据时突然挂掉了,重启后从上一次挂掉的位置继续传输数据)
  • Exec source 适用于监控一个实时追加的文件,不能实现断点续传
  • Spooldir Source适合用于同步新文件,但不适合对实时追加日志的文件进行监听并同步;

对新文件进行监听,那么采用spooldir source即可。
tip:flume通过监听inode(文件唯一标识符)、绝对路径以及上次读取的节点来读取数据。

13、flume 如何保证数据不丢失?

  1. 对channel作持久化,比如放入文件,适用于不追求吞吐量的情况下
  2. 对于source,使用taildir source的断点续传功能
  3. 对于sink,使用事务。发送方的sink开启一个事务,接收方的source开启一个事务。两个事务(发送数据的事务和接收数据的事物)都成功提交,才算数据传输完成
  4. source和channel间以及channel和sink间的数据传输也通过事务进行保证数据不丢失。source中的数据通过doPut()方法传到putList中,通过doCommit判断数据是否可以放入channel(这里需要考虑takeList发生回滚的时候,channel的容量在加入putList放入的数据的情况下,可以回滚成功)。sink和channel之间也是同理。

14、spark算法 如何判断DAG?

拓扑排序。
思考:bfs和拓扑的区别?
bfs相邻即入队,拓扑入度为0才入队。

15、spark 任务全流程

16、spark shuffle

当前版本有三种spark shuffle,各种使用的场景,优势
之前版本还有hashShuffle 以及优化版本

17、spark RDD

RDD叫做弹性分布式数据集,是 Spark 中最基本的数据处理模型。代码中是一个抽象类,它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

➢ 弹性

⚫ 存储的弹性:内存与磁盘的自动切换;

⚫ 容错的弹性:数据丢失可以自动恢复;

⚫ 计算的弹性:计算出错重试机制;

⚫ 分片的弹性:可根据需要重新分片。

➢ 分布式:数据存储在大数据集群不同节点上

➢ 数据集:RDD 封装了计算逻辑,并不保存数据

➢ 数据抽象:RDD 是一个抽象类,需要子类具体实现

➢ 不可变:RDD 封装了计算逻辑,是不可以改变的,想要改变,只能产生新的 RDD,在新的 RDD 里面封装计算逻辑

➢ 可分区、并行计算

18、100w条数据的全排序

桶排序或者归并排序都可以。
这里展开讲一下桶排序,找到数据的最大最小值。比如1和100
那么可以在1-10,11-20…91-100划分10个桶,每个桶中进行排序,最后再合并。
可能会出现数据倾斜,解决方法是使用sample函数推算数据的一个大概分布,再对应的调整桶个数。

引申:找中位数,8个赛道64匹马找跑的最快的4匹马…

19、spark中的分区有哪几种?

哈希分区、范围分区、自定义分区
哈希分区即取hash值对分区数取模,范围分区则根据key的范围进行分区划分。

20、spark 水塘抽样算法

问题背景:在不知道文件总行数的情况下,如何从文件中随机的抽取一行?
水塘抽样算法

21、hdfs文件读写流程

22、hdfs启动流程

23、hadoop 架构

24、Zookeeper的同步过程

拜占庭建军问题
paxos算法
ZAB协议
CAP理论
事务回滚

25、Zookeeper的选举机制

26、kafka如何保证不丢数据?

从生产者,broker,消费者展开说。
消费者需要维护幂等性。offset不安全

消费者角度,下游消费者必须支持事务,才可以做到精确一次性事务
比如MySQL,将消息的消费和offset的提交在一个事务中进行。

27、spark节点通信问题

sparkContext开启sparkEnv,sparkEnv开启RPCEnv,RPCEnv开启NettyRPCEnv,支持AIO(Linux用epoll代替AIO操作)
架构:消息发送线程(发件箱)、消息接收线程(收件箱)、传输服务端、传输客户端。
消息由消息发送线程发到收件箱,传输客户端轮询收件箱,有消息就发送到目标节点的传输服务端。服务端将消息放入收件箱。
与SMTP和POP3异曲同工吧

28、scala闭包以及函数柯里化

如果一个函数,访问到了它的外部(局部)变量的值,那么这个函数和他所处的环境,称为闭包

柯里化就是将函数作为实参传入函数中。

def f(a: => Int, res: Int): Unit = {
      println(res)
    }

    //    这种内嵌的函数一般都是先执行外层,外层返回内层函数再调用,所以返回值需要是内层的函数
    //    逻辑如下   val  innerFunc =myWhile(n>0) 如果返回值不是函数的话,就不能传入代码块了
    //    注意,这里是=>Boolean 而不是 Boolean 上面也有答案了,没有加=》就是一个结果,就是代码块的结果,如果没有多次调用结果当然一样,多次调用就G
    def myWhile(b: => Boolean): (=> Unit) => Unit = {
      def loop(op: => Unit): Unit = {
        if (b) {
          op
          loop(op)
        }
      }
      loop _
    }
    def myWhile3(b: => Boolean): (=> Unit) => Unit = {
      def loop(op: => Unit): Unit = {
        if (b) {
          op
          myWhile3(b)(op)
        }
      }
      loop _
    }
    var n = 10
    println(n)
    var innerFunc = myWhile(n > 0)
    myWhile(n > 0) {
      println(n)
      n -= 1
    }
    n = 10
    println("======================")
    innerFunc {
      println(n)
      n -= 1
    }

    //    while (n>0){
    //      println(n)
    //      n-=1
    //    }
    //    柯里化 这个好懂
    def myWhile2(condition: => Boolean)(op: => Unit): Unit = {
      if (condition) {
        op
        myWhile2(condition)(op)
      }
    }

29、spark运行模式

单机模式、standlone、yarn模型
stardlone模式和yarn模式的区别就是master代替RM,word代替NM

集群模式和客户端模式就是driver所在节点不同。

30、spark和MR的主要区别

  1. 表达能力方面。MR中,只有Map和Reduce两种操作,表达能力欠缺;Spark中封装了丰富的算子,主要分为转换算子和行动算子。
  2. IO方面。MR将中间结果存储到磁盘中,存在较多的IO操作,性能较低,适合高时延环境下批处理计算的应用;Spark中主要将中间结果存储在内存中(RDD为弹性数据集,分布弹性是指数据的存储模式可以在内存和磁盘之间切换,内存不够了就会进行落盘。),性能较快,但是如果数据量过大容易内存溢出。
  3. 迭代式计算的性能方面。一个MR任务的表达能力是有限的,所以作业一般需要多个MR来完成,基于MR的迭代式计算会将中间结果存储在内存中。而Spark由于拥有灵活的算子以及阶段划分、任务划分机制,使得基于内存的迭代式计算更加灵活、效率更高。
  4. Spark粗粒度申请资源,MR细粒度申请资源。Spark提交任务之前会先去申请资源,申请到了再提交task任务;MR会先提交task,之后由task去申请资源。

31、当 Spark 涉及到数据库的操作时,如何减少 Spark 运行中的 数据库连接数?

使用foreachPartition代替foreach,

在foreachPartition内获取数据库的连接。

通过spark.sql得到数据之后,由于数据是懒加载的,没有遇到行动算子就不会去执行,那么这个sql也是如此。(RDD,DS,DF都是懒加载的)如果使用foreach,则是操作每一条数据,对每一条数据执行查询语句,那么连接数就是查询数。如果使用foreachPartition,则是一个分区一连接。
转换算子也是同理,除非你不使用行动算子去触发任务提交。(map替换为mapPartitions)

32、RPC和HTTP的区别

两者都是基于TCP协议

  1. 速度方面,RPC较HTTP更快
  2. RPC要求两边接口规范,语言相同、框架相同等;HTTP只需要满足rest规范即可。因此HTTP更为通用,可以跨平台。

为什么RPC比HTTP快?

  1. RPC的报文都是有效荷载(就是干货),HTTP报文则略显臃肿,由于需要跨平台,追求通用性,因此需要在请求头中放置较多信息。所以传输效率还是RPC高。
  2. 在HTTP1.1的时候,是对文本数据进行序列化,RPC对二进制数据进行序列化,RPC序列化更快。但是HTTP2.0也支持了二进制数据。
  3. RPC稳定长连接(TCP复用),HTTP1.1也支持长连接,但是要看服务端是否接收长连接,不稳定。

33、常用端口号

  1. 8080(Master):sparkwebUI的端口号,常用于任务监控,task执行时间就可以在这里看,通过task执行时间以及资源占用判断是否发生了数据倾斜
  2. 8088(yarn任务运行情况)
  3. 8020/9000 hdfs UI
  4. 18080 spark历史服务器端口

34、kafka脑裂

controller的选举机制是controller分别在zk上注册临时节点,谁先注册成功,谁就称为master。
由于zookeeper临时节点的有效性是通过session来判断的,如果controller中发生GC导致session timeout,那么zk会重新选举controller。这时候就会有两个controller存在了。
解决方法:1、重启过期controller;2、新的controller产生的时候,在zk中注册一个更大的epoch标识符,并同步给其他broker保存。这样旧的controller发的指令就失效了,因为epoch小。

35、kafka选举机制

每个broker上都有一个controller,这些controller会去辅助分区选举自己的leader。
那么这些controller怎么去选一个老大呢?先到先得,哪个controller先向zk注册,哪个就是老大。
之后如果要选举分区,controller根据ISR和AR去选举,其中AR是服务器上线顺序(比如有0,1,2上台服务器,1先上线,之后是0,2,那么AR为[1,0,2]),按照AR顺序去轮询ISR,轮询到了将这个follower设为leader。
之后controller将信息上传到zk中,即更新该分区的leader分区,然后其他controller会去zk中同步信息。
https://blog.csdn.net/yanshu2012/article/details/54894629

36、spark数据倾斜排查

37、kafka为什么不在ZK存储offset

在生产环境下,消费者需要频繁的读取与提交offset,这就意味着需要频繁的去连接zk,与zk进行交互。导致较大的网络IO,因此kafka0.9之后选择额外配置一个topic来存储offset,这相当于一个用空间换时间的优化。

38、kafka如何保证数据不丢失不重复

从三个方面去考虑

  • producer如何不丢不重数据ack=0,1,-1;维护key
    • ack为响应方式,当leader认为写数据成功了就发送ack给produce。ack=0代表leader接收消息后立即返回,ack=1代表leader落盘后返回,ack=-1代表leader以及follower都落盘后返回。
    • 维护key,即给消息添加一个key,为生产者id、分区号以及消息序列组成的三元组。broker对于相同key的数据只维护一条。但是有一个问题,producer重启后id会变化,基于此,kafka将producerID交给transaction去管理,生产者重启后会去transaction中获取上次的id。
  • broker如何不丢数据?副本机制
  • consumer如何不丢不重数据?offset,事务,幂等性。
    • 维护offset是最基础的消息定位方式,提交offset后,消息还没处理完,这时候消费者挂了,数据丢失;提交offset前,消息已经处理完了,消费者挂掉会导致数据重复。
    • 事务要求下游消费者支持数据,如MySQL。将消费逻辑和offset的提交逻辑放在一个事务中。
    • 幂等性,举个例子,消费者读取消息:给A 100块钱。此时数据库中,A有500块钱,如果按照消息原来的意思,就是500加上100,改为600块。如果消息被重复消费,那么A会一直得到100块。对消息幂等性处理就是将语义理解为 将A的余额修改为600块。这样的话即便消息被重复读取也不会产生影响。

39、kafka保证存储一致性

40、大数据中的设计模式

  1. Spark中的Builder模式

       val spark = SparkSession
    			      .builder()
    			      .appName("SparkDemo")
    			      .master("local[*]")
    			      .getOrCreate()
    

    建造者模式将复杂产品的创建步骤分解在在不同的方法中,使得创建过程更加清晰,从而更精确控制复杂对象的产生过程;通过隔离复杂对象的构建与使用,也就是将产品的创建与产品本身分离开来,使得同样的构建过程可以创建不同的对象;并且每个具体建造者都相互独立,因此可以很方便地替换具体建造者或增加新的具体建造者,用户使用不同的具体建造者即可得到不同的产品对象。
    sparkSession的构建的步骤不同,或者方法参数不同,对象都是不同的。建造者模式将建造步骤一步一步的划分,更加清晰。

  2. Spark中的装饰器模式
    装饰器模式在不改变原代码的情况下,对代码进行拓展,进行增强。Spark中的RDD就采用了这种模式。实现了功能上的嵌套、扩展。

  3. ZooKeeper中的观察者模式。
    zookeeper中存储大家都关心的数据,观察者注册对该数据的监听事件,当数据发生改变或者状态变化的时候,zookeeper会通知观察者作出对应的动作。kafka中的controller、hadoop配置HA等都用到了该设计模式。

  4. spark中的伴生对象参考了单例模式。

41、zk和kafka的关系

kafka 2.8已经弃用zk,改用内置的仲裁器kraft。
这里主要讨论2.8之前
4. 在zk的/kafka/brokders下存储了kafka当前上线的服务器,/topics/first/partition中还存储了某个主题的某个分区的leader是哪个节点,该分区的ISR集合
5. 在zk的/kafka/controller下记录了当前起作用的controller
6. 0.9之前offset也是交给zk去存储的,但是由于频繁交互导致性能下降,0.9之后存储在offset的topic里面

42、kafka Kraft模式

kafka2.8后新增的模式,剔除了zk,由三台controller替代zk来管理kafka集群。
原因:

  • kafka和zk交互需要大量的网络通信,影响性能。
  • kafka和zk需要版本对应,不利于后续维护和升级。

优点:

  • 减少网络通信(与zk通信),减少响应时间
  • 手动配置controller,针对性增加controller来应对controller高负载。
  • kafka与外部框架耦合降低,可以独立运行。

43、kafka分区分配以及再平衡

消费者组消费模式
消费者组里的每一个消费者给分区的coordinator发送消费组请求,coordinator选出一个消费者作为leader,leader定制消费方案发送给coordinator。然后coordinator把消费方案发给所有的消费者,告知他应该消费的分区。
消费者与coordinator保持心跳通讯,超时则被剔除,触发再平衡
分区数只能增加,不能减少


Range

分区有0-6,消费者有0-2,那么每个消费者分7/3=2个分区,剩下的按序分配。
消费者0:P0,P1,P2
消费者1:P3,P4
消费者2:P5,P6
如果消费者0挂了,
那么
消费者1:0,1,2,3
消费者2:4,5,6

大数据环境容易产生数据倾斜


RoundRobin

轮询排序
在这里插入图片描述


Sticky

再分区的尽量少的去改变旧的分配方案。
举个例子,现在的分区方案:
C0:P0,P1,P2
C1:P3,P4
C2:P5,P6
消费者C0挂了之后,
C1:P0,P1,P3,P4
C2:P5,P6,P2
range是按序分配,而sticky是随机的。不一定就是012,34,56
尽量均匀可以减少再分区的开销。


CooperativeSticky

多种配合使用


默认的是range+cooperativeSticky

思考:为什么不能减少分区?
从数据流向方面考虑,减少的数据如何加入其它分区?加入分区会不会破坏分区内的有序性?offset如何维护?
实现逻辑非常复杂,所以为了性能考虑,不能减少分区。

44、为什么kafka读写文件这么快?

  1. 顺序写磁盘,将数据写入log文件的时候采用追加模式,一直在文件末尾加入数据,为顺序写。原因是省去了大量磁头寻址的方式
  2. 索引文件,并分段。读数据的时候,进行二分查找。同时也增加了并行能力。
  3. 零拷贝,对数据进行网络IO时,不会将数据从内核态拷贝到用户态。只是在内核态中进行传输。具体区别:
    • 非零拷贝模式下,消费者请求消费数据,请求会发送kafka后,kafka将请求的数据从磁盘或者内核态的页缓存中拉取到应用程序中,应用程序进行系统调用,将数据放入内核态的socket进程,再由网卡进行网络传输;
    • 在零拷贝模式下,消费者请求数据时,kafka直接将磁盘或者页缓存中的数据传到网卡中进行网络传输。
  4. 批量发送,可以指定缓冲区中的消息达到阈值后在发送,这种策略减少了网络IO。
  5. 数据压缩,比如snappy压缩。将数据压缩也可以减少网络IO。虽然需要消费者解压缩,需要消耗CPU,但是省去了大量数据的网络IO,性价比还是很高的。
  6. 主读主写:kafka不支持主写从读,主写从读可能会出现数据不一致的情况。也会造成网络通信导致速度下降。

45、flume的sink、source、channel种类

source

  • kafka:监听topic,转化为事件
  • avro:监听flume sink
  • netcat:将每行文本转化为一个事件
  • 文件
    • Spooldir Source适合用于同步新文件,但不适合对实时追加日志的文件进行监听并同步。
    • taildir source可以做到监听多个实时追加文件,实现断点续传(flume传输数据时突然挂掉了,重启后从上一次挂掉的位置继续传输数据)。
    • Exec source 适用于监控一个实时追加的文件,不能实现断点续传。

channel

根据通道种类:

  • file:文件式存储,可保证不丢数据
  • memory:存在内存中,性能高,安全性较差
  • jdbc:存入数据库,提供事务功能

根据sink发送策略:

  • defaultSinkProcess:默认策略,发送给某一个sink
  • LoadBalanceSinkProcess:负载均衡,每一个sink轮询抓取channel中的数据。
  • FailoverSinkProcess:按照sink优先级发送。前一个sink崩溃了,那就发送给下一个。

sink

  • hdfs
  • kafka
  • avro,发送给下一个flume source
  • hive

46、Hive的执行引擎mr,Tez,Spark,mr和Tez问题,适合什么场景,Tez底层,数据倾斜

MR是多job串联,基于磁盘,适合用于超大数据,虽然慢,但是可以跑完
Spark由于RDD机制,比较灵活,以天为单位的任务可以使用spark。
Tez则是完全基于内存,他将Map和Reduce任务拆解成一个个小任务,再灵活组合。一般适用于较小的数据,但是可以快速出结果。

47、kafka事务

在kafka的每个broker,都有一个transaction coordinator,即事务协调器。
由于事务过程需要持久化一些信息,这些信息会被存到一个主题里面。
每个broker都有事务协调器,事务协调器负责事务。
事务id % 分区数得到的就是事务管理的分区号,根据分区的leader所在的broker得到该broker的事务协调器,这个事务协调器就负责管理这个事务。
流程如下:

  1. 生产者向事务协调器请求producer ID,事务协调器返回producerID
  2. 生产者向分区发送数据。并向事务协调器发送commit请求。
  3. 事务协调器持久化commit请求,成功后向生产者返回确认信息。
  4. 事务协调器之后向分区发送commit,确认数据是否写成功了。
  5. 消费者返回成功后,事务协调器将事务成功信息持久化到对应的主题中。

在这里插入图片描述

总结

空着的是比较大头也是比较重要的知识点,很多人应该都知道。作者比较懒…就不写了,总结了一些犄角旮旯里面的知识点或者作者生疏的。

Logo

华为云1024程序员节送福利,参与活动赢单人4000元礼包,更有热门技术干货免费学习

更多推荐