简述大数据实时处理框架

现如今，我们来到了数据时代，数据信息化与我们的生活与工作息息相关。此篇文章简述利用大数据框架，实时处理数据的流程与相关框架的介绍，主要包括：

• 数据实时处理的概念和意义
• 数据实时处理能做什么
• 数据实时处理架构简介
• 数据实时处理代码演示

数据实时处理的概念和意义

什么是数据实时处理呢？我个人对数据实时处理的理解为：数据从生成->实时采集->实时缓存存储->（准）实时计算->实时落地->实时展示->实时分析。这一个流程线下来，处理数据的速度在秒级甚至毫秒级。

数据实时处理有什么意义呢？我们得到数据可以进行数据分析，利用数据统计方法，从错综复杂的数据关系中梳理出事物的联系，比如发展趋势、影响因素、因果关系等。甚至建立一些BI，对一些数据的有用信息进行可视化呈现，并形成数据故事。

数据实时处理能做什么

数据的实时计算

何为数据的实时计算？我们从数据源端拿到数据，可能不尽如人意，我们想对得到的数据进行 ETL 操作、或者进行关联等等，那么我们就会用到数据的实时计算。目前主流的实时计算框架有 spark，storm，flink 等。

数据的实时落地

数据的实时落地，意思是将我们的源数据或者计算好的数据进行实时的存储。在大数据领域，推荐使用 HDFS，ES 等进行存储。

数据的实时展示与分析

我们拿到了数据，要会用数据的价值。数据的价值体现在数据中相互关联关系，或与历史关联，或能预测未来。我们实时得到数据，不仅能够利用前端框架进行实时展示，还可以对其中的一些数据进行算法训练，预测未来走势等。

example：

淘宝双 11 大屏，每年的双 11 是淘宝粉丝疯狂的日子。马云会在双 11 的当天在阿里总部竖起一面大的电子屏幕，展示淘宝这一天的成绩。例如成交额，访问人数，订单量，下单量，成交量等等。这个电子大屏的背后，就是用到的我们所说的数据的实时处理。首先，阿里的服务器遍布全国各地，这些服务器收集PC端、手机端等日志，上报到服务器，在服务上部署数据采集工具。接下来，由于数据量庞大，需要做数据的缓存缓冲处理。下一步，对原始日志进行实时的计算，比如筛选出上面所述的各个指标。最后，通过接口或者其他形式，进行前端屏幕的实时展示。

数据实时处理架构简介

接下来是我们介绍的重点，先放一张数据流程图：

• 数据采集端，选用目前采集数据的主流控件 flume。
• 数据缓冲缓存，选用分布式消息队列 kafka。
• 数据实时计算，选用 spark 计算引擎。
• 数据存储位置，选用分布式数据存储 ES。
• 其他，指从 ES 中拿到数据后进行可视化展示，数据分析等。

下面将分别简单的介绍下各个组件：

flume

flume 是一个分布式的数据收集系统,具有高可靠、高可用、事务管理、失败重启、聚合和传输等功能。数据处理速度快,完全可以用于生产环境。

flume 的核心概念有：event，agent，source，channel，sink

event

flume 的数据流由事件 (event) 贯穿始终。event 是 flume 的基本数据单位，它携带日志数据并且携带数据的头信息，这些 event 由 agent 外部的 source 生成，当 source 捕获事件后会进行特定的格式化，然后 source 会把事件推入 channel 中。可以把 channel 看作是一个缓冲区，它将保存事件直到 sink 处理完该事件。sink 负责持久化日志或者把事件推向另一个 source。

agent

flume 的核心是 agent。agent 是一个 java 进程,运行在日志收集端,通过 agent 接收日志,然后暂存起来,再发送到目的地。 每台机器运行一个 agent。 agent 里面可以包含多个 source，channel，sink。

source

source 是数据的收集端，负责将数据捕获后进行特殊的格式化，将数据封装到 event 里，然后将事件推入 channel 中。flume 提供了很多内置的 source，支持 avro,log4j,syslog 等等。如果内置的 source 无法满足环境的需求，flume 还支持自定义 source。

channel

channel 是连接 source 和 sink 的组件，大家可以将它看做一个数据的缓冲区（数据队列），它可以将事件暂存到内存中也可以持久化到本地磁盘上， 直到 sink 处理完该事件。两个较为常用的 channel，MemoryChannel 和 FileChannel。

sink

sink 从 channel 中取出事件，然后将数据发到别处，可以向文件系统、数据库、hadoop、kafka，也可以是其他 agent 的 source。

flume 的可靠性与可恢复性

• flume 的可靠性：当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume 提供了可靠性保障，收到数据首先写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送。
• flume 的可恢复性：可恢复性是靠 channel。

口述抽象，上两张官网贴图:

单个 agent 收集数据流程图

多个 agent 协作处理数据流程图

kafka

Kafka 是一个高吞吐量的分布式发布-订阅消息系统。企业中一般使用 kafka 做消息中间件，做缓冲缓存处理。需要 zookeeper 分布式协调组件管理。

kafka 的设计目标：

1. 提供优秀的消息持久化能力，对 TB 级以上数据也能保证常数时间的访问性能。
2. 高吞吐率。即使在非常廉价的机器上也能做到每台机每秒 100000 条消息的传输。
3. 支持 kafka server 间的消息分区，及分布式消费，同时保证每个 partition 内的消息顺序传输。
4. 同时支持离线数据处理和实时数据处理。

