标签:
转自:http://blog.qiniu.com/archives/3928
7月30日,七牛数据平台工程师王团结在CSDN Spark微信用户群,与近千名Spark技术开发人员,结合七牛内部使用的数据平台,深入分享了团队是如何利用Flume、Kafka、Spark Streaming等技术搞定每天500亿条日志的,并详细讲解了各个工具使用的注意点。王团结,主要负责七牛数据平台的设计研发工作,关注大数据处理和 高性能系统服务,对Hadoop、Flume、Kafka、Spark等离线、分布式计算技术十分感兴趣。以下为演讲实录。
数据平台在大部分公司都属于支撑性平台,做的不好立刻会被吐槽,这点和运维部门很 像。所以在技术选型上优先考虑现成的工具,快速出成果,没必要去担心有技术负担。早期,我们走过弯路,认为没多少工作量,收集存储和计算都自己研发,发现 是吃力不讨好。去年上半年开始,我们全面拥抱开源工具,搭建自己的数据平台。
公司的主要数据来源是散落在各个业务服务器上的半结构化日志,比如系统日志、程序日 志、访问日志、审计日志等。日志是最原始的数据记录,如果不是日志,肯定会有信息上的丢失。说个简单的例子,需求是统计Nginx上每个域名的的流量,这 个完全可以通过一个简单的Nginx模块去完成,但是如果需要统计不同来源的流量就无法做了,所以需要原始的完整的日志。
有种手法是业务程序把日志通过网络直接发送出去,但是这并不可取,因为网络和接收端并不完全可靠,当出问题时会对业务造成影响或者日志丢失。因此,对业务侵入最小最自然的方式是把日志落到本地硬盘上。
Agent设计需求
每台机器上会有一个Agent去同步这些日志,这是个典型的队列模型,业务进程在不 断的push,Agent在不停地pop。Agent需要有记忆功能,用来保存同步的位置(offset),这样才尽可能保证数据准确性,但不可能做到完 全准确。由于发送数据和保存offset是两个动作,不具有事务性,不可避免的会出现数据不一致性情况,通常是发送成功后保存offset,那么在 Agent异常退出或机器断电时可能会造成多余的数据。 Agent需要足够轻,这主要体现在运维和逻辑两个方面。Agent在每台机器上都会部署,运维成本、接入成本是需要考虑的。Agent不应该有解析日 志、过滤、统计等动作,这些逻辑应该给数据消费者。倘若Agent有较多的逻辑,那它是不可完成的,不可避免的经常会有升级变更动作。
数据收集流程
数据收集这块的技术选择,Agent是用Go自己研发的,消息中间件Kafka,数 据传输工具Flume。说到数据收集经常有人拿Flume和Kafka做比较,我看来这两者定位是不同的,Flume更倾向于数据传输本身,Kakfa是 典型的消息中间件用于解耦生产者消费者。
具体架构上,Agent并没把数据直接发送到Kafka,在Kafka前面有层由Flume构成的forward。这样做有两个原因:
forward层不含状态,完全可以做到水平扩展,不用担心成为瓶颈。出于高可用考 虑,forward通常不止一个实例,这会带来日志顺序问题,Agent按一定规则(round-robin、failover等)来选择forward 实例,即使Kafka partition key一样,由于forward层的存在,最终落入Kafka的数据顺序和Agent发送的顺序可能会不一样。我们对乱序是容忍的,因为产生日志的业务基 本是分布式的,保证单台机器的日志顺序意义不大。如果业务对顺序性有要求,那得把数据直接发到Kafka,并选择好partition key,Kafka只能保证partition级的顺序性。
多机房的情形,通过上述流程,先把数据汇到本地机房Kafka集群,然后汇聚到核心机房的Kafka,最终供消费者使用。由于Kafka的mirror对网络不友好,这里我们选择更加的简单的Flume去完成跨机房的数据传送。
Flume使用要点
Flume在不同的数据源传输数据还是比较灵活的,但有以下几个点需要注意。
数据到Kafka后,一路数据同步到HDFS,用于离线统计,另一路用于实时计算。由于今天时间有限,接下来只能和大家分享下实时计算的一些经验。
实时计算选择的是Spark Streaming。目前只有统计需求,没迭代计算的需求,Spark Streaming使用比较保守。从Kakfa读数据统计完存入MongoDB中,中间状态数据很少,好处是系统吞吐量很大,但没遇到内存相关问题。
Spark Streaming对保存计算结果的数据库TPS要求较高。假如有10w个域名需要统计流量,batch interval为10s,每个域名有4个相关统计项,算下来平均是4w TPS,这只是平均值,实际峰值更高,固态硬盘上的MongoDB也只能抗1w TPS,后续我们会考虑用Redis来抗这么高的TPS。
当开启speculation参数或代码层面没处理好异常时,task可能会被重放。但是有外部状态的task是不可重入的,否则会造成计算结果的不准确。说个简单的例子,如下:
这个任务,如果被重放了,会造成落入MongoDB的结果比实际多。
有些对象会包含状态,这些状态的生成需要较大的代价,不能做到在每次使用时都去new一个。我们对这种对象的处理策略是JVM内一个对象,同时在代码层面做好并发控制。类似下面:
如果这种使用形式有性能问题,可以考虑实现一个pool来管理。
在Spark1.3的后版本,streaming为Kafka引入了新的 Direct API试图解决数据准确性问题。Direct API把Kafka consumer offset的管理暴露出来(旧版本是异步存入Zookeeper),让使用者透明的管理consumer offset,这在一定程度上能缓解准确性问题,但不可避免还会有一致性问题。为什么这样说呢?只有计算结果和offset两者的保存具有事务性,才能完 全准确。这个事务有两种手段做到,一种是用Mysql这种支持事务的数据库,把计算结果和offset的保存放在一个事务里,另一种是自己实现两阶段提 交。Direct API 还会有性能问题,因为它到计算的时候才实际从Kafka读数据,这对整体吞吐有很大影响。
七牛数据平台规模
要分享的就这些了,最后秀下实时计算这边规模:Flume+Kafka+Spark 混部在8台高配机器,日均500亿条数据,峰值80w TPS。
标签:
原文地址:http://www.cnblogs.com/leoncfor/p/4704080.html