数据处理平台架构中的SMACK组合：Spark、Mesos、Akka、Cassandra以及Kafka

综述

Spark - 一套高速通用型引擎，用于实现分布式大规模数据处理任务。

Mesos - 集群资源管理系统，能够立足于分布式应用程序提供行之有效的资源隔离与共享能力。

Akka - 一套用于在JVM之上构建高并发、分布式及弹性消息驱动型应用程序的工具包与运行时。

Cassandra - 一套分布式高可用性数据库，旨在跨越多座数据中心处理大规模数据。

Kafka -一套高吞吐能力、低延迟、分布式消息收发系统/提交日志方案，旨在处理实时数据供给。

存储层: Cassandra

Cassandra一直以其高可用性与高吞吐能力两大特性而备受瞩目，其同时能够处理极为可观的写入负载并具备节点故障容错能力。以CAP原则为基础，Cassandra能够为业务运营提供可调整的一致性/可用性水平。

更有趣的是，Cassandra在处理数据时拥有线性可扩展能力（即可通过向集群当中添加节点的方式实现负载增容）并能够提供跨数据中心复制（简称XDCR）能力。事实上，跨数据中心复制功能除了数据复制，同时也能够实现以下各类扩展用例：

• 地理分布式数据中心处理面向特定区域或者客户周边位置之数据。
• 在不同数据中心之间者数据迁移，从而实现故障后恢复或者将数据移动至新数据中心。
• 对运营工作负载与分析工作负载加以拆分。

但上述特性也都有着自己的实现成本，而对于Cassandra而言这种成本体现为数据模型——这意味着我们需要通过聚类对分区键及入口进行分组/分类，从而实现嵌套有序映射。以下为简单示例：

CREATE TABLE campaign(      id uuid,    year int,    month int,    day int,    views bigint,    clicks bigint,    PRIMARY KEY (id, year, month, day)    );        INSERT INTO campaign(id, year, month, day, views, clicks)      VALUES(40b08953-a…,2015, 9, 10, 1000, 42);        SELECT views, clicks FROM campaign      WHERE id=40b08953-a… and year=2015 and month>8;  

为了获取某一范围内的特定数据，我们必须指定全键，且不允许除列表内最后一列之外的其它任何范围划定得以执行。这种限制用于针对不同范围进行多重扫描限定，否则其可能带来随机磁盘访问并拖慢整体性能表现。这意味着该数据模型必须根据读取查询进行认真设计，从而限制读取/扫描量——但这同时也会导致对新查询的支持灵活性有所下降。

那么如果我们需要将某些表加入到其它表当中，又该如何处理？让我们考虑下一种场景：针对特定月份对全部活动进行总体访问量计算。

CREATE TABLE event(      id uuid,    ad_id uuid,    campaign uuid,    ts bigint,    type text,    PRIMARY KEY(id)    );

在特定模型之下，实现这一目标的惟一办法就是读取全部活动、读取全部事件、汇总各属性值（其与活动id相匹配）并将其分配给活动。实现这类应用程序操作显然极具挑战，因为保存在Casandra中的数据总量往往非常庞大，内存容量根本不足以加以容纳。因此我们必须以分布式方式对此类数据加以处理，而Spark在这类用例中将发挥重要作用。

处理层: Spark

Spark的抽象核心主要涉及RDD（即弹性分布式数据集，一套分布式元素集合）以及由以下四个主要阶段构成的工作流：

• RDD操作（转换与操作）以DAG（即有向无环图）形式进行
• DAG会根据各任务阶段进行拆分，并随后被提交至集群管理器
• 各阶段无需混洗/重新分配即可与任务相结合
• 任务运行在工作程序之上，而结果随后返回至客户端

以下为我们如何利用Spark与Cassandra解决上述问题：

