1、开发前准备：假定您以搭建好了集群。

设置本地master，如果指定local的话，必须配置至少二条线程，也可通过sparkconf来设置，因为Spark Streaming应用程序在运行的时候，至少有一条线程用于不断的循环接收数据，并且至少有一条线程用于处理接收的数据（否则的话无法有线程用于处理数据），随着时间的推移，内存和磁盘都会不堪重负）。

对于集群而言，每隔exccutor一般肯定不只一个Thread，那对于处理Spark Streaming应用程序而言，每个executor一般分配多少core比较合适？根据我们过去的经验，5个左右的core是最佳的（段子:分配为奇数个core的表现最佳,例如：分配3个、5个、7个core等）

接下来，让我们开始动手写写代码吧！

　　我们将数据来源配置为本地端口9999（注意端口要求没有被占用）:

第四步：我们就像对RDD编程一样，基于DStream进行编程，原因是DStream是RDD产生的模板，在Spark Streaming发生计算前，其实质是把每个Batch的DStream的操作翻译成为了RDD操作。

除了print()方法将处理后的数据输出之外，还有其他的方法也非常重要，在开发中需要重点掌握，比如SaveAsTextFile,SaveAsHadoopFile等，最为重要的是foreachRDD方法，这个方法可以将数据写入,DB,DashBoard等，甚至可以随意的定义数据放在哪里，功能非常强大。

第四步，reduce操作：

第五步，print()等操作：

使用Spark Streaming可以处理各种数据来源类型，如：数据库、HDFS，服务器log日志、网络流，其强大超越了你想象不到的场景，只是很多时候大家不会用，其真正原因是对Spark、spark streaming本身不了解。

• 提供了各种创建DStream的方法接收各种流入的数据源（例如：Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等）；

• 将接收的数据流传入DStreams对象中；

• 通过Spark Streaming对象实例的start方法启动当前应用程序的流计算框架或通过stop方法结束当前应用程序的流计算框架；

三、Receiver功能及源码剖析：

1、Receiver代表数据的输入，接收外部输入的数据，如从Kafka上抓取数据；

4、Receiver是抽象类，其抓取数据的实现子类如下图所示：

5、  如果上述实现类都满足不了您的要求，您自己可以定义Receiver类，只需要继承Receiver抽象类来实现自己子类的业务需求。

数据数据在流进来之后最终会生成Job，最终还是基于Spark

使用Spark Streaming可以处理各种数据来源类型，如：数据库、HDFS，服务器log日志、网络流，其强大超越了你想象不到的场景，只是很多时候大家不会用，其真正原因是对Spark、spark streaming本身不了解。

1. 启动HDFS，如下图所示：

通过web端查看节点正常启动，如下图所示：

2.启动Spark集群，如下图所示：

通过web端查看集群启动正常，如下图所示：

此时Spark Streaming会每隔5秒执行一次，不断的扫描监控目录下是否有新的文件。

1. 而我们在具体实现的时候覆写了该方法，内部就是调用createContext方法来具体实现。上述实战案例中我们实现了createContext方法。

　　在开始和结束的时候每个batch都会进行checkpoint

流式处理中过一段时间数据就会被清理掉，但是可以通过remember可以延长数据在程序中的生命周期，另外延长RDD更长的时间。

