IMF传奇行动第86课：Spark Streaming第五课：Spark Streaming数据源Flume实际案例分享



本课分三部分内容：

第一部分讲什么是Flume；

第二部分讲Flume+Kafka+Spark
Streaming应用场景；

第三部分讲Kafka数据写入Spark
Streaming有两种方式。

一、什么是Flume?

　　Flume
作为
Cloudera
开发的实时日志收集系统，受到了业界的认可与广泛应用。Flume
初始的发行版本目前被统称为
Flume OG（original
generation），属于
Cloudera。但随着
Flume
功能的扩展，Flume OG
代码工程臃肿、核心组件设计不合理、核心配置不标准等缺点暴露出来，尤其是在
Flume OG
的最后一个发行版本 0.94.0
中，日志传输不稳定的现象尤为严重，为了解决这些问题，2011
年 10
月 22
号，Cloudera
完成了
Flume-728，对
Flume
进行了里程碑式的改动：重构核心组件、核心配置以及代码架构，重构后的版本统称为
Flume NG（next
generation）；改动的另一原因是将
Flume
纳入 apache
旗下，Cloudera Flume
改名为 Apache Flume。

Flume的特点　　Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力
。

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

Flume的可靠性

　　当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障，从强到弱依次分别为：end-to-end（收到数据agent首先将event写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送）；Store
on failure（这也是scribe采用的策略，当数据接收方crash时，将数据写到本地，待恢复后，继续发送）；Besteffort（数据发送到接收方后，不会进行确认）。

Flume的可恢复性

　　还是靠Channel。推荐使用FileChannel，事件持久化在本地文件系统里(性能较差)。

Flume的一些核心概念

　　Agent

　　使用JVM
运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多sources和sinks。

1. Client
生产数据，运行在一个独立的线程。

2. Source
从Client收集数据，传递给Channel。

3. Sink
从Channel收集数据，运行在一个独立线程。

4. Channel
连接sources
和sinks
，这个有点像一个队列。

5. Events
可以是日志记录、 avro
对象等。

　　Flume以agent为最小的独立运行单位。一个agent就是一个JVM。单agent由Source、Sink和Channel三大组件构成，如下图：



　　值得注意的是，Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source，Channel和Sink可以自由组合。组合方式基于用户设置的配置文件，非常灵活。比如：Channel可以把事件暂存在内存里，也可以持久化到本地硬盘上。Sink可以把日志写入HDFS、HBase，甚至是另外一个Source等等。Flume支持用户建立多级流，也就是说，多个agent可以协同工作，并且支持Fan-in、Fan-out、ContextualRouting、Backup
Routes，这也正是其强大之处。如下图所示:

二、Flume+Kafka+Spark Streaming应用场景

1、Flume集群采集外部系统的业务信息，将采集后的信息发生到Kafka集群，最终提供Spark
Streaming流框架计算处理，流处理完成后再将最终结果发送给Kafka存储，架构如下图：



2、Flume集群采集外部系统的业务信息，将采集后的信息发生到Kafka集群，最终提供Spark
Streaming流框架计算处理，流处理完成后再将最终结果发送给Kafka存储，同时将最终结果通过Ganglia监控工具进行图形化展示，架构如下图：



3、我们如果对Spark改进的话，可以做的事情有：SparkStreaming
交互式的360度的可视化，Spark Streaming
交互式3D可视化UI；Flume集群采集外部系统的业务信息，将采集后的信息发生到Kafka集群，最终提供Spark
Streaming流框架计算处理，流处理完成后再将最终结果发送给Kafka存储，将最终结果同时存储在数据库（Mysql）、内存中间件（Redis、MemSQL）中，同时将最终结果通过Ganglia监控工具进行图形化展示。架构如下图：

三、Kafka数据写入SparkStreaming有两种方式：

1、一种是Receivers，这个方法使用了Receivers来接收数据，Receivers的实现使用到Kafka高层次的消费者API，对于所有的Receivers，接收到的数据将会保存在Spark 分布式的Executors中，然后由Spark
Streaming启动的Job来处理这些数据；然而，在默认的配置下，这种方法在失败的情况下会丢失数据，为了保证零数据丢失，你可以在SparkStreaming中使用WAL日志功能，这使得我们可以将接收到的数据保存到WAL中（WAL日志可以存储在HDFS上），所以在失败的时候，我们可以从WAL中恢复，而不至于丢失数据。

2、另一种是DirectAPI，产生数据和处理数据的时候是在两台机器上？其实是在同一台数据上，由于在一台机器上有Driver和Executor，所以这台机器要足够强悍。

Flume集群将采集的数据放到Kafka集群中，Spark
Streaming会实时在线的从Kafka集群中通过DirectAPI拿数据，可以通过Kafka中的topic+partition查询最新的偏移量（offset）来读取每个batch的数据，即使读取失败也可再根据偏移量来读取失败的数据，保证应用运行的稳定性和数据可靠性。

温馨提示：

1、Flume集群数据写入Kafka集群时可能会导致数据存放不均衡，即有些Kafka节点数据量很大、有些不大，后续会对分发数据进行自定义算法来解决数据存放不均衡问题。

2、强烈推荐在生产环境下用DirectAPI，但是我们可以进行改进，对DirectAPI进行优化，降低其延迟。

总结：

实际生产环境下，搜集分布式的日志以Kafka为核心。

课程笔记来源:

DT大数据梦工厂IMF传奇行动课程学员整理。YY直播永久课堂频道68917580每晚8点准时开课。

Lifeis short, you need spark!