如何用 Hadoop/Spark 构建七牛数据平台

标签: spark | 发表时间:2015-07-30 17:30 | 作者:
分享到:
出处:http://my.oschina.net/leejun2005

数据平台在大部分公司都属于支撑性平台,做的不好立刻会被吐槽,这点和运维部门很像。所以在技术选型上优先考虑现成的工具,快速出成果,没必要去担心有技术负担。早期,我们走过弯路,认为没多少工作量,收集存储和计算都自己研发,发现是吃力不讨好。去年上半年开始,我们全面拥抱开源工具,搭建自己的数据平台。

1、数据平台设计理念

公司的主要数据来源是散落在各个业务服务器上的半结构化日志,比如系统日志、程序日志、访问日志、审计日志等。日志是最原始的数据记录,如果不是日志,肯定会有信息上的丢失。说个简单的例子,需求是统计Nginx上每个域名的的流量,这个完全可以通过一个简单的Nginx模块去完成,但是如果统计的是不同来源的流量就无法做了。所以需要原始的完整的日志。

有种手法是业务程序把日志通过网络直接发送出去,但是这并不可取,因为网络和接收端并不完全可靠,当出问题时会对业务造成影响或者日志丢失。因此,对业务侵入最小最自然的方式是把日志落到本地硬盘上。

2、数据平台设计架构

 

2.1 Agent设计需求

每台机器上会有一个Agent去同步这些日志,这是个典型的队列模型,业务进程在不断的push,Agent在不停的pop。Agent需要有记忆功能,用来保存同步的位置(offset),这样才尽可能保证数据准确性,但不可能做到完全准确。由于发送数据和保存offset是两个动作,不具有事务性,不可避免的会出现数据不一致性情况,通常是发送成功后保存offset,那么在Agent异常退出或机器断电时可能会造成多余的数据。

在这里,Agent需要足够轻,这主要体现在运维和逻辑两个方面。Agent在每台机器上都会部署,运维成本、接入成本是需要考虑的。Agent不应该有解析日志、过滤、统计等动作,这些逻辑应该给数据消费者。倘若Agent有较多的逻辑,那它是不可完成的,不可避免的经常会有升级变更动作。

2.2 数据收集流程

数据收集这块的技术选择,Agent是用Go自己研发的,消息中间件Kafka,数据传输工具Flume。说到数据收集经常有人拿Flume和Kafka做比较,我看来这两者定位是不同的,Flume更倾向于数据传输本身,Kakfa是典型的消息中间件用于解耦生产者消费者。

具体架构上,Agent并没把数据直接发送到Kafka,在Kafka前面有层由Flume构成的forward。这样做有两个原因:

1. Kafka的API对非JVM系的语言支持很不友好,forward对外提供更加通用的http接口。

2. forward层可以做路由、Kafka topic和Kafka partition key等逻辑,进一步减少Agent端的逻辑。

forward层不含状态,完全可以做到水平扩展,不用担心成为瓶颈。出于高可用考虑,forward通常不止一个实例,这会带来日志顺序问题,Agent按一定规则(round-robin、failover等)来选择forward实例,即使Kafka partition key一样,由于forward层的存在,最终落入Kafka的数据顺序和Agent发送的顺序可能会不一样。我们对乱序是容忍的,因为产生日志的业务基本是分布式的,保证单台机器的日志顺序意义不大。如果业务对顺序性有要求,那得把数据直接发到Kafka,并选择好partition key,Kafka只能保证partition级的顺序性。

2.3 跨机房收集要点

多机房的情形,通过上述流程,先把数据汇到本地机房Kafka 集群,然后汇聚到核心机房的Kafka,最终供消费者使用。由于Kafka的mirror对网络不友好,这里我们选择更加的简单的Flume去完成跨机房的数据传送。Flume在不同的数据源传输数据还是比较灵活的,但有几个点需要注意:

1. memory-channel效率高但可能有丢数据的风险,file-channel安全性高但性能不高。我们是用memory-channel,但把capacity设置的足够小,使内存中的数据尽可能少,在意外重启和断电时丢的数据很少。个人比较排斥file-channel,效率是一方面,另一个是对Flume的期望是数据传输,引入file-channel时,它的角色会向存储转变,这在整个流程中是不合适的。通常Flume的sink端是Kafka和HDFS这种可用性和扩张性比较好的系统,不用担心数据拥堵问题。