val sc = new SparkContext(conf)        case class Event(id: UUID, ad_id: UUID, campaign: UUID, ts: Long, `type`: String)        sc.cassandraTable[Event]("keyspace", "event")      .filter(e => e.`type` == "view" && checkMonth(e.ts))    .map(e => (e.campaign, 1))    .reduceByKey(_ + _)    .collect()

指向Cassandra的交互通过Spark-Cassandra-连接器负责执行，其能够让整个流程变得更为直观且简便。另有一个非常有趣的选项能够帮助大家实现对NoSQL存储内容的交互——SparkSQL，其能够将SQL语句翻译成一系列RDD操作。

case class CampaignReport(id: String, views: Long, clicks: Long)        sql("""SELECT campaign.id as id, campaign.views as views,      campaign.clicks as clicks, event.type as type       FROM campaign       JOIN event ON campaign.id = event.campaign    """).rdd    .groupBy(row => row.getAs[String]("id"))    .map{ case (id, rows) =>    val views = rows.head.getAs[Long]("views")    val clicks = rows.head.getAs[Long]("clicks")        val res = rows.groupBy(row => row.getAs[String]("type")).mapValues(_.size)    CampaignReport(id, views = views + res("view"), clicks = clicks + res("click"))        }.saveToCassandra(“keyspace”, “campaign_report”)

通过几行代码，我们已经能够实现原生Lambda设计——其复杂度显然较高，但这一示例表明大家完全有能力以简单方式实现既定功能。

类MapReduce解决方案：拉近处理与数据间的距离

Spark-Cassandra连接器拥有数据位置识别能力，并会从集群内距离最近的节点处读取数据，从而最大程度降低数据在网络中的传输需求。为了充分发挥Spark-C*连接器的数据位置识别能力，大家应当让Spark工作程序与Cassandra节点并行协作。

除了Spark与Cassandra的协作之外，我们也有理由将运营（或者高写入强度）集群同分析集群区分开来，从而保证：

• 不同集群能够独立进行规模伸缩
• 数据由Cassandra负责复制，而无需其它机制介入
• 分析集群拥有不同的读取/写入负载模式
• 分析集群能够容纳额外数据（例如词典）与处理结果
• Spark对资源的影响只局限于单一集群当中

下面让我们再次回顾Spark的应用程序部署选项：

目前我们拥有三种主要集群资源管理器选项可供选择：

单独使用Spark——Spark作为主体，各工作程序以独立应用程序的形式安装并执行（这明显会增加额外资源负担，且只支持为每工作程序分配静态资源）

如果大家已经拥有Hadoop生态系统，那么YARN绝对是个不错的选项

Mesos自诞生之初就在设计中考虑到对集群资源的动态分配，而且除了Hadoop应用程序之外，同时也适合处理各类异构工作负载

Mesos架构

Mesos集群由各主节点构成，它们负责资源供应与调度，而各从节点则实际承担任务执行负载。在HA模式当中，我们利用多个主ZooKeeper节点负责进行主节点选择与服务发现。Mesos之上执行的各应用程序被称为“框架（Framework）”，并利用API处理资源供应及将任务提交至Mesos。总体来讲，其任务执行流程由以下几个步骤构成：

-从节点为主节点提供可用资源

-主节点向框架发送资源供应

-调度程序回应这些任务及每任务资源需求

-主节点将任务发送至从节点

将Spark、Mesos以及Cassandra加以结合

正如之前所提到，Spark工作程序应当与Cassandra节点协作，从而实现数据位置识别能力以降低网络流量与Cassandra集群负载。下图所示为利用Mesos实现这一目标的可行部署场景示例：

-Mesos主节点与ZooKeeper协作

-Mesos从节点与Cassandra节点协作，从而为Spark提供更理想的数据位置

-Spark二进制文件部署至全部工作节点当中，而spark-env.sh则配置以合适的主端点及执行器jar位置

park执行器JAR被上传至S3/HDFS当中

根据以上设置流程Spark任务可利用简单的spark-submit调用从任意安装有Spark二进制文件并上传有包含实际任务逻辑jar的工作节点被提交至集群中。

