一.StreamingContext详解
有两种创建StreamingContext的方式：
val conf = new SparkConf().setAppName(appName).setMaster(master);
val ssc = new StreamingContext(conf, Seconds(1));
StreamingContext，还可以使用已有的SparkContext来创建
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(1));
appName，是用来在Spark UI上显示的应用名称。master，是一个Spark、Mesos或者Yarn集群的URL，或者是local[*]。
batch interval可以根据你的应用程序的延迟要求以及可用的集群资源情况来设置。
一个StreamingContext定义之后，必须做以下几件事情：
1、通过创建输入DStream来创建输入数据源。
2、通过对DStream定义transformation和output算子操作，来定义实时计算逻辑。
3、调用StreamingContext的start()方法，来开始实时处理数据。
4、调用StreamingContext的awaitTermination()方法，来等待应用程序的终止。可以使用CTRL+C手动停止，或者就是让它持续不断的运行进行计算。
5、也可以通过调用StreamingContext的stop()方法，来停止应用程序。
需要注意的要点：
1、只要一个StreamingContext启动之后，就不能再往其中添加任何计算逻辑了。比如执行start()方法之后，还给某个DStream执行一个算子。
2、一个StreamingContext停止之后，是肯定不能够重启的。调用stop()之后，不能再调用start()
3、一个JVM同时只能有一个StreamingContext启动。在你的应用程序中，不能创建两个StreamingContext。
4、调用stop()方法时，会同时停止内部的SparkContext，如果不希望如此，还希望后面继续使用SparkContext创建其他类型的Context，比如SQLContext，那么就用stop(false)。
5、一个SparkContext可以创建多个StreamingContext，只要上一个先用stop(false)停止，再创建下一个即可。
二.输入DStream和Receiver详解
输入DStream代表了来自数据源的输入数据流。在之前的wordcount例子中，lines就是一个输入DStream（JavaReceiverInputDStream），代表了从netcat（nc）服务接收到的数据流。
除了文件数据流之外，所有的输入DStream都会绑定一个Receiver对象，该对象是一个关键的组件，用来从数据源接收数据，并将其存储在Spark的内存中，以供后续处理。
Spark Streaming提供了两种内置的数据源支持:
**1、基础数据源：**StreamingContext API中直接提供了对这些数据源的支持，比如文件、socket、Akka Actor等。
**2、高级数据源：**诸如Kafka、Flume、Kinesis、Twitter等数据源，通过第三方工具类提供支持。这些数据源的使用，需要引用其依赖。
**3、自定义数据源：**我们可以自己定义数据源，来决定如何接受和存储数据。
要注意的是，如果你想要在实时计算应用中并行接收多条数据流，可以创建多个输入DStream。这样就会创建多个Receiver，从而并行地接收多个数据流。但是要注意的是，一个Spark Streaming Application的Executor，是一个长时间运行的任务，因此，它会独占分配给Spark Streaming Application的cpu core。从而只要Spark Streaming运行起来以后，这个节点上的cpu core，就没法给其他应用使用了。
使用本地模式，运行程序时，绝对不能用local或者local[1]，因为那样的话，只会给执行输入DStream的executor分配一个线程。而Spark Streaming底层的原理是，至少要有两条线程，一条线程用来分配给Receiver接收数据，一条线程用来处理接收到的数据。因此必须使用local[n]，n>=2的模式。
如果不设置Master，也就是直接将Spark Streaming应用提交到集群上运行，那么首先，必须要求集群节点上，有>1个cpu core，其次，给Spark Streaming的每个executor分配的core，必须>1，这样，才能保证分配到executor上运行的输入DStream，两条线程并行，一条运行Receiver，接收数据；一条处理数据。否则的话，只会接收数据，不会处理数据。
因此，基于此，特此声明，我们本系列课程所有的练习，都是基于local[2]的本地模式，因为我们的虚拟机上都只有一个1个cpu core。但是大家在实际企业工作中，机器肯定是不只一个cpu core的，现在都至少4核了。到时记得给每个executor的cpu core，设置为超过1个即可。
三. updateStateByKey
updateStateByKey操作，可以让我们为每个key维护一份state，并持续不断的更新该state。
1、首先，要定义一个state，可以是任意的数据类型；
2、其次，要定义state更新函数——指定一个函数如何使用之前的state和新值来更新state。
对于每个batch，Spark都会为每个之前已经存在的key去应用一次state更新函数，无论这个key在batch中是否有新的数据。如果state更新函数返回none，那么key对应的state就会被删除。
当然，对于每个新出现的key，也会执行state更新函数。
注意，updateStateByKey操作，要求必须开启Checkpoint机制。
案例：基于缓存的实时wordcount程序（在实际业务场景中，这个是非常有用的）
代码如下
1package com.hzk.sparkStreaming; 2 3import org.apache.spark.SparkConf; 4import org.apache.spark.api.java.Optional; 5import org.apache.spark.api.java.function.FlatMapFunction; 6import org.apache.spark.api.java.function.Function2; 7import org.apache.spark.api.java.function.PairFunction; 8import org.apache.spark.streaming.Durations; 9import org.apache.spark.streaming.api.java.JavaDStream; 10import org.apache.spark.streaming.api.java.JavaPairDStream; 11import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; 12import org.apache.spark.streaming.api.java.JavaStreamingContext; 13import scala.Tuple2; 14 15import java.util.Arrays; 16import java.util.Iterator; 17import java.util.List; 18 19/** 20 * 基于updateStateByKey算子实现缓存机制的实时wordcount程序 21 * @author Administrator 22 * 23 */ 24public class UpdateStateByKeyWordCount { 25 public static void main(String[] args) throws InterruptedException { 26 SparkConf conf = new SparkConf() 27 .setMaster("local[2]") 28 .setAppName("HDFSWordCount"); 29 JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5)); 30 // 第一点，如果要使用updateStateByKey算子，就必须设置一个checkpoint目录，开启checkpoint机制 31 // 这样的话才能把每个key对应的state除了在内存中有，那么是不是也要checkpoint一份 32 // 因为你要长期保存一份key的state的话，那么spark streaming是要求必须用checkpoint的，以便于在 33 // 内存数据丢失的时候，可以从checkpoint中恢复数据 34 // 开启checkpoint机制，很简单，只要调用jssc的checkpoint()方法，设置一个hdfs目录即可 35 jssc.checkpoint("hdfs://hadoop-001:9000/datas/checkpoint"); 36 JavaReceiverInputDStream<String> lines=jssc.socketTextStream("localhost",9999); 37 JavaDStream<String> words=lines.flatMap(new FlatMapFunction<String, String>() { 38 @Override 39 public Iterator<String> call(String s) throws Exception { 40 return Arrays.asList(s.split(" ")).iterator(); 41 } 42 }); 43 JavaPairDStream<String,Integer> pairs=words.mapToPair(new PairFunction<String, String, Integer>() { 44 @Override 45 public Tuple2<String, Integer> call(String s) throws Exception { 46 return new Tuple2<>(s,1); 47 } 48 }); 49 // 到了这里，就不一样了，之前的话，是不是直接就是pairs.reduceByKey 50 // 然后，就可以得到每个时间段的batch对应的RDD，计算出来的单词计数 51 // 然后，可以打印出那个时间段的单词计数 52 // 但是，有个问题，你如果要统计每个单词的全局的计数呢？ 53 // 就是说，统计出来，从程序启动开始，到现在为止，一个单词出现的次数，那么就之前的方式就不好实现 54 // 就必须基于redis这种缓存，或者是mysql这种db，来实现累加 55 // 但是，我们的updateStateByKey，就可以实现直接通过Spark维护一份每个单词的全局的统计次数 56 JavaPairDStream<String,Integer> wordCounts=pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { 57 // 这里的Optional，相当于Scala中的样例类，就是Option，可以这么理解 58 // 它代表了一个值的存在状态，可能存在，也可能不存在 59 // 这里两个参数 60 // 实际上，对于每个单词，每次batch计算的时候，都会调用这个函数 61 // 第一个参数，values，相当于是这个batch中，这个key的新的值，可能有多个吧 62 // 比如说一个hello，可能有2个1，(hello, 1) (hello, 1)，那么传入的是(1,1) 63 // 第二个参数，就是指的是这个key之前的状态，state，其中泛型的类型是你自己指定的 64 @Override 65 public Optional<Integer> call(List<Integer> integers, Optional<Integer> state) throws Exception { 66 // 首先定义一个全局的单词计数 67 Integer newValue = 0; 68 // 其次，判断，state是否存在，如果不存在，说明是一个key第一次出现 69 // 如果存在，说明这个key之前已经统计过全局的次数了 70 if (state.isPresent()){ 71 newValue=state.get(); 72 } 73 // 接着，将本次新出现的值，都累加到newValue上去，就是一个key目前的全局的统计 74 // 次数 75 for (Integer value:integers ) { 76 newValue+=value; 77 } 78 return Optional.of(newValue); 79 } 80 }); 81 // 到这里为止，相当于是，每个batch过来是，计算到pairs DStream，就会执行全局的updateStateByKey 82 // 算子，updateStateByKey返回的JavaPairDStream，其实就代表了每个key的全局的计数 83 // 打印出来 84 wordCounts.print(); 85 86 jssc.start(); 87 jssc.awaitTermination(); 88 jssc.close(); 89 } 90}
四.输入DStream之Kafka数据源实战（基于Direct的方式）
这种新的不基于Receiver的直接方式，是在Spark 1.3中引入的，从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后，这种方式会周期性地查询Kafka，来获得每个topic+partition的最新的offset，从而定义每个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。
这种方式有如下优点：
**1、简化并行读取：**如果要读取多个partition，不需要创建多个输入DStream然后对它们进行union操作。Spark会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。
**2、高性能：**如果要保证零数据丢失，在基于receiver的方式中，需要开启WAL机制。这种方式其实效率低下，因为数据实际上被复制了两份，Kafka自己本身就有高可靠的机制，会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不需要开启WAL机制，只要Kafka中作了数据的复制，那么就可以通过Kafka的副本进行恢复。
3、一次且仅一次的事务机制：
基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。
基于direct的方式，使用kafka的简单api，Spark Streaming自己就负责追踪消费的offset，并保存在checkpoint中。Spark自己一定是同步的，因此可以保证数据是消费一次且仅消费一次。
JavaPairReceiverInputDStream<String, String> directKafkaStream =
KafkaUtils.createDirectStream(streamingContext,
[key class], [value class], [key decoder class], [value decoder class],
[map of Kafka parameters], [set of topics to consume]);
4.WordCount程序代码如下
1package com.hzk.sparkStreaming; 2 3import kafka.serializer.StringDecoder; 4import org.apache.spark.SparkConf; 5import org.apache.spark.api.java.Optional; 6import org.apache.spark.api.java.function.FlatMapFunction; 7import org.apache.spark.api.java.function.Function2; 8import org.apache.spark.api.java.function.PairFunction; 9import org.apache.spark.streaming.Durations; 10import org.apache.spark.streaming.api.java.JavaDStream; 11import org.apache.spark.streaming.api.java.JavaPairDStream; 12import org.apache.spark.streaming.api.java.JavaPairInputDStream; 13import org.apache.spark.streaming.api.java.JavaStreamingContext; 14import org.apache.spark.streaming.kafka.KafkaUtils; 15import scala.Tuple2; 16 17import java.util.*; 18 19public class KafkaDirectWordCount { 20 public static void main(String[] args) throws InterruptedException { 21 SparkConf conf = new SparkConf() 22 .setMaster("local[2]") 23 .setAppName("KafkaDirectWordCount"); 24 JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5)); 25 jssc.checkpoint("hdfs://hadoop-001:9000/datas/checkpoint"); 26 // 首先，要创建一份kafka参数map 27 Map<String, String> kafkaParams = new HashMap<String, String>(); 28 kafkaParams.put("metadata.broker.list", 29 "hadoop-001:9092"); 30 31 // 然后，要创建一个set，里面放入，你要读取的topic 32 // 这个，就是我们所说的，它自己给你做的很好，可以并行读取多个topic 33 Set<String> topics = new HashSet<String>(); 34 topics.add("test"); 35 36 // 创建输入DStream 37 JavaPairInputDStream<String, String> lines = KafkaUtils.createDirectStream( 38 jssc, 39 String.class, 40 String.class, 41 StringDecoder.class, 42 StringDecoder.class, 43 kafkaParams, 44 topics); 45 46 // 执行wordcount操作 47 JavaDStream<String> words = lines.flatMap( 48 49 new FlatMapFunction<Tuple2<String,String>, String>() { 50 51 private static final long serialVersionUID = 1L; 52 53 @Override 54 public Iterator<String> call(Tuple2<String, String> tuple) 55 throws Exception { 56 return Arrays.asList(tuple._2.split(" ")).iterator(); 57 } 58 59 }); 60 61 JavaPairDStream<String, Integer> pairs = words.mapToPair( 62 63 new PairFunction<String, String, Integer>() { 64 65 private static final long serialVersionUID = 1L; 66 67 @Override 68 public Tuple2<String, Integer> call(String word) throws Exception { 69 return new Tuple2<String, Integer>(word, 1); 70 } 71 72 }); 73 74 JavaPairDStream<String, Integer> wordCounts = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { 75 @Override 76 public Optional<Integer> call(List<Integer> integers, Optional<Integer> integerOptional) throws Exception { 77 int value=0; 78 if (integerOptional.isPresent()){ 79 value=integerOptional.get(); 80 } 81 for (Integer integer:integers){ 82 value+=integer; 83 } 84 85 return Optional.of(value); 86 } 87 }); 88 89 wordCounts.print(); 90 91 jssc.start(); 92 jssc.awaitTermination(); 93 jssc.close(); 94 } 95}