kafka 核心概念

• broker：消息中间件处理结点，一个 kafka 节点就是一个 broker，多个 broker 可以组成一个 kafka 集群。
• topic：主题，kafka 集群能够同时负责多个 topic 的分发。
• partition：topic 物理上的分组，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。
• offset：每个 partition 都由一系列有序的、不可变的消息组成，这些消息被连续的追加到 partition 中。partition 中的每个消息都有一个连续的序列号叫做 offset，用于 partition 唯一标识一条消息。
• producer：负责发布消息到 kafka broker。
• consumer：消息消费者，向 kafka broker读取消息的客户端。
• consumer group：每个 consumer 属于一个特定的 consumer group。

贴两张官网图

prodecer-broker-consumer

分区图

spark

spark 是一个分布式的计算框架，是我目前认为最火的计算框架。

spark，是一种"one stack to rulethem all"的大数据计算框架,期望使用一个技术栈就完美地解决大数据领域的各种计算任务。apache 官方,对 spark 的定义是:通用的大数据快速处理引擎（一“栈”式）。

spark组成

1. spark core 用于离线计算
2. spark sql 用于交互式查询
3. spark streaming，structed streaming 用于实时流式计算
4. spark MLlib 用于机器学习
5. spark GraphX 用于图计算

spark 特点

1. 速度快：spar k基于内存进行计算(当然也有部分计算基于磁盘,比如 shuffle)。
2. 容易上手开发：spark 的基于 rdd 的计算模型，比 hadoop 的基于 map-reduce 的计算模型要更加易于理解,更加易于上手开发,实现各种复杂功能。
3. 通用性：spark 提供的技术组件,可以一站式地完成大数据领域的离线批处理、交互式查询、流式计算、机器学习、图计算等常见的任务。
4. 与其他技术的完美集成：例如 hadoop，hdfs、hive、hbase 负责存储，yarn 负责资源调度，spark 负责大数据计算。
5. 极高的活跃度：spark 目前是 apache 的顶级项目,全世界有大量的优秀工程师是 spark 的 committer，并且世界上很多顶级的 IT 公司都在大规模地使用 spark。

贴个spark架构图

数据实时处理代码演示

搭建好各个集群环境

需要搭建 flume 集群，kafka 集群，es 集群，zookeeper 集群，由于本例 spark 是在本地模式运行，所以无需搭建 spark 集群。

配置好组件之间整合的配置文件

搭建好集群后，根据集群组件直接的整合关系，配置好配置文件。其中主要的配置为 flume 的配置，如下图：

可以看到，我们的 agent 的 source 为 r1，channel 为 c1，sink 为 k1，source 为我本地 nc 服务，收集日志时，只需要打开 9999 端口就可以把日志收集。channel 选择为 memory 内存模式。sink 为 kafka 的 topic8 主题。

开启各个集群进程

1. 开启 zookeeper 服务。其中 QuorumPeerMain 为 zookeeper 进程。
2. 开启 kafka 服务。
3. 开启 es 服务。
4. 开启 flume 服务。其中 Application 为 flume 进程。

创建好 es 对应 table

创建好 es 对应的表，表有三个字段，对应代码里面的 case class（代码随后贴上）。

代码如下:

package run import org.apache.kafka.common.serialization.StringDeserializer import org.apache.log4j.Logger import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.sql.SparkSession import org.apache.spark.streaming.dstream.DStream import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe import org.apache.spark.streaming.kafka010.KafkaUtils import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent import org.apache.spark.streaming.{Seconds, StreamingContext} import org.elasticsearch.spark.rdd.EsSpark /**
  * @author wangjx
  * 测试kafka数据进行统计  kafka自身维护offset（建议使用自定义维护方式维护偏移量）
  */ object SparkStreamingAutoOffsetKafka { //定义样例类 与es表对应 case class people(name:String,country:String,age:Int) def main(args: Array[String]): Unit = {
    val logger = Logger.getLogger(this.getClass); //spark 配置 val conf = new SparkConf().setAppName("SparkStreamingAutoOffsetKafka").setMaster("local[2]")
    conf.set("es.index.auto.create","true")
    conf.set("es.nodes","127.0.0.1")
    conf.set("es.port","9200") //spark streaming实时计算初始化 定义每10秒一个批次 准实时处理 企业一般都是准实时 比如每隔10秒统计近1分钟的数据等等 val ssc = new StreamingContext(conf, Seconds(10))
    val spark = SparkSession.builder()
      .config(conf)
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN"); //设置kafka参数 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "x:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "exactly-once", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean)
    ) //kafka主题 val topic = Set("kafka8") //从kafka获取数据 val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topic, kafkaParams)
    ) //具体的业务逻辑 val kafkaValue: DStream[String] = stream.flatMap(line=>Some(line.value()))
    val peopleStream = kafkaValue
      .map(_.split(":")) //形成people样例对象 .map(m=>people(m(0),m(1),m(2).toInt)) //存入ES peopleStream.foreachRDD(rdd =>{
      EsSpark.saveToEs(rdd, "people/man")
    }) //启动程序入口 ssc.start()
    ssc.awaitTermination()
  }
}