spark-submit --class io.datastrophic.SparkJob /etc/jobs/spark-jobs.jar  

由于现有选项已经能够运行Docker化Spark，因此我们不必将二进制文件分发至每个单一集群节点当中。

定期与长期运行任务之执行机制

每套数据处理系统迟早都要面对两种必不可少的任务运行类别：定期批量汇聚型定期/阶段性任务以及以数据流处理为代表的长期任务。这两类任务的一大主要要求在于容错能力——各任务必须始终保持运行，即使集群节点发生故障。Mesos提供两套出色的框架以分别支持这两种任务类别。

Marathon是一套专门用于实现长期运行任务高容错性的架构，且支持与ZooKeeper相配合之HA模式。其能够运行Docker并提供出色的REST API。以下shell命令示例为通过运行spark-submit实现简单任务配置：

Chronos拥有与Marathon相同的特性，但其设计目标在于运行定期任务，而且总体而言其分布式HA cron支持任务图谱。以下示例为利用简单的bash脚本实现S3压缩任务配置：

目前已经有多种框架方案可供选择，或者正处于积极开发当中以对接各类系统中所广泛采用的Mesos资源管理功能。下面列举其中一部分典型代表：

-Hadoop

-Cassandra

-Kafka

-Myriad: YARN on Mesos

-Storm

-Samza

数据提取

到目前为止可谓一切顺利：存储层已经设计完成，资源管理机制设置妥当，而各任务亦经过配置。接下来惟一要做的就是数据处理工作了。

假定输入数据将以极高速率涌来，这时端点要顺利应对就需要满足以下要求：

-提供高吞吐能力/低延迟

-具备弹性

-可轻松实现规模扩展

-支持背压

背压能力并非必需，不过将其作为选项来应对负载峰值是个不错的选择。

Akka能够完美支持以上要求，而且基本上其设计目标恰好是提供这套功能集。下面来看Akka的特性：

-JVM面向JVM的角色模型实现能力

-基于消息且支持异步架构

-强制执行非共享可变状态

-可轻松由单一进程扩展至设备集群

-利用自上而下之监督机制实现角色层级

-不仅是并发框架：akka-http、akka-stream以及akka-persistence

以下简要示例展示了三个负责处理JSON HttpRequest的角色，它们将该请求解析为域模型例类，并将其保存在Cassandra当中：