假设数据流进来，进行ML或者Graphx的时候有时需要很长时间，但是bacth定时定条件的清除RDD，所以就可以通过remember使得数据可以延长更长时间。/**

1. 重新启动程序的时候，他会从曾经checkpoint的目录中，如果没有做额外配置的时候，所有的信息都会放在checkpoint的目录中(包括曾经应用程序信息)，因此下次再次启动的时候就会报错，无法初始化ShuffleDStream。

使用Spark Streaming可以处理各种数据来源类型，如：数据库、HDFS，服务器log日志、网络流，其强大超越了你想象不到的场景，只是很多时候大家不会用，其真正原因是对Spark、spark streaming本身不了解。

　　值得注意的是，Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source,Channel和Sink可以自由组合。组合方式基于用户设置的配置文件，非常灵活。比如：Channel可以把事件暂存在内存里，也可以持久化到本地硬盘上。Sink可以把日志写入HDFS,

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

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

Streaming流框架计算处理，流处理完成后再将最终结果发送给Kafka存储，将最终结果同时存储在（）、内存中间件（Redis、MemSQL）中，同时将最终结果通过Ganglia监控工具进行图形化展示，架构如下图：

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

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

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

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

2、个人强烈推荐在生产环境下用DirectAPI，但是我们的发行版，会对DirectAPI进行优化，降低其延迟。

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

使用Spark Streaming可以处理各种数据来源类型，如：数据库、HDFS，服务器log日志、网络流，其强大超越了你想象不到的场景，只是很多时候大家不会用，其真正原因是对Spark、spark streaming本身不了解。

第一部分，讲解Kafka的概念、和用例场景；

第二部分，讲解Kafka的安装和实战。

由于时间关系，今天的课程只讲到如何用官网的例子验证Kafka的安装是否成功。后续课程会接着讲解如何集成 Streaming和Kafka。

一、Kafka的概念、架构和用例场景

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。zookeeper和kafka和是不只能用于大数据的，集群启不启动，它都可以使用，也可以用于server普通的企业级平台上。

以帅哥和美女聊天为例，帅哥如何和美女交流呢？这中间通常想到的是、QQ、电话、邮件等通信媒介，这些通信媒介就是消息组件，帅哥把聊天信息发送给消息组件、消息组件将消息推送给美女，这就是常说的生产者、消费者模型，kafka不仅仅说是生产者消费者模式中广播的概念，也可以实现队列的方式，kafka的消费者中有一个group的概念，group中可以有很多实体，也可以只有一个实体，group中只有一个实体的话，就是队列的方式，所以从消息驱动的角度讲，它是广播的方式和队列的方式的完美结合体。而且在发送信息时可以将内容进行分类，即所谓的Topic主题。Kafka就是这样的通信组件，将不同对象组件粘合起来的纽带， 且是解耦合方式传递数据。

完善的流处理系统的特点：

1）能在线的以非常低的延迟，来处理数据，而且是稳定可靠的

2）能对流进来的数据进行非常复杂的分析，而不是简单的仅仅统计的分析

3）不仅能处理当前在线的数据，也能处理过去一天，一周，一个月甚至一年的数据

Apache Kafka与传统消息系统相比，有以下不同的特点：

• 分布式系统，易于向外扩展；

• 在线低延迟，同时为发布和订阅提供高吞吐量；

• 流进来的数据一般处理完后就消失了，也可以将消息存储到磁盘，因此可以处理1天甚至1周前内容，所以kafka不仅是一个消息中间件，还是一个存储系统

Kafka既然具备消息系统的基本功能，那么就必然会有组成消息系统的组件：

代表一种数据的类别或类型，工作、娱乐、生活有不同的Topic，生产者需要说明把说明数据分别放在那些Topic中，里面就是一个个小对象，并将数据数据推到Kafka，消费者获取数据是pull的过程。一组相同类型的消息数据流。这些消息在Kafka会被分区存放，并且有多个副本，以防数据丢失。每个分区的消息是顺序写入的，并且不可改写。

- Kafka Cluster（Kafka集群）：把推到Kafka系统的消息保存起来的一组服务器，也叫Broker。因为Kafka集群用到了Zookeeper作为底层支持框架，所以由一个选出的服务器作为Leader来处理所有消息的读和写的请求，其他服务器作为Follower接受Leader的广播同步备份数据，以备灾难恢复时用。

- Consumer（消费者）：从Kafka系统订阅消息的任何对象。

消费者可以有多个，并且某些消费者还可以组成Consumer Group。多个Consumer Group之间组成消息广播的关系，所以各个Group可以拉相同的消息数据。在Consumer Group内部，各消费者之间对Consumer Group拉出来的消息数据是队列先进先出的关系，某个消息数据只能给该Group的一个消费者使用，同一个Group中的实体是互斥的，对一个消息，这样是避免重复消费。如果有多个group，每个group中只有一个实体，这就是队列的方式了，因为它是互斥的。如果不是一个实体，则是广播模式，如下图所示，广播只能广播给一个group中的一个消费实体

kafka的数据传输是基于kernel（内核）级别的（传输速度接近0拷贝-ZeroCopy）、没有用户空间的参与。本身是软件，软件启动时第一个启动进程叫init，在init进程启动后会进入用户空间；kafka是用java写的，是基于jvm虚拟机的。例如：在分布式系统中，机器A上的应用程序需要读取机器B上的Java服务数据，由于Java程序对应的JVM是用户空间级别而且数据在磁盘上，A上应用程序读取数据时会首先进入机器B上的内核空间再进入机器B的用户空间，读取用户空间的数据后，数据再经过B机器上的内核空间分发到网络中(之所以要再经过B的内核，因为要通过网络通信，不通过内核，哪里来的网络通信)，机器A网卡接收到传输过来的数据后再将数据写入A机器的内核空间，从而最终将数据传输给A的用户空间进行处理。如下图：网络本身是一种硬件，磁盘只是硬件的一种。

正常情况下，外部系统从Java程序中读取数据，传输给内核空间并依赖网卡将数据写入到网络中，从而把数据传输出去。其实Java本身是内核的一层外衣， Socket编程，操作的各种数据都是在JVM的用户空间中进行的。而Kafka操作数据是放在内核空间的，通常内核空间处理数据的速度比用户空间快上万倍，因为没用用户态和内核态的切换，所以通过kafka可以实现高速读、写数据。只要磁盘空间足够大，可以无限量的存储数据，kafka的数据就是存储在磁盘中的，不是存在内核中的。而很多消息组件是把数据存内存中的。kafka用zookeeperg管理元数据，而且按顺序写数据，比随机写要快很多。又有副本！

类似，手机和邮箱等等这样大家熟悉的消息组件，Kafka也可以：

从内容消费的角度，Kafka把邮箱中的邮件类型看成是Topic。

二、Kafka的安装和实战

进入，你可以选择其他镜像网址去下载，用官网推荐的镜像：。提示：可以直接下载群里的Zookeeper安装文件。

提示：下面的步骤发生在master服务器。

以ubuntu14.04举例，把下载好的文件放到/root目录，用下面的命令解压：

解压后在/root目录会多出一个zookeeper-3.4.6的新目录，用下面的命令把它剪切到指定目录即安装好Zookeeper了：

提示：下面的步骤发生在master服务器。

# 配置Zookeeper的日志和服务器身份证号等数据存放的目录。

# 千万不要用默认的/tmp/zookeeper目录，因为/tmp目录的数据容易被意外删除。

# 在文件最后新增3行配置每个服务器的2个重要端口：Leader端口和选举端口

# server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器；

# C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；

# D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，

# 选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

# 如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信

# 端口号不能一样，所以要给它们分配不同的端口号。

1. 创建并打开myid文件

提示：下面的步骤发生在master服务器。

以ubuntu14.04举例，把下载好的文件放到/root目录，用下面的命令解压：

解压后在/root目录会多出一个kafka_2.10-0.9.0.1的新目录，用下面的命令把它剪切到指定目录即安装好Kafka了：

提示：下面的步骤发生在master服务器。

过一会儿，你会看到打印的消息内容：

至此，代表Kafka已经安装和配置成功。

使用Spark Streaming可以处理各种数据来源类型，如：数据库、HDFS，服务器log日志、网络流，其强大超越了你想象不到的场景，只是很多时候大家不会用，其真正原因是对Spark、spark streaming本身不了解。