kafka/spark/elasticsearch/hbase笔记

有赞日志平台架构(详见后文参考链接)中用到的一些组件,通过查阅文档,进行初步的了解.

各组件作用

在有赞日志平台中,kafka是日志中心,所有收集到的日志都会通过不同的topic发送到kafka,利用kafka的高吞吐量和消息中间件的异步解耦特性做一个中间桥梁.spark消费日志并且做一些处理逻辑然后或者写入elasticsearch做查询,或者匹配告警逻辑后做一些监控告警.其中elasticsearch保存索引,hbase保存原始数据,具体做法可参考链接3,这样可以使elasticsearch保存的数据量减小从而匹配filesystem cache,做到高效的查询(查询过程为通过es查询到对应的docid,然后以docid为键去hbase中查询,查到完整的数据).

各组件介绍

kafka

Topic/Partition:数据以topic来区分,每个topic可以划分为多个Partition,每个Partition中的数据赋予一个顺序的id number,称之为offset.kafka会持久化所有publish的消息,持久化时间根据配置的策略决定.但是大量的堆积并不会影响性能,性能是O(1)
集群中以partition来分区.每个partition会复制多份到其他机器来做到高可用,有一个server会作为该partition的leader负责读写,其他server作为follower来提供高可用性.topic划分为多个partition来保证性能的线性扩展
Producers:负责生产消息,可以指定发送到哪个topic的哪个partition
Consumers:消费者隶属于某个consumer group.某个消息只会发送给consumer group中的一个consumer,这样能做到消息消费的负载均衡
Kafka保证如下几点:
- 同样的一个producer发送到同一个topic的同一个partition确定有序性
- 一个consumer以partition中保存的顺序获得消息
- 复制因子为N的话,能保证N-1个server的failure不会丢提交的消息
Kafka作为一个消息系统:
- 每个partition给一个consumer消费
- 每个partition可以给不同的consumer group中的consumer消费
- 按partition消费可以做到并发消费,但按consumer group可以做到多个订阅者同时消费同一个partition
Kafka作为一个存储系统:
- 可以和producer有确认机制,确保partition中数据高可用落盘之后才回复
- 不论数据多少,存取都是O(1)的复杂度
底层存储设计:底层如何做到O(1)的存取参考该文:https://cloud.tencent.com/developer/article/1057763
kafka如何实现exactly once语义:通过类似tcp的seqno来保证不会重复append 数据.而且seqno会持久化,以免一个broker失败后其他broker能继续按当前的seqno来处理
参考:https://www.confluent.io/blog/exactly-once-semantics-are-possible-heres-how-apache-kafka-does-it/

hbase

源于Google的”Bigtable: A Distributed Storage System for Structured Data”.关键特性是可实时的随机读写超大规模的数据集.可通过增加节点来实现线性扩展
数据模型
- 列族:同一个列族的所有成员具有相同的前缀.并且列族作为表模式定义的一部分必须预先给出.但是列族成员可以动态添加.以一个图片表作为示例,例如有两个列族,contents以及info,contents:image为图片内容,info:format为图片格式,info:geo为图片拍摄的坐标值
- 所有的调优和存储都是在列族的格式上进行的,所以最好使所有列族成员具有相同的访问模式和大小特征.
- 单元格:cell.行列交叉处为一个cell,cell是有版本的,默认为插入时的时间戳.(默认保存多少个版本?并且查询时返回多少个版本?)
- 行是升序排列的,按字节序.行的键值和cell的内容都是二进制字节流,但是列族的前缀必须是可打印字符
- 区域:region.每个区域由它所属于的表,它所包含的第一行及其最后一行(左闭右开区间)来表示.
  区域是在集群上分布数据的最小单位.
实现

Hbase

Master:启动一个安装,把区域分配给注册的regionserver,恢复regionserver的故障.
RegionServer:负责零个或者多个区域的管理以及响应客户端的读写请求.
Hbase通过HDFS来持久化存储数据
hbase:meta保存在zookeeper,维护着当前集群上所有区域的列表、状态和位置.区域名作为键,区域名由所属的表名、区域起始行、区域的创建时间及其整体的MD5组成.
客户端首先通过zookeeper查找hbase:meta的位置,然后通过区域名获取用户空间区域所在节点及其位置,接着可以直接和regionserver交互.为了较少交互需要缓存hbase:meta信息直到碰到错误之前会一直直接使用缓存
写操作:先写commit log,然后是内存中的memstore,然后会被刷入文件系统
读操作:读memstore找到就返回,否则依次从新到旧查flush file.会对文件系统的flush file进行压缩合并,并且有一个进程会检测是否超出区域的大小,超出会进行分割.(类似leveldb)
leveldb的实现参考https://juejin.im/post/5c36ad6051882525616db7fa

elasticsearch

存储术语:index->database,type->table,mapping->schema,每个index默认有五个shard,每个shard默认有一个replica,每个shard是一个lucene index
集群术语:node,masternode:负责创建和删除index,确定哪些node属于集群以及给node分配shards.
datanode:存储数据和倒排索引.
存储:倒排索引有序存储.es中一条记录是一个document.
写入时必须等待primary和replica的translog都fsyncd.删除时写入一个.del文件,最后返回时过滤.更新时先删除再写入一个新的版本.查询时会从所有shard取出之后在协调节点取top10然后通过docid获取到doc并返回.lucene segment合并时会将.del文件中的数据删除

spark

Apache Spark是用于大数据处理的集群计算框架.最突出的表现在于它能将作业与作业之间产生的大规模的工作数据集存储在内存中
RDD(Resilient Distributed Dataset)弹性分布式数据集是所有Spark程序的核心,RDD的创建有三种方法
- 来自一个内存中的对象集合
- 使用外部存储器（例如HDFS)中的数据集
- 对现有RDD进行转换
转换和动作:转换时从现有RDD生成新的RDD,而动作则触发对RDD的计算并对计算结果执行某种操作,要么立即返回给用户,要么保存到外部存储器中
适合于迭代算法,上次的结果可以保存到内存中供下次使用,并且提供了不同的持久化方法,可以直接保存,可以序列化后保存,也可以保存到磁盘上

spark统计词频

val conf = new SparkConf().setAppName("wiki_test") // create a spark config object
val sc = new SparkContext(conf) // Create a spark context
val data = sc.textFile("/path/to/somedir") // Read files from "somedir" into an RDD of (filename, content) pairs.
val tokens = data.flatMap(_.split(" ")) // Split each file into a list of tokens (words).
val wordFreq = tokens.map((_, 1)).reduceByKey(_ + _) // Add a count of one to each token, then sum the counts per word type.
wordFreq.sortBy(s => -s._2).map(x => (x._2, x._1)).top(10) // Get the top 10 words. Swap word and count to sort by count.

DAG和Spark? Spark的运行机制