class HttpActor extends Actor {      def receive = {    case req: HttpRequest =>      system.actorOf(Props[JsonParserActor]) ! req.body    case e: Event =>     system.actorOf(Props[CassandraWriterActor]) ! e    }    }        class JsonParserActor extends Actor {      def receive = {    case s: String => Try(Json.parse(s).as[Event]) match {     case Failure(ex) => //error handling code     case Success(event) => sender ! event    }    }    }        class CassandraWriterActor extends Actor with ActorLogging {      //for demo purposes, session initialized here    val session = Cluster.builder()    .addContactPoint("cassandra.host")    .build()    .connect()        override def receive: Receive = {    case event: Event =>     val statement = new SimpleStatement(event.createQuery)       .setConsistencyLevel(ConsistencyLevel.QUORUM)         Try(session.execute(statement)) match {       case Failure(ex) => //error handling code       case Success => sender ! WriteSuccessfull     }    }

}

看起来只需几行代码即可实现上述目标，不过利用Akka向Cassandra当中写入原始数据（即事件）却有可能带来以下问题：

-Cassandra的设计思路仍然偏重高速交付而非批量处理，因此必须对输入数据进行预汇聚。

-汇聚/汇总所带来的计算时间会随着数据总量的增长而逐步加长。

-由于采用无状态设计模式，各角色并不适合用于执行汇聚任务。

-微批量机制能够在一定程度上解决这个难题。

-仍然需要为原始数据提供某种可靠的缓冲机制

Kafka充当输入数据之缓冲机制

为了保留输入数据并对其进行预汇聚/处理，我们也可以使用某种类型的分布式提交日志机制。在以下用例中，消费程序将批量读取数据，对其进行处理并将其以预汇聚形式保存在Cassandra当中。该示例说明了如何利用akka-http通过HTTP将JSON数据发布至Kafka当中：

val config = new ProducerConfig(KafkaConfig())      lazy val producer = new KafkaProducer[A, A](config)      val topic = “raw_events”        val routes: Route = {      post{    decodeRequest{     entity(as[String]){ str =>       JsonParser.parse(str).validate[Event] match {         case s: JsSuccess[String] => producer.send(new KeyedMessage(topic, str))         case e: JsError => BadRequest -> JsError.toFlatJson(e).toString()       }     }    }    }        }        object AkkaHttpMicroservice extends App with Service {      Http().bindAndHandle(routes, config.getString("http.interface"), config.getInt("http.port"))

}

数据消费：Spark Streaming

尽管Akka也能够用于消耗来自Kafka的流数据，但将Spark纳入生态系统以引入Spark Streaming能够切实解决以下难题：

-其支持多种数据源

-提供“至少一次”语义

-可在配合Kafka Direct与幂等存储实现“仅一次”语义

以下代码示例阐述了如何利用Spark Streaming消费来自Kinesis的事件流：

val ssc = new StreamingContext(conf, Seconds(10))        val kinesisStream = KinesisUtils.createStream(ssc,appName,streamName,      endpointURL,regionName, InitialPositionInStream.LATEST,         Duration(checkpointInterval), StorageLevel.MEMORY_ONLY)    }        //transforming given stream to Event and saving to C*    kinesisStream.map(JsonUtils.byteArrayToEvent)      .saveToCassandra(keyspace, table)        ssc.start()      ssc.awaitTermination()  

故障设计：备份与补丁安装

通常来讲，故障设计是任何系统当中最为枯燥的部分，但其重要性显然不容质疑——当数据中心不可用或者需要对崩溃状况加以分析时，尽可能保障数据免于丢失可谓至关重要。

那么为什么要将数据存储在Kafka/Kinesis当中？截至目前，Kinesis仍然是惟一在无需备份的情况下能够确保全部处理结果丢失后保留数据的解决方案。虽然Kafka也能够支持数据长期保留，但硬件持有成本仍是个需要认真考虑的问题，因为S3存储服务的使用成本要远低于支持Kafka所需要的大量实例——另外，S3也提供非常理想的服务水平协议。

除了备份能力，恢复/补丁安装策略还应当考虑到前期与测试需求，从而保证任何与数据相关的问题能够得到迅速解决。程序员们在汇聚任务或者重复数据删除操作中可能不慎破坏计算结果，因此修复这类错误的能力就变得非常关键。简化这类操作任务的一种简便方式在于在数据模型当中引入幂等机制，这样同一操作的多次重复将产生相同的结果（例如SQL更新属于幂等操作，而计数递增则不属于）。

以下示例为Spark任务读取S3备份并将其载入至Cassandra：

val sc = new SparkContext(conf)        sc.textFile(s"s3n://bucket/2015/*/*.gz")      .map(s => Try(JsonUtils.stringToEvent(s)))    .filter(_.isSuccess).map(_.get)    .saveToCassandra(config.keyspace, config.table)

宏观构成

利用SMACK构建数据平台顶层设计

纵观全文，SMACK堆栈的卓越能力包括：

• 简明的工具储备以解决范围极广的各类数据处理场景
• 软件方案久经考验且拥有广泛普及度，背后亦具备强大的技术社区
• 易于实现规模伸缩与数据复制，且提供较低延迟水平
• 统一化集群管理以实现异构负载
• 可面向任意应用程序类型的单一平台
• 面向不同架构设计（批量、流数据、Lambda、Kappa）的实现平台
• 出色的产品发布速度（例如用于MVP验证）

</div>