案例分享：集群技术在七牛云存储中的应用

分享人介绍：七牛数据平台工程师王团结，主要负责数据平台的设计和研发。关注大数据处理、高性能系统服务、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。这样做有两个原因1.kafka的api对非jvm系的语言支持非常不友好，forward提供更通用的http接口2.forward层可以做路由、kafkatopic、kafkapartitionkey等逻辑，进一步减少agent端的逻辑forward层不含状态，完全可以实现水平扩展，别担心成为瓶颈。由于高可用性，forward通常有不止一个例子，这将导致日志顺序问题，agent遵循一定的规则(round-robin、failover等。)选择forward实例，即使kafkapartitionkey是一样的，由于forward层的存在，最终落入kafka的数据顺序可能与agent发送的顺序不同。我们容忍乱序，因为生成日志的业务基本上是分布式的，保证单台机器的日志顺序意义不大。我们容忍乱序，因为生成日志的业务基本上是分布式的，保证单台机器的日志顺序意义不大。如果业务对顺序有要求，则必须直接将数据发送到kafka，并选择partitionkey。kafka只能保证partition级别的顺序。通过上述流程，将数据汇集到当地机房kafka集群，然后汇集到核心机房kafka，最终供消费者使用。由于kafka的miror对网络不友好，我们选择了更简单的flume来完成跨机房的数据传输。flume在不同的数据源中传输数据是灵活的，但有几点需要注意1.memory-Channel效率很高，但是可能会有丢失数据的风险，file-安全性高但性能低的chanel。我们用memory-channel，然而，将capacity设置得足够小，使内存中的数据尽可能少，在意外重启和断电时丢失的数据也很少。个人排斥file-channel，一方面，效率，另一方面，对flume的期望是数据传输。当引入file-chanel时，其角色将转换为存储，这在整个过程中是不合适的。通常flume的sink端是kafka和hdfs，可用性和扩展性都很好，不用担心数据拥堵。2.默认情况下，httpsouce没有设置线程池，存在性能问题。如果有用，需要自己修改代码。3.单sink速度跟不上时，需要多个sink。例如，如果跨机房数据传输网络延迟高单rpcsink吞吐和hdfssink效率低下，我们将在一个channel后匹配十多个sink。kafka使用的关键点kafka在性能和可扩展性方面都很好，以下几点需要注意以下1点.topic的划分，大topic对生产者有利，维护成本低，小topic对消费者友好。若是完全不相关的相关数据源，且topic数不发散，则优先考虑topic。2.kafka的并行单位是partition，partition的数量与整体吞吐量直接相关，但partition的数量并不是越大越高，三个partition可以吃一个普通的硬盘io。因此，partition数是由数据规模决定的，最终还是需要硬盘来抵抗。3.选择不当的partitionkey可能会导致数据倾斜。partititionkey只有在对数据有顺序要求时才能使用。kafka的producersdk在没有指定partitionkey的情况下，只会在一定时间内将数据写入一个partition。在这种情况下，当producer的数量少于partition时，也会导致数据倾斜，可以增加producer的数量来解决这个问题。数据到kafka后，数据同步到hdfs进行离线统计。另一种方法用于实时计算。由于今天的时间有限，我们只能与您分享一些实时计算的经验，以实时计算我们选择的sparkstreaming。目前我们只有统计需求，没有迭代计算需求，所以sparkstreaming使用保守，从kakfa读数据统计落入mongo，中间状态数据很少。好处是系统吞吐量大，但很少遇到与内存相关的问题。sparkstreaming对存储计算结果的dbtps要求很高。例如，有10w域名需要统计流量，batchinterval为10s，每个域名有4个相关统计项目，平均为4wtps。考虑到峰值可能更高，固态硬盘上的mongo只能抵抗1wtps，我们将考虑使用redis来抵抗如此高的tps具有外部状态的task逻辑不能重新访问。当speculation参数打开时，计算结果可能不准确。举个简单的例子，这是一个将计算结果存储在mongo中的task任务。如果重做，落入mongo的结果会比实际结果多。举个简单的例子，这是一个将计算结果存储在mongo中的task任务。如果重做，落入mongo的结果会比实际情况多。有状态的对象的生命周期不容易管理，不可能每个task都去new。我们的策略是jvm中的对象，并在代码层面进行并发控制。类似下面。在spark1.3的后版本中，kafkadirectapi被引入试图解决数据准确性问题。使用direct可以在一定程序中缓解准确性问题，但不可避免地会出现一致性问题。为什么这么说？directapi暴露了kafkaconsumerofset的管理(以前是在zookeeper中异步存储)。只有保存计算结果和ofset在同一事务中，才能保证准确性。这项事务有两种方式可以实现，一种是用mysql等支持事务的数据库保存计算结果ofset，另一种是实现两个阶段的提交。流式计算中实现这两种方法的成本都很高。其次，directapi存在性能问题，因为它实际上是在计算时从kafka阅读数据的，这对整体吞吐量有很大影响。这就是我们想分享的，最后展示我们的在线规模。flume＋kafka＋spark8台高配机，日均数据500亿，峰值80wtps。