2. 默认的http souce 没有设置线程池,有性能问题,如果有用到,需要自己修改代码。

3. 单sink速度跟不上时,需要多个sink。像跨机房数据传输网络延迟高单rpc sink吞吐上不去和HDFS sink效率不高情形,我们在一个channel后会配十多个sink。

2.4 Kafka使用要点

Kafka在性能和扩展性很不错,以下几个点需要注意下:

1. topic的划分,大topic对生产者有利且维护成本低,小topic对消费者比较友好。如果是完全不相关的相关数据源且topic数不是发散的,优先考虑分topic。

2. Kafka的并行单位是partition,partition数目直接关系整体的吞吐量,但parition数并不是越大越高,3个partition就能吃满一块普通硬盘IO了。所以partition数是由数据规模决定,最终还是需要硬盘来抗。

3. partition key选择不当,可能会造成数据倾斜。在对数据有顺序性要求才需使用partition key。Kafka的producer sdk在没指定partition key时,在一定时间内只会往一个partition写数据,这种情况下当producer数少于partition数也会造成数据倾斜,可以提高producer数目来解决这个问题。

2.5 数据离线和实时计算

数据到Kafka后,一路数据同步到HDFS,用于离线统计。另一路用于实时计算。由于今天时间有限,接下来只能和大家分享下实时计算的一些经验。

实时计算我们选择的Spark Streaming。我们目前只有统计需求,没迭代计算的需求,所以Spark Streaming使用比较保守,从Kakfa读数据统计完落入mongo中,中间状态数据很少。带来的好处是系统吞吐量很大,但几乎没遇到内存相关问题

Spark Streaming对存储计算结果的数据库tps要求较高。比如有10万个域名需要统计流量,batch interval为10s,每个域名有4个相关统计项,算下来平均是4万 tps,考虑到峰值可能更高,固态硬盘上的mongo也只能抗1万tps,后续我们会考虑用redis来抗这么高的tps

有外部状态的task逻辑上不可重入的,当开启speculation参数时候,可能会造成计算的结果不准确。说个简单的例子。这个任务,如果被重做了,会造成落入mongo的结果比实际多。

有状态的对象生命周期不好管理,这种对象不可能做到每个task都去new一个。我们的策略是一个JVM内一个对象,同时在代码层面做好并发控制。类似下面。

 

在Spark1.3的后版本,引入了 Kafka Direct API试图来解决数据准确性问题,使用Direct在一定程序能缓解准确性问题,但不可避免还会有一致性问题。为什么这样说呢?Direct API 把Kafka consumer offset的管理暴露出来(以前是异步存入ZooKeeper),当保存计算结果和保存offset在一个事务里,才能保证准确。

这个事务有两种手段做到,一是用MySQL这种支持事务的数据库保存计算结果offset,一是自己实现两阶段提交。这两种方法在流式计算里实现的成本都很大。其次Direct API 还有性能问题,因为它到计算的时候才实际从Kafka读数据,这对整体吞吐有很大影响。

3、七牛数据平台规模

 

要分享的就这些了,最后秀下我们线上的规模:Flume + Kafka + Spark8台高配机器,日均500亿条数据,峰值80万tps。


相关 [hadoop spark 数据] 推荐:

Spark:比Hadoop更强大的分布式数据计算项目

- - 标点符
Spark是一个由加州大学伯克利分校(UC Berkeley AMP)开发的一个分布式数据快速分析项目. 它的核心技术是弹性分布式数据集(Resilient distributed datasets),提供了比Hadoop更加丰富的MapReduce模型,可以快速在内存中对数据集进行多次迭代,来支持复杂的数据挖掘算法和图计算算法.

如何用 Hadoop/Spark 构建七牛数据平台

- - leejun_2005的个人页面
数据平台在大部分公司都属于支撑性平台,做的不好立刻会被吐槽,这点和运维部门很像. 所以在技术选型上优先考虑现成的工具,快速出成果,没必要去担心有技术负担. 早期,我们走过弯路,认为没多少工作量,收集存储和计算都自己研发,发现是吃力不讨好. 去年上半年开始,我们全面拥抱开源工具,搭建自己的数据平台. 公司的主要数据来源是散落在各个业务服务器上的半结构化日志,比如系统日志、程序日志、访问日志、审计日志等.

