一、Spark Streaming on Polling from Flume实战
二、Spark Streaming on Polling from Flume源码
第一部分：
推模式(Flume push SparkStreaming) VS 拉模式（SparkStreaming poll Flume）
采用推模式：推模式的理解就是Flume作为缓存，存有数据。监听对应端口，如果服务可以链接，就将数据push过去。(简单，耦合要低)，缺点是SparkStreaming 程序没有启动的话，Flume端会报错，同时会导致Spark Streaming 程序来不及消费的情况。
采用拉模式：拉模式就是自己定义一个sink，SparkStreaming自己去channel里面取数据，根据自身条件去获取数据，稳定性好。
Flume poll 实战：
1.Flume poll 配置
进入http://spark.apache.org/docs/latest/streaming-flume-integration.html官网，下载
spark-streaming-flume-sink_2.10-1.6.0.jar、scala-library-2.10.5.jar、commons-lang3-3.3.2.jar三个包：
![]()
将下载后的三个jar包放入Flume安装lib目录：
![]()
配置Flume conf环境参数：
![]()
![]()
编写业务代码：
public class SparkStreamingPollDataFromFlume {
public static void main(String[] args) {
/*
* 第一步：配置SparkConf：
* 1，至少2条线程：因为Spark Streaming应用程序在运行的时候，至少有一条
* 线程用于不断的循环接收数据，并且至少有一条线程用于处理接受的数据（否则的话无法
* 有线程用于处理数据，随着时间的推移，内存和磁盘都会不堪重负）；
* 2，对于集群而言，每个Executor一般肯定不止一个Thread，那对于处理Spark Streaming的
* 应用程序而言，每个Executor一般分配多少Core比较合适？根据我们过去的经验，5个左右的
* Core是最佳的（一个段子分配为奇数个Core表现最佳，例如3个、5个、7个Core等）；
*/
SparkConf conf = new SparkConf().setAppName("SparkStreamingPollDataFromFlume").setMaster("local[2]");
/*
* 第二步：创建SparkStreamingContext：
* 1，这个是SparkStreaming应用程序所有功能的起始点和程序调度的核心
* SparkStreamingContext的构建可以基于SparkConf参数，也可基于持久化的SparkStreamingContext的内容
* 来恢复过来（典型的场景是Driver崩溃后重新启动，由于Spark Streaming具有连续7*24小时不间断运行的特征，
* 所有需要在Driver重新启动后继续上衣系的状态，此时的状态恢复需要基于曾经的Checkpoint）；
* 2，在一个Spark Streaming应用程序中可以创建若干个SparkStreamingContext对象，使用下一个SparkStreamingContext
* 之前需要把前面正在运行的SparkStreamingContext对象关闭掉，由此，我们获得一个重大的启发SparkStreaming框架也只是
* Spark Core上的一个应用程序而已，只不过Spark Streaming框架箱运行的话需要Spark工程师写业务逻辑处理代码；
*/
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(30));
/*
* 第三步：创建Spark Streaming输入数据来源input Stream：
* 1，数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等
* 2, 在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口
* 的数据（当然该端口服务首先必须存在）,并且在后续会根据业务需要不断的有数据产生(当然对于Spark Streaming
* 应用程序的运行而言，有无数据其处理流程都是一样的)；
* 3,如果经常在每间隔5秒钟没有数据的话不断的启动空的Job其实是会造成调度资源的浪费，因为并没有数据需要发生计算，所以
* 实例的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；
*/
JavaReceiverInputDStream lines = FlumeUtils.createPollingStream(jsc, "Master", 9999);
/*
* 第四步：接下来就像对于RDD编程一样基于DStream进行编程！！！原因是DStream是RDD产生的模板（或者说类），在Spark Streaming具体
* 发生计算前，其实质是把每个Batch的DStream的操作翻译成为对RDD的操作！！！
*对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
* 第4.1步：讲每一行的字符串拆分成单个的单词
*/
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<SparkFlumeEvent, String>() { //如果是Scala，由于SAM转换，所以可以写成val words = lines.flatMap { line => line.split(" ")}
@Override
public Iterable<String> call(SparkFlumeEvent event) throws Exception {
String line = new String(event.event().getBody().array());
return Arrays.asList(line.split(" "));
}
});
/*
* 第四步：对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
* 第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
*/
JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String word) throws Exception {
return new Tuple2<String, Integer>(word, 1);
}
});
/*
* 第四步：对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
* 第4.3步：在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
*/
JavaPairDStream<String, Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { //对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
/*
* 此处的print并不会直接出发Job的执行，因为现在的一切都是在Spark Streaming框架的控制之下的，对于Spark Streaming
* 而言具体是否触发真正的Job运行是基于设置的Duration时间间隔的
* 诸位一定要注意的是Spark Streaming应用程序要想执行具体的Job，对Dtream就必须有output Stream操作，
* output Stream有很多类型的函数触发，类print、saveAsTextFile、saveAsHadoopFiles等，最为重要的一个
* 方法是foraeachRDD,因为Spark Streaming处理的结果一般都会放在Redis、DB、DashBoard等上面，foreachRDD
* 主要就是用用来完成这些功能的，而且可以随意的自定义具体数据到底放在哪里！！！
*
*/
wordsCount.print();
/*
* Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于
* 接受应用程序本身或者Executor中的消息；
*/
jsc.start();
jsc.awaitTermination();
jsc.close();
}
启动HDFS集群：
![]()
启动运行Flume：
![]()
![]()
启动eclipse下的应用程序：
![]()
copy测试文件hellospark.txt到Flume flume-conf.properties配置文件中指定的/usr/local/flume/tmp/TestDir目录下：
![]()
![]()
隔30秒后可以在eclipse程序控制台中看到上传的文件单词统计结果。
第二部分：源码分析
1、创建createPollingStream （FlumeUtils.scala ）
注意：默认的存储方式是MEMORY_AND_DISK_SER_2
/**
* Creates an input stream that is to be used with the Spark Sink deployed on a Flume agent.
* This stream will poll the sink for data and will pull events as they are available.
* This stream will use a batch size of 1000 events and run 5 threads to pull data.
* @param hostname Address of the host on which the Spark Sink is running
* @param port Port of the host at which the Spark Sink is listening
* @param storageLevel Storage level to use for storing the received objects
*/
def createPollingStream(
ssc: StreamingContext,
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[SparkFlumeEvent] = {
createPollingStream(ssc, Seq(new InetSocketAddress(hostname, port)), storageLevel)
}
2、参数配置：默认的全局参数，private 级别配置无法修改
private val DEFAULT_POLLING_PARALLELISM = 5
private val DEFAULT_POLLING_BATCH_SIZE = 1000
/**
* Creates an input stream that is to be used with the Spark Sink deployed on a Flume agent.
* This stream will poll the sink for data and will pull events as they are available.
* This stream will use a batch size of 1000 events and run 5 threads to pull data.
* @param addresses List of InetSocketAddresses representing the hosts to connect to.
* @param storageLevel Storage level to use for storing the received objects
*/
def createPollingStream(
ssc: StreamingContext,
addresses: Seq[InetSocketAddress],
storageLevel: StorageLevel
): ReceiverInputDStream[SparkFlumeEvent] = {
createPollingStream(ssc, addresses, storageLevel,
DEFAULT_POLLING_BATCH_SIZE, DEFAULT_POLLING_PARALLELISM)
}
3、创建FlumePollingInputDstream对象
/**
* Creates an input stream that is to be used with the Spark Sink deployed on a Flume agent.
* This stream will poll the sink for data and will pull events as they are available.
* @param addresses List of InetSocketAddresses representing the hosts to connect to.
* @param maxBatchSize Maximum number of events to be pulled from the Spark sink in a
* single RPC call
* @param parallelism Number of concurrent requests this stream should send to the sink. Note
* that having a higher number of requests concurrently being pulled will
* result in this stream using more threads
* @param storageLevel Storage level to use for storing the received objects
*/
def createPollingStream(
ssc: StreamingContext,
addresses: Seq[InetSocketAddress],
storageLevel: StorageLevel,
maxBatchSize: Int,
parallelism: Int
): ReceiverInputDStream[SparkFlumeEvent] = {
new FlumePollingInputDStream[SparkFlumeEvent](ssc, addresses, maxBatchSize,
parallelism, storageLevel)
}
4、继承自ReceiverInputDstream并覆写getReciver方法，调用FlumePollingReciver接口
private[streaming] class FlumePollingInputDStream[T: ClassTag](
_ssc: StreamingContext,
val addresses: Seq[InetSocketAddress],
val maxBatchSize: Int,
val parallelism: Int,
storageLevel: StorageLevel
) extends ReceiverInputDStream[SparkFlumeEvent](_ssc) {
override def getReceiver(): Receiver[SparkFlumeEvent] = {
new FlumePollingReceiver(addresses, maxBatchSize, parallelism, storageLevel)
}
}
5、ReceiverInputDstream 构建了一个线程池，设置为后台线程；并使用lazy和工厂方法创建线程和NioClientSocket（NioClientSocket底层使用NettyServer的方式）
lazy val channelFactoryExecutor =
Executors.newCachedThreadPool(new ThreadFactoryBuilder().setDaemon(true).
setNameFormat("Flume Receiver Channel Thread - %d").build())
lazy val channelFactory =
new NioClientSocketChannelFactory(channelFactoryExecutor, channelFactoryExecutor)
6、receiverExecutor 内部也是线程池；connections是指链接分布式Flume集群的FlumeConnection实体句柄的个数，线程拿到实体句柄访问数据。
lazy val receiverExecutor = Executors.newFixedThreadPool(parallelism,
new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Flume Receiver Thread - %d").build())
private lazy val connections = new LinkedBlockingQueue[FlumeConnection]()
7、启动时创建NettyTransceiver，根据并行度(默认5个)循环提交FlumeBatchFetcher
override def onStart(): Unit = {
// Create the connections to each Flume agent.
addresses.foreach(host => {
val transceiver = new NettyTransceiver(host, channelFactory)
val client = SpecificRequestor.getClient(classOf[SparkFlumeProtocol.Callback], transceiver)
connections.add(new FlumeConnection(transceiver, client))
})
for (i <- 0 until parallelism) {
logInfo("Starting Flume Polling Receiver worker threads..")
// Threads that pull data from Flume.
receiverExecutor.submit(new FlumeBatchFetcher(this))
}
}
8、FlumeBatchFetcher run方法中从Receiver中获取connection链接句柄ack跟消息确认有关
def run(): Unit = {
while (!receiver.isStopped()) {
val connection = receiver.getConnections.poll()
val client = connection.client
var batchReceived = false
var seq: CharSequence = null
try {
getBatch(client) match {
case Some(eventBatch) =>
batchReceived = true
seq = eventBatch.getSequenceNumber
val events = toSparkFlumeEvents(eventBatch.getEvents)
if (store(events)) {
sendAck(client, seq)
} else {
sendNack(batchReceived, client, seq)
}
case None =>
}
} catch {
9、获取一批一批数据方法
/**
* Gets a batch of events from the specified client. This method does not handle any exceptions
* which will be propogated to the caller.
* @param client Client to get events from
* @return [[Some]] which contains the event batch if Flume sent any events back, else [[None]]
*/
private def getBatch(client: SparkFlumeProtocol.Callback): Option[EventBatch] = {
val eventBatch = client.getEventBatch(receiver.getMaxBatchSize)
if (!SparkSinkUtils.isErrorBatch(eventBatch)) {
// No error, proceed with processing data
logDebug(s"Received batch of ${eventBatch.getEvents.size} events with sequence " +
s"number: ${eventBatch.getSequenceNumber}")
Some(eventBatch)
} else {
logWarning("Did not receive events from Flume agent due to error on the Flume agent: " +
eventBatch.getErrorMsg)
None
}
}
1：SparkSteaming基于kafka获取数据的方式，主要有俩种，即Receiver和Derict，基于Receiver的方式，是sparkStreaming给我们提供了kafka访问的高层api的封装，而基于Direct的方式，就是直接访问，在sparkSteaming中直接去操作kafka中的数据，不需要前面的高层api的封装。而Direct的方式，可以对kafka进行更好的控制！同时性能也更好。
2：实际上做kafka receiver的时候，通过receiver来获取数据，这个时候，kafka receiver是使用的kafka高层次的comsumer api来实现的。receiver会从kafka中获取数据，然后把它存储到我们具体的Executor内存中。然后Spark streaming也就是driver中，会根据这获取到的数据，启动job去处理。
3：注意事项：
1）在通过kafka receiver去获取kafka的数据，在正在获取数据的过程中，这台机器有可能崩溃了。如果来不及做备份，数据就会丢失，切换到另外一台机器上，也没有相关数据。这时候，为了数据安全，采用WAL的方式。write ahead log，预写日志的方式会同步的将接收到的kafka数据，写入到分布式文件系统中。但是预写日志的方式消耗时间，所以存储时建议Memory_and_Disc，不要2.如果是写到hdfs，会自动做副本。如果是写到本地，这其实有个风险，就是如果这台机器崩溃了，再想恢复过来，这个是需要时间的。
2）：我们的kafka receiver接收数据的时候，通过线程或者多线程的方式，kafka中的topic是以partition的方式存在的。sparkstreaming中的kafka receiver接收kafka中topic中的数据，也是通过线程并发的方式去获取的不同的partition，例如用五条线程同时去读取kafka中的topics中的不同的partition数据，这时你这个读取数据的并发线程数，和RDD实际处理数据的并发线程数是没任何关系的。因为获取数据时都还没产生RDD呢。RDD是Driver端决定产生RDD的。
3）默认情况下，一个Executor中是不是只有一个receiver去接收kafka中的数据。那能不能多找一些Executor去更高的并发度，就是使用更多的机器去接收数据，当然可以，基于kafa的api去创建更多的Dstream就可以了。很多的Dstream接收kafka不同topics中的不同的数据，最后你计算的时候，再把他优联就行了。其实这是非常灵活的，因为可以自由的组合。
![]()
kafka + spark streaming 集群
前提：
spark 安装成功，spark 1.6.0
zookeeper 安装成功
kafka 安装成功
启动集群和zookeeper和kafka
步骤：
1：创建topic为test
kafka-topics.sh --create --zookeeper master1:2181,work1:2181,work2:2181 --replication-factor 3 --partitions 1 --topic test
在worker1中启动kafka 生产者：
root@worker1:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
![]()
在worker2中启动消费者：
root@worker2:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-consumer.sh --zookeeper master1:2181 --topic test
![]()
生产者生产的消息，消费者可以消费到。说明kafka集群没问题。进入下一步。
在master中启动spark-shell
./spark-shell --master local[2] --packages org.apache.spark:spark-streaming-kafka_2.10:1.6.0
笔者用的spark 是 1.6.0 ，读者根据自己版本调整。
shell中的逻辑代码（wordcount），启动完成，把下面代码直接丢进去:
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Durations, StreamingContext}
val ssc = new StreamingContext(sc, Durations.seconds(5))
// 第二个参数是zk集群信息，zk的client host:port，生动的说明了kafka读取数据获取offset
//等元数据等信息，是从zk里面获取的。所以要连zk
// 第三个参数是Consumer groupID，随便写的
//第4个参数是消费的topic，以及并发读取topic中Partition的线程数，这个Map指定了你
//要消费什么topic，以及怎么消费topic
KafkaUtils.createStream(ssc, "master:2181,worker1:2181,worker2:2181", "StreamingWordCountSelfKafkaScala", Map("test" -> 1)).map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
生产者再生产消息：
![]()
spark streaming的反应：
![]()
返回worker2查看消费者
![]()
可见，groupId不一样，相互之间没有互斥。
上述是使用 createStream 方式链接kafka
还有更高效的方式，请使用createDirectStream
参考：
http://spark.apache.org/docs/latest/streaming-kafka-integration.html
1：Direct方式特点：
1）Direct的方式是会直接操作kafka底层的元数据信息，这样如果计算失败了，可以把数据重新读一下，重新处理。即数据一定会被处理。拉数据，是RDD在执行的时候直接去拉数据。
2）由于直接操作的是kafka，kafka就相当于你底层的文件系统。这个时候能保证严格的事务一致性，即一定会被处理，而且只会被处理一次。而Receiver的方式则不能保证，因为Receiver和ZK中的数据可能不同步，Spark Streaming可能会重复消费数据，这个调优可以解决，但显然没有Direct方便。而Direct api直接是操作kafka的，spark streaming自己负责追踪消费这个数据的偏移量或者offset，并且自己保存到checkpoint，所以它的数据一定是同步的，一定不会被重复。即使重启也不会重复，因为checkpoint了，但是程序升级的时候，不能读取原先的checkpoint，面对升级checkpoint无效这个问题，怎么解决呢?升级的时候读取我指定的备份就可以了，即手动的指定checkpoint也是可以的，这就再次完美的确保了事务性，有且仅有一次的事务机制。那么怎么手动checkpoint呢？构建SparkStreaming的时候，有getorCreate这个api，它就会获取checkpoint的内容，具体指定下这个checkpoint在哪就好了。或者如下图：
![]()
而如果从checkpoint恢复后，如果数据累积太多处理不过来，怎么办?1）限速2）增强机器的处理能力3）放到数据缓冲池中。
3）由于底层是直接读数据，没有所谓的Receiver，直接是周期性(Batch Intervel)的查询kafka，处理数据的时候，我们会使用基于kafka原生的Consumer api来获取kafka中特定范围(offset范围)中的数据。这个时候，Direct Api访问kafka带来的一个显而易见的性能上的好处就是，如果你要读取多个partition，Spark也会创建RDD的partition，这个时候RDD的partition和kafka的partition是一致的。而Receiver的方式，这2个partition是没任何关系的。这个优势是你的RDD，其实本质上讲在底层读取kafka的时候，kafka的partition就相当于原先hdfs上的一个block。这就符合了数据本地性。RDD和kafka数据都在这边。所以读数据的地方，处理数据的地方和驱动数据处理的程序都在同样的机器上，这样就可以极大的提高性能。不足之处是由于RDD和kafka的patition是一对一的，想提高并行度就会比较麻烦。提高并行度还是repartition，即重新分区，因为产生shuffle，很耗时。这个问题，以后也许新版本可以自由配置比例，不是一对一。因为提高并行度，可以更好的利用集群的计算资源，这是很有意义的。
4）不需要开启wal机制，从数据零丢失的角度来看，极大的提升了效率，还至少能节省一倍的磁盘空间。从kafka获取数据，比从hdfs获取数据，因为zero copy的方式，速度肯定更快。
2：实战部分
kafka + spark streaming 集群
前提：
spark 安装成功，spark 1.6.0
zookeeper 安装成功
kafka 安装成功
步骤：
1：先启动三台机器上的ZK，然后三台机器同样启动kafka，
2：在kafka上创建topic test
3：在worker1中启动kafka 生产者：
root@worker1:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
![]()
在worker2中启动消费者：
root@worker2:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test
![]()
生产者生产的消息，消费者可以消费到。说明kafka集群没问题。进入下一步。
在master中启动spark-shell
./spark-shell --master local[2] --packages org.apache.spark:spark-streaming-kafka_2.10:1.6.0,org.apache.kafka:kafka_2.10:0.8.2.1
笔者用的spark 是 1.6.0 ，读者根据自己版本调整。
![]()
shell中的逻辑代码（wordcount）:
import org.apache.spark.SparkConf
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Durations, StreamingContext}
val ssc = new StreamingContext(sc, Durations.seconds(5))
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc
, Map("bootstrap.servers" -> "master:2181,worker1:2181,worker2:2181", "metadata.broker.list" -> "master:9092,worker1:9092,worker2:9092", "group.id" -> "StreamingWordCountSelfKafkaDirectStreamScala")
, Set("test")).map(t => t._2).flatMap(_.toString.split(" ")).map((_, 1)).reduceByKey(_ + _).print()
ssc.start()
生产者再生产消息：
![]()
spark streaming的反应：
![]()
返回worker2查看消费者
![]()
可见，groupId不一样，相互之间没有互斥。
上述是使用 createDirectStream 方式链接kafka，实际使用中，其实就是和Receiver在api以及api中参数上有不同，其它基本一样
参考：
http://spark.apache.org/docs/latest/streaming-kafka-integration.html
1.SparkStreaming中的Transforamtions
2.SparkStreaming中的状态管理
一.DStream就是一个RDD之上的一个抽象，DStream和时间结合起来就不断的触发产生RDD的实例，可以说我们对Dstream的操作就初步定义了对RDD的操作，只不过需要时间的间隔也就是internalbatch去激活这个模板，生成具体的RDD的实例和具体的job.
二.我们鼓励Repartition，更多的是把更多的partition变成更少的partition，进行流的碎片的整理，我们不太鼓励把更少的partition变成更多的partion，因为会牵扯shuffle。
三.DStream是离散流，离散流就没状态，除了计算每个时间间隔产生一个job，我们还有必要计算过去十分钟或者半个小时，所以这个时候我们需要维护这个状态。后台Spark提供了专门维护这个状态的函数updateStateByKey(func),即基于key，我们可以进行多个状态的维护。因为你可以把每一个时间间隔都做为一个状态，例如每一秒钟做为一个状态，我算下过去十分钟或者半个小时。值的更新就是通过传进来的func函数。
四.Transform
| transform(func) |
Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream. 编程的逻辑是作用于RDD |
Transform操作，允许任意的RDD和RDD的操作被应用在DStream上。他可以使这些RDD不容易暴露在DstreamAPI中。比如让两个batch产生join操作而不暴露在DstreamAPi中，然后你可以很容易的使用transform来做这。这将是非常有作用的，例如，能够将实时数据清理通过将输入的数据流和预先计算的垃圾信息过滤掉。
五.UpdateByKey
| updateStateByKey(func) |
Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values for the key. This can be used to maintain arbitrary state data for each key. |
UpdaeStateByKey的操作，允许你维护任意的不断通过新的信息来更新的状态。使用这个函数你必须遵守两个步骤
1.定义一个状态：这个状态可以是任意的数据类型
2.定义一个状态更新函数：怎么样去使用从一个数据流中产生的旧的状态和新的状态来更新出一个状态。
六.forecachRDD(func)
| foreachRDD(func) |
The most generic output operator that applies a function,func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function func is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs. |
mapWithState将流式的状态管理性能提高10倍以上
foreachRDD(func)中的函数func是作用于最后一个RDD,也就是结果RDD，如果RDD没有数据，就不需要进行操作，foreachRDD()可以将数据写在Redis/Hbase/数据库/具体文件中，foreachRDD是在Driver程序中执行的，func就是action。
七.updateStateByKey
val cogroupedRDD = parentRDD.cogroup(prevStateRDD, partitioner)
val stateRDD = cogroupedRDD.mapPartitions(finalFunc, preservePartitioning)
Some(stateRDD)
cogroup是性能的瓶颈，所有的老数据，过去的数据都要进行cogroup操作，即使新的数据pairedRDD只有一条记录，也要把所有的老记录都要进行cogroup操作。这时相当耗时的。理论上讲，只应对这条记录对应的key和历史的一批数据中对应的这个key进行更新操作就行了，而它更新全部的，99%的时间都是浪费和消耗。性能非常低。也会产生shuffle。而下面的MapWithState则只更新你必须要更新的，所以极大提升了性能。
MapWithState只需要更新你必须更新的，没有必要更新所有的记录，官方宣传这个api会把流式的状态管理性能提升10倍以上。
第一部分：
updateStateByKey它的主要功能是随着时间的流逝，在Spark Streaming中可以为每一个key可以通过CheckPoint来维护一份state状态，通过更新函数对该key的状态不断更新；在更新的时候，对每一个新批次的数据（batch）而言，Spark Streaming通过使用updateStateByKey为已经存在的key进行state的状态更新（对每个新出现的key，会同样执行state的更新函数操作）；但是如果通过更新函数对state更新后返回none的话，此时刻key对应的state状态会被删除掉，需要特别说明的是state可以是任意类型的数据结构，这就为我们的计算带来无限的想象空间；
非常重要：
如果要不断的更新每个key的state，就一定会涉及到状态的保存和容错，这个时候就需要开启checkpoint机制和功能，需要说明的是checkpoint可以保存一切可以存储在文件系统上的内容，例如：程序未处理的但已经拥有状态的数据。
虽然说DStream是流式处理，但是由于我们保存了前面处理的结果，所以我可以不断在历史的基础上进行次数的更新。
补充说明：
关于流式处理对历史状态进行保存和更新具有重大实用意义，例如进行广告点击全面的动态评估（动态评估就是既有历史的数据又有现在的数据）（投放广告和运营广告效果评估的价值意义，热点随时追踪、热力图）
案例实战源码：
1.编写源码：
ackage org.apache.spark.examples.streaming;
import Java.util.Arrays;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import com.google.common.base.Optional;
import scala.Tuple2;
public class UpdateStateByKeyDemo {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local[2]").
setAppName("UpdateStateByKeyDemo");
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
//报错解决办法做checkpoint,开启checkpoint机制，把checkpoint中的数据放在这里设置的目录中，这里必须做checkpoint
//checkpoint如果挂了，那就挂了。所以生产环境下一般放在HDFS中，因为checkpoint有三份副本，一份挂了，还有另外2份容错。每次都要checkpoint，是会耗性能的，后面可以改进
jsc.checkpoint("/usr/local/tmp/checkpoint");
/*
* 第三步：创建Spark Streaming输入数据来源input Stream：
* 1，数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等
* 2, 在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口
* 的数据（当然该端口服务首先必须存在）,并且在后续会根据业务需要不断的有数据产生(当然对于Spark Streaming
* 应用程序的运行而言，有无数据其处理流程都是一样的)；
* 3,如果经常在每间隔5秒钟没有数据的话不断的启动空的Job其实是会造成调度资源的浪费，因为并没有数据需要发生计算，所以
* 实例的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；
*/
JavaReceiverInputDStream lines = jsc.socketTextStream("Master", 9999);
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { //如果是Scala，由于SAM转换，所以可以写成val words = lines.flatMap { line => line.split(" ")}
@Override
public Iterable<String> call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});
JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String word) throws Exception {
return new Tuple2<String, Integer>(word, 1);
}
});
/*
*第4.3步：在这里是通过updateStateByKey来以Batch Interval为单位来对历史状态进行更新，
* 这是功能上的一个非常大的改进，否则的话需要完成同样的目的，就可能需要把数据保存在Redis、
* Tagyon或者HDFS或者HBase或者数据库中来不断的完成同样一个key的State更新，如果你对性能有极为苛刻的要求，
* 且数据量特别大的话，可以考虑把数据放在分布式的Redis或者Tachyon内存文件系统中，如精准的秒杀系统；
* 当然从Spark1.6.x开始可以尝试使用mapWithState，Spark2.X后mapWithState应该非常稳定了。这样就去除了cogroup的弊端
*/
//如果发现不识别报错，一般是导包导错了，这里就导错了Optional的包，搞了好久
JavaPairDStream<String, Integer> wordsCount = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { //对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）
@Override
public Optional<Integer> call(List<Integer> values, Optional<Integer> state)
throws Exception {
//第一个参数就是key传进来的数据，第二个参数是曾经已有的数据
Integer updatedValue = 0 ;//如果第一次，state没有，updatedValue为0，如果有，就获取
if(state.isPresent()){
updatedValue = state.get();
}
//遍历batch传进来的数据可以一直加，随着时间的流式会不断去累加相同key的value的结果。
for(Integer value: values){
updatedValue += value;
}
return Optional.of(updatedValue);//返回更新的值
}
});
/*
*此处的print并不会直接出发Job的执行，因为现在的一切都是在Spark Streaming框架的控制之下的，对于Spark Streaming
*而言具体是否触发真正的Job运行是基于设置的Duration时间间隔的
*诸位一定要注意的是Spark Streaming应用程序要想执行具体的Job，对Dtream就必须有output Stream操作，
*output Stream有很多类型的函数触发，类print、saveAsTextFile、saveAsHadoopFiles等，最为重要的一个
*方法是foraeachRDD,因为Spark Streaming处理的结果一般都会放在Redis、DB、DashBoard等上面，foreachRDD
*主要就是用用来完成这些功能的，而且可以随意的自定义具体数据到底放在哪里！！！
*/
wordsCount.print();
/*
* Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于
* 接受应用程序本身或者Executor中的消息；
*/
jsc.start();
jsc.awaitTermination();
jsc.close();
}
2.创建checkpoint目录：
jsc.checkpoint("/usr/local/tmp/checkpoint");
3. 在eclipse中通过run 方法启动main函数：
![]()
4.启动hdfs服务并发送nc -lk 9999请求：
![]()
继续输入Hello hdfs
结果就是累加后的结果，
(Hello,3)
(SPark,1)
(hdfs,2)
再次输入Hello SPark，下一个batch就会继续累加，(Hello,3)(SPark,2) (hdfs,2)的结果输出
5.查看checkpoint目录输出：因为是二进制
![]()
![]()
源码解析：
1.PairDStreamFunctions类：
/**
* Return a new "state" DStream where the state for each key is updated by applying
* the given function on the previous state of the key and the new values of each key.
* Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
* @param updateFunc State update function. If `this` function returns None, then
* corresponding state key-value pair will be eliminated.
* @tparam S State type
*/
def updateStateByKey[S: ClassTag](
updateFunc: (Seq[V], Option[S]) => Option[S]
): DStream[(K, S)] = ssc.withScope {
updateStateByKey(updateFunc, defaultPartitioner())
}
/**
* Return a new "state" DStream where the state for each key is updated by applying
* the given function on the previous state of the key and the new values of the key.
* org.apache.spark.Partitioner is used to control the partitioning of each RDD.
* @param updateFunc State update function. If `this` function returns None, then
* corresponding state key-value pair will be eliminated.
* @param partitioner Partitioner for controlling the partitioning of each RDD in the new
* DStream.
* @tparam S State type
*/
def updateStateByKey[S: ClassTag](
updateFunc: (Seq[V], Option[S]) => Option[S],
partitioner: Partitioner
): DStream[(K, S)] = ssc.withScope {
val cleanedUpdateF = sparkContext.clean(updateFunc)
val newUpdateFunc = (iterator: Iterator[(K, Seq[V], Option[S])]) => {
iterator.flatMap(t => cleanedUpdateF(t._2, t._3).map(s => (t._1, s)))
}
updateStateByKey(newUpdateFunc, partitioner, true)
}
/**
* Return a new "state" DStream where the state for each key is updated by applying
* the given function on the previous state of the key and the new values of each key.
* org.apache.spark.Partitioner is used to control the partitioning of each RDD.
* @param updateFunc State update function. Note, that this function may generate a different
* tuple with a different key than the input key. Therefore keys may be removed
* or added in this way. It is up to the developer to decide whether to
* remember the partitioner despite the key being changed.
* @param partitioner Partitioner for controlling the partitioning of each RDD in the new
* DStream
* @param rememberPartitioner Whether to remember the paritioner object in the generated RDDs.
* @tparam S State type
*/
def updateStateByKey[S: ClassTag](
updateFunc: (Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)],
partitioner: Partitioner,
rememberPartitioner: Boolean
): DStream[(K, S)] = ssc.withScope {
new StateDStream(self, ssc.sc.clean(updateFunc), partitioner, rememberPartitioner, None)
}
override def compute(validTime: Time): Option[RDD[(K, S)]] = {
// Try to get the previous state RDD
getOrCompute(validTime - slideDuration) match {
case Some(prevStateRDD) => { // If previous state RDD exists
// Try to get the parent RDD
parent.getOrCompute(validTime) match {
case Some(parentRDD) => { // If parent RDD exists, then compute as usual
computeUsingPreviousRDD (parentRDD, prevStateRDD)
}
case None => { // If parent RDD does not exist
// Re-apply the update function to the old state RDD
val updateFuncLocal = updateFunc
val finalFunc = (iterator: Iterator[(K, S)]) => {
val i = iterator.map(t => (t._1, Seq[V](), Option(t._2)))
updateFuncLocal(i)
}
val stateRDD = prevStateRDD.mapPartitions(finalFunc, preservePartitioning)
Some(stateRDD)
}
}
}
case None => { // If previous session RDD does not exist (first input data)
// Try to get the parent RDD
parent.getOrCompute(validTime) match {
case Some(parentRDD) => { // If parent RDD exists, then compute as usual
initialRDD match {
case None => {
// Define the function for the mapPartition operation on grouped RDD;
// first map the grouped tuple to tuples of required type,
// and then apply the update function
val updateFuncLocal = updateFunc
val finalFunc = (iterator : Iterator[(K, Iterable[V])]) => {
updateFuncLocal (iterator.map (tuple => (tuple._1, tuple._2.toSeq, None)))
}
val groupedRDD = parentRDD.groupByKey (partitioner)
val sessionRDD = groupedRDD.mapPartitions (finalFunc, preservePartitioning)
// logDebug("Generating state RDD for time " + validTime + " (first)")
Some (sessionRDD)
}
case Some (initialStateRDD) => {
computeUsingPreviousRDD(parentRDD, initialStateRDD)
}
}
}
case None => { // If parent RDD does not exist, then nothing to do!
// logDebug("Not generating state RDD (no previous state, no parent)")
None
}
}
}
}
