SparkStreaming

很多重要的应用要处理大量在线流式数据， 并返结果，比如社交网络趋势追踪，网站指标统计，广告系统，可以使用Spark Streaming来处理

流计算的处理流程一般包含三个阶段：数据实时采集、数据实时计算、实时查询服务

Spark Streaming可整合多种输入数据源，如Kafka、Flume、HDFS，甚至是普通的TCP套接字。经处理后的数据可存储至文件系统、数据库，或显示在仪表盘里

Spark Streaming的 基本原理 是将实时输入数据流以时间片（秒级）为单位进行拆分，然后经Spark引擎以类似批处理的方式处理每个时间片数据

Storm和SparkStreaming对比

目前SparkStreaming用的越来越多，为什么呢？

易用性好

• 供很多高级算子， 实现复杂运算非常简单
• 流式API和批处理API很类似， 学习成本低
  平台统一
• 不需要维护两套系统分别用于批处理和流式处理
• 可以自由调用Spark的组件如Spark SQL & Mllib
  生态丰富
• 支持各种数据源和数据格式
• 社区活跃， 发展迅猛

Spark Streaming的运行步骤：

• Driver端会要求Executor端启动接收器Receiver，接收数据
• 每个Receiver都会负责一个inputDStream（比如从文件中读取数据的文件流，比如套接字流，或者从Kafka中读取的一个输入流等等）
• Receiver会把接收到的数据分解生多个block放在内存中，如果设置了多副本它会把数据备份，然后会把blocks的信息高速StreamingContext。
• 当一个批次的时间到了之后，StreamingContext会告诉SparkContext，让SparkContext把job分发到Eexcutor上去执行

编写Spark Streaming程序的基本步骤

1. 通过创建输入DStream来定义输入源
2. 通过对DStream应用转换操作和输出操作来定义流计算
3. 用streamingContext.start()来开始接收数据和处理流程
4. 通过streamingContext.awaitTermination()方法来等待处理结束（手动结束或因为错误而结束）
5. 可以通过streamingContext.stop()来手动结束流计算进程

//创建Spark Streaming的上下文
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf,
Durations.seconds(1));
//获取数据 这里是从socket中获取
JavaReceiverInputDStream<String> lines = ssc.socketTextStream(
args[0], Integer.parseInt(args[1]),
StorageLevels.MEMORY_AND_DISK_SER);
//对获取的数据进行转换
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String,
String>() {
@Override
public Iterator<String> call(String x) {
return Arrays.asList(SPACE.split(x)).iterator();
}
});
JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<>(s, 1);
}
}).reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
});
//输出结果
wordCounts.print();
//启动
ssc.start();
ssc.awaitTermination();

Spark Streaming的输入流的数据来源

JavaDStream<String> lines = streamingContext.textFileStream("/data/input");

JavaReceiverInputDStream<String> lines =
streamingContext.socketTextStream("master", 9999 )

以RDD队列方式创建数据源

streamingContext.queueStream(queueOfRDD)

kafka作为输入源在公司的大数据生态系统中，可以把Kafka作为数据交换枢纽，不同类型的分布式系统（关系数据库、NoSQL数据库、流处理系统、批处理系统等），可以统一接入到Kafka，实现和Hadoop各个组件之间的不同类型数据的实时高效交换

安装文件为Kafka_2.11-0.10.2.0.tgz，前面的2.11就是该Kafka所支持的Scala版本号，后面的0.10.2.0是Kafka自身的版本号

Spark Streaming提供了两种访问Kafka的API

Receiver-based Approach：最原始的API，启动若干特殊的Task（Receiver），从Kafka上拉取数据，保存成RDD供处

理

Direct Approach：新API，无需启动Receiver，每一轮Sparkjob直接从Kafka上读取数据

import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import scala.Tuple2;

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);

Collection<String> topics = Arrays.asList("topicA", "topicB");

JavaInputDStream<ConsumerRecord<String, String>> stream =
  KafkaUtils.createDirectStream(
    streamingContext,
    LocationStrategies.PreferConsistent(),
    ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
  );

stream.mapToPair(record -> new Tuple2<>(record.key(), record.value()));

• window(windowLength, slideInterval) 基于源DStream产生的窗口化的批数据，计算得到一个新的Dstream
• countByWindow(windowLength, slideInterval) 返回流中元素的一个滑动窗口数
• reduceByWindow(func, windowLength, slideInterval) 返回一个单元素流。利用函数func聚集滑动时间间隔的流的元素创建这个单元素流。函数func必须满足结合律，从而可以支持并行计算
• reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks]) 应用到一个(K,V)键值对组成的DStream上时，会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。注意：在默认情况下，这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数
• reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]) 更加高效的reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce”操作。但是，只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”（以InvFunc参数传入）
• countByValueAndWindow(windowLength, slideInterval, [numTasks]) 当应用到一个(K,V)键值对组成的DStream上，返回一个由(K,V)键值对组成的新的DStream。每个key的值都是它们在滑动窗口中出现的频率

SparkStreaming 一定要做CheckPoint

JavaStreamingContext ssc = JavaStreamingContext.getOrCreate(checkPointDir, new
Function0<JavaStreamingContext>() {
@Override
public JavaStreamingContext call() throws Exception {
return createContext(checkPointDir, host, port);
}
});

JavaMapWithStateDStream<String, Integer, Integer, Tuple2<String, Integer>>
stateWordCounts =
wordPairs.mapWithState(StateSpec.function(new Function3<String,
Optional<Integer>, State<Integer>, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> call(String word, Optional<Integer> one,
State<Integer> state) throws Exception {
Option<Integer> stateCount = state.getOption();
Integer sum = one.orElse(0);
if (stateCount.isDefined()) {
sum += stateCount.get();
}
state.update(sum);
return new Tuple2<String, Integer>(word, sum);
}
}));

words.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {
@Override
public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {
return rdd.distinct();
}
});