Spark是否会替代Hadoop?

- - CSDN博客推荐文章
我经常会从客户或者网上听到这个问题,尤其是最近几年. 那么关于spark哪些被我们神化了,哪些又是真实的,以及它在“大数据”的生态系统中又是怎样的. 说实话,其实我把这个问题作为标题是有问题的,但是我们经常会这样问. Hadoop并不是一个单独的产品而是一个生态系统,而spark也是一样的. 目前Hadoop生态系统主要包括:.

hadoop/spark关闭钩子研究

- - 开源软件 - ITeye博客
引子:在使用spark和hadoop的时候,遇到一些进程退出时的报错. 因此顺便研究了一下jvm以及一些开源框架的关闭钩子的机制. 这篇文章不涉及底层native实现,仅限Java层面. 注册jvm关闭钩子通过Runtime.addShutdownHook(),实际调用ApplicationShutdownHooks.add().

Spark-ML-数据获取/处理/准备

- - CSDN博客综合推荐文章
UCL机器学习知识库:包括近300个不同大小和类型的数据集,可用于分类、回归、聚类 和推荐系统任务. 数据集列表位于: http://archive.ics.uci.edu/ml/. Amazon AWS公开数据集:包含的通常是大型数据集,可通过Amazon S3访问. 这些数据 集包括人类 基因组项目 、 Common Crawl 网页语料 库、维基百 科数据和 Google Books Ngrams.

实用教程|Spark性能优化之道——解决Spark数据倾斜

- - IT瘾-geek
实用教程|Spark性能优化之道——解决Spark数据倾斜.     2017-03-16 11:31  浏览次数:108. 为何要处理数据倾斜(Data Skew). 对Spark/Hadoop这样的大数据系统来讲,数据量大并不可怕,可怕的是数据倾斜. 数据倾斜指的是,并行处理的数据集中,某一部分(如Spark或Kafka的一个Partition)的数据显著多于其它部分,从而使得该部分的处理速度成为整个数据集处理的瓶颈.

Hadoop的数据管理

- - 技术改变世界 创新驱动中国 - 《程序员》官网
本文主要介绍Hadoop的数据管理,主要包括Hadoop的分布式文件系统HDFS、分布式数据库HBase和数据仓库工具Hive. HDFS是分布式计算的存储基石,Hadoop分布式文件系统和其他分布式文件系统有很多类似的特性:. 对于整个集群有单一的命名空间;. 具有数据一致性,都适合一次写入多次读取的模型,客户端在文件没有被成功创建之前是无法看到文件存在的;.

大数据架构hadoop

- - CSDN博客云计算推荐文章
摘要:Admaster数据挖掘总监 随着互联网、移动互联网和物联网的发展,谁也无法否认,我们已经切实地迎来了一个海量数据的时代,数据调查公司IDC预计2011年的数据总量将达到1.8万亿GB,对这些海量数据的分析已经成为一个非常重要且紧迫的需求. 随着互联网、移动互联网和物联网的发展,谁也无法否认,我们已经切实地迎来了一个海量数据的时代,数据调查公司IDC预计2011年的数据总量将达到1.8万亿GB,对这些海量数据的分析已经成为一个非常重要且紧迫的需求.

虚拟座谈会:大数据一栈式方案Spark

- - CSDN博客架构设计推荐文章
Spark正在占据越来越多的大数据新闻的重要位置,除了性能优异,Spark到底具备了那些特性,让学术界和工业界对其充满了兴趣. 同时,Spark还处在快速发展的阶段,开发者和用户不得不解决不稳定和bug,Scala语言也有较高的学习门槛,这些也会成为Spark普及的障碍. 当然,尽管Spark提供了一栈式的大数据方案,但并不意味着他适合任何场景.

实用 | 从Apache Kafka到Apache Spark安全读取数据

- - IT瘾-bigdata
随着在CDH平台上物联网(IoT)使用案例的不断增加,针对这些工作负载的安全性显得至关重要. 本篇博文对如何以安全的方式在Spark中使用来自Kafka的数据,以及针对物联网(IoT)使用案例的两个关键组件进行了说明. Cloudera Distribution of Apache Kafka 2.0.0版本(基于Apache Kafka 0.9.0)引入了一种新型的Kafka消费者API,可以允许消费者从安全的Kafka集群中读取数据.