对比点

Storm

Spark Streaming

实时计算模型

纯实时，来一条数据，处理一条数据

准实时，对一个时间段内的数据收集起来，作为一个RDD，再进行处理

实时计算延迟度

毫秒级

秒级

吞吐量

低

高

事务机制

支持完整

支持，但是不够完善

健壮性/容错性

Zooleeper,Acker,非常强

Checkpoint,WAL，一般

动态调整并行度

支持

不支持

package com.spark.spark_streaming;

import java.util.Arrays;

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 scala.Tuple2;

/**

 * 实时wordcount程序

 */

public class WordCount {

public static void main(String[] args) {

//创建一个sparkconf对象

//但是这有一点不同，我们是要给他设置一个master属性，但是我们测试的时候使用local模式

//local后面必须跟一个方括号，里面填一个数子，数字代表了，我们用几个线程来执行我们的spark streaming程序

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordCount");

//创建一javastreamingContext对象

//该对象，就类似与spark core的javasparkcontext/ spark sql中的sqlcontext

//该对象处理接收一个batch interval 参数之外

//还必须接收一个batch interval参数，就是说，每收集多长时间的数据，划分为一个batch 进行处理

//这里设置为10秒

JavaStreamingContext jssc = new JavaStreamingContext(conf,Durations.seconds(10));

//首先，创建输入dstream,代表了一个从数据源（kafkasocket）来的持续不断的实时数据流

//数据流，JavaReceiverInputDStream代表了一个输入的dstream

JavaReceiverInputDStream<String> lines= jssc.socketTextStream("localhost", 9999);

//JavaReceiverInputDStream中每隔10秒就会有一个RDD，其中封装了这10秒发送过来的数据

//RDD的类型是string 即一行一行的文本

//使用spark core提供的算子，直接引用在dstream上

//在底层实际上会对DStream中的一个一个的RDD执行我们应用在DStream上的算子

//产生新的RDD，会作为新的dstream中的RDD

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

private static final long serialVersionUID = 1L;

@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>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

@Override

public Integer call(Integer v1, Integer v2) throws Exception {

return v1 + v2;

}

});

//每10秒钟发送到指定的端口上的数据，都会被lines DStream接收到

//然后lines DStream会把每10秒的数据也就是一行一行的文本，诸如hello world,封装成一个RDD

//然后就会对每10秒对应的RDD执行一系列的算子操作

//比如对lines RDD执行flatmap之后，得到words RDD 作为words Dstream的一个RDD

//以此类推，直到生成最后一个wordcount RDD，作为wordcounts DStream中的一个RDD

//此时就到了每10秒钟发过来的数据进行单词统计

///但是一定要注意，Spark Streaming的计算模型，就决定了，我们必自己来进行中间缓存的控制

//比如写到redis等缓存

//他的计算模型跟Storm是完全不同的，Storm是自己编写的一个一个的程序，运行在节点上

//相当于一个一个的对象，可以自己在对象中控制缓存

//但是spark本身是函数式编程的计算模型，所以，比如在words或pairs Dstream中没法在实例变量中进行缓存

//此时，就只能将最后的计算出的wordcounts中一个一个的RDD，写入外部的缓存，或者持久化到DB

wordCounts.print();

//对JavaStreamingContext做后续的处理

//必须调用JavaStreamingContext的start()方法，整个spark streaming application才会启动执行

//否则是不会执行的

jssc.start();

jssc.awaitTermination();

jssc.close();

}

}

package com.spark.spark_streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.sql.catalyst.expressions.Second

import org.apache.spark.streaming.Seconds

object WordCount {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount")

    val ssc = new StreamingContext(conf,Seconds(10));

    val lines = ssc.socketTextStream("localhost", 9999)

    val words = lines.flatMap(line => line.split(" "))

    val pairs = words.map(word => (word , 1))

    val wordCounts = pairs.reduceByKey(_+_)

    wordCounts.print()

    ssc.start()

    ssc.awaitTermination()

  }

}

package com.spark.spark_streaming;

import java.util.Arrays;

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.JavaStreamingContext;

import scala.Tuple2;

public class HDFSWordCount {

public static void main(String[] args) {

SparkConf conf = new SparkConf().setAppName("HDFSWordCount").setMaster("local[2]");

JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5));

JavaDStream<String> lines = jsc.textFileStream("hdfs://hadoop01:8020/spark_streaming/");

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

private static final long serialVersionUID = 1L;

@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>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

JavaPairDStream<String, Integer> wordcount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

private static final long serialVersionUID = 1L;

@Override

public Integer call(Integer v1, Integer v2) throws Exception {

return v1 + v2;

}

});

wordcount.print();

jsc.start();

jsc.awaitTermination();

jsc.close();

}

}

package com.spark.spark_streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.Seconds

object HDFSWordCount {

  val conf = new SparkConf().setMaster("local[2]").setAppName("HDFSWordCount")

  val ssc = new StreamingContext(conf,Seconds(5))

  val lines = ssc.textFileStream("hdfs://hadoop01:8020/spark_streaming")

  val words = lines.flatMap(line => line.split(" "))

  val pairs = words.map(word => (word,1))

  val count = pairs.reduceByKey(_+_)

  count.print();

  ssc.start()

  ssc.awaitTermination();

}

transformation

meaning

map

对传入的每一个元素，返回一个新的元素

flatMap

对传入的每一个元素，返回一个或多个元素

filter

对传入的元素返回true或false，返回true的元素被保留，返回false的元素被过滤

union

将两个DStream进行合并

count

返回元素的个数

reduce

对所有的values进行聚合

countByValue

对元素按照值进行分组，对每个组进行计数，最后返回<k,v>的格式

reduceByKey

对key对应的value进行聚合

cogroup

对两个DStream进行连接操作，一个key连接起来的两个RDD的数据，都会以Iterable<V>的形式，出现在一个Tuple中

join

对两个DStream进行join操作，每个连接起来的pair，最为新的DStream的RDD的一个元素

transform

对数据进行转换操作

UpdateStateByKey

为每一个key维护一份state，并进行更新（普通的实时计算中最有用的一种操作）

window

对滑动窗口数据执行操作（实时计算中最有特色的一种操作）

package com.spark.spark_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;

/**

 * 基于缓存的实时wordcount程序

 * @author Administrator

 *

 */

public class UpdateStateByKeyWordCount {

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("UpdateStateByKeyWordCount");

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

JavaReceiverInputDStream<String> lines = jsc.socketTextStream("localhost", 9999);

//第一点：如果要使用updateStateByKey算子，就必须设置checkpoint目录，开启checkpoint用checkpoint的机制

//这样的话，才能把每个key对应的state除了在内存中有，也要checkpoint一份

//因为要长期保存一份key的state的话，那么spark streaming是要求必须用checkpoint的，以便在内存数据丢失的时候，

//可以从checkpoint中恢复数据

//开启checkpoint机制，很简单，指点调用jsc的checkpoint方法，设置一个hdfs目录即可

jsc.checkpoint("hdfs://hadoop01:8020/spark_checkpoint");

//然后实现基础的wordcount逻辑

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

private static final long serialVersionUID = 1L;

@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>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

//到这里就不一样了，之前的话，就只直接是reduceBykey，然后就可以得到每个时间段batch对应的RDD，计算出单词计数

//但是，有个问题，你如果要统计每个单词的全局计数呢？

//就是说，统计出来，从程序开始带现在为止，一个单词出现的次数，那么就之前的方式不好实现

//就必须基于redis这种缓存，或者是MySQL这种DB，来实现累加

//但是。updateStateByKey,就可以实现直接通过spark维护每个单词的全局统计次数

JavaPairDStream<String, Integer> count = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {

//这里的optional相当于scala中的样例类，就是option

//它代表了一个值的存在的状态，可能存在，也可能不存在

//这里面两个参数

//实际上，对于每个单词，每次batch计算的时候，都会调用这个函数

//第一个参数：values相当于是这个batch中，这个key的新值，可能有多个

//比如说一个hello 可能有两个1 ,(hello ,1)(hello ,1)那么传入的是(1,1)

//第二个参数，就是指的是这个key之前的状态，state，其中泛型类型是你自己执行的

@Override

public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception {

//首先定义一个全局的单词计数

Integer newValue=0;

//其次，判断state是否存在，如果不存在，说明是一个key第一次出现

//如果存在，说明这个key之前统计过全局的次数了

if(state.isPresent()){

newValue=state.get();

}

//接着将本次新出现的值，都累加到newvalue上去，就是一个key目前的全局的统计次数

for (Integer value : values) {

newValue +=value;

}

return Optional.of(newValue);

}

});

//到这里为止，相当于是每一个batch过来后，计算到pairs DStream.就会执行全局的updateStateByKey算子

//updateStateByKey，返回javapairDStream，其实就是代表每个key的全局的统计。

//打印出来

count.print();

jsc.start();

jsc.awaitTermination();

jsc.close();

}

}

package com.spark.spark_streaming;

import java.util.ArrayList;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.function.Function;

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;

/**

 * 基于transform的实时黑名单过滤

 * @author Administrator

 *

 */

public class TransformBlacklist {

@SuppressWarnings("deprecation")

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("TransformBlacklist");

JavaStreamingContext jsc =new JavaStreamingContext(conf, Durations.seconds(5));

//用户对我们的网站上的广告点击，点击之后，进行实时计费

//但是对于那些无良商家刷广告的人，有一个黑名单，只要是黑名单中的用户点击的广告，我们就过滤掉

//这里的日志格式date username的方式

JavaReceiverInputDStream<String> adsClickLogDStream = jsc.socketTextStream("hadoop01", 9999);

//模拟黑名单RDD

List<Tuple2<String, Boolean>> blacklist = new ArrayList<Tuple2<String,Boolean>>();

blacklist.add(new Tuple2<String, Boolean>("tom", true));

final JavaPairRDD<String, Boolean> blacklistRDD = jsc.sc().parallelizePairs(blacklist);

JavaPairDStream<String, String> userAdsClickLogDStream = adsClickLogDStream.mapToPair(new PairFunction<String, String, String>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, String> call(String adsClickLog) throws Exception {

return new Tuple2<String, String>(adsClickLog.split(" ")[1],adsClickLog);

}

});

//然后，就可以执行join操作了，将每个batch的RDD与黑名单RDD进行join map filter等操作；

//实时进行过滤

JavaDStream<String> validAdsClickLogDStream = userAdsClickLogDStream.transform(new Function<JavaPairRDD<String,String>, JavaRDD<String>>() {

private static final long serialVersionUID = 1L;

@Override

public JavaRDD<String> call(JavaPairRDD<String, String> userAdsClickLogRDD) throws Exception {

//这为什么用左外连接

//因为并不是每个用户都存在于黑名单中

//所以直接使用join ，那么没有存在于黑名单中的数据会无法join到

//就给丢弃掉了

//所以，这里用leftOuterJoin，就是说哪怕一个user不在黑名单RDD中,没有join到

//也还是会被保存下来的

JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinedRDD = userAdsClickLogRDD.leftOuterJoin(blacklistRDD);

JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filteredRDD = joinedRDD.filter(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, Boolean>() {

private static final long serialVersionUID = 1L;

@Override

public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {

//这里的tuple就是每个用户，对用的访问日志和用户在黑名中的状态

if(tuple._2._2().isPresent() && tuple._2._2().get()){

return false;

}

return true;

}

});

//此时。filteredRDD就只剩下没有被黑名单过滤的用户点击了

//进行map操作，装换成我们想要的日志

JavaRDD<String> validAdsClickLogRDD = filteredRDD.map(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, String>() {

@Override

public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {

return tuple._2._1;

}

});

return validAdsClickLogRDD;

}

});

//有效的广告点击日志

//企业真实的场景是写入kafkaActiveMQ等中间件消息对列

//在开发一个专门的后台服务，最为广告计费服务，执行实时的广告计费，这里就是指导有效的广告点击

validAdsClickLogDStream.print();

//所以，要先对输入的数据，进行下一转换操作，变成（username, date username）

//以便于后面的每个batch RDD与定义好的黑名单RDD进行join操作

jsc.start();

jsc.awaitTermination();

jsc.close();

}

}

package com.spark.spark_streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.Seconds

object TransformBlacklist {

  def main(args: Array[String]): Unit = {

    val conf= new SparkConf().setMaster("local[2]").setAppName("TransformBlacklist")

    val ssc = new StreamingContext(conf,Seconds(10))

    val blacklist = Array(("tom",true))

    val blacklistRDD = ssc.sparkContext.parallelize(blacklist, 3)

    val adsClickLogDStream = ssc.socketTextStream("hadoop01", 9999)

    val userAdsClickLogDStream  = adsClickLogDStream.map(adsClickLog =>(adsClickLog.split(" ")(1),adsClickLog))

    val validAdsClickLogDStream = userAdsClickLogDStream.transform(userAdsClickLogRDD =>{

     val joinedRDD = userAdsClickLogRDD.leftOuterJoin(blacklistRDD)

     val filteredRDD = joinedRDD.filter(tuple => {

       if(tuple._2._2.getOrElse(false)){

          false

       }else{

         true

       }

     })

     val validAdsClickLogRDD = filteredRDD.map(tuple => tuple._2._1)

      validAdsClickLogRDD

    })

    validAdsClickLogDStream.print

    ssc.start()

    ssc.awaitTermination()

  }

}

transform

意义

window

对每个滑动窗口的数据执行自定义的计算

countByWindow

对每个滑动窗口的数据执行count操作

reduceByWindow

对每个滑动窗口的数据执行reduce操作

reduceByKeyAndWindow

对每个滑动窗口的数据执行reduceByKey操作

countByValueAndWindow

对每个滑动窗口的操作执行countByValue操作

package com.spark.spark_streaming;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.function.Function;

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 scala.Tuple2;

/**

 * 基于滑动窗口热点搜索词实时统计

 * @author Administrator

 *

 */

public class WindowHotWord {

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WindowHotWord");

JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(2));

//说明一下，这里的搜索日志的格式：

//leo hello

//tom world

JavaReceiverInputDStream<String> searchLogDStream = jsc.socketTextStream("hadoop01", 9999);

//将搜索日志转为只有一个所搜词即可

JavaDStream<String> searchWordDStream = searchLogDStream.map(new Function<String, String>() {

private static final long serialVersionUID = 1L;

@Override

public String call(String searchLog) throws Exception {

return searchLog.split(" ")[1];

}

});

//将搜索词映射为(searchWord,1)的tuple的格式

JavaPairDStream<String, Integer> searchWordPairDStream = searchWordDStream.mapToPair(new PairFunction<String, String, Integer>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

//针对(searchword,1)的tuple格式的DStream，执行reduceByKeyAndWindow，滑动窗口操作

//第二个参数是窗口的长度，这里是60秒

//第三个参数是滑动的间隔，这里是10秒

//也就是说每隔10秒钟，将最近60秒的数据作为一个窗口，进行内部的RDD聚合，然后统一对一个RDD进行后溪的计算

//所以说，这里的意思，就是之前的searchWordPairDStream为止，其实都是不会立即执行的

//而是只是放在那里，等待我们的滑动时间到了之后，十秒钟到了，会将之前60秒的RDD，因为一个batch间隔是5秒，

//所以之前60秒，就有12个RDD，给聚合起来。然后统一执行reduceByKey,

//所以。这里的reduceByKeyAndWindow是针对每个窗口执行计算的，而不是针对某个DStream中国的RDD

JavaPairDStream<String, Integer> seachWordCountDStream = searchWordPairDStream.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {

private static final long serialVersionUID = 1L;

@Override

public Integer call(Integer v1, Integer v2) throws Exception {

// TODO Auto-generated method stub

return v1+v2;

}

}, Durations.seconds(60),Durations.seconds(10));

//到这里为止，就已经可以做到每隔10秒，出来，将之前60秒的收集到的单词的统计次数

//执行transform操作，因为，一个窗口，就是一个60秒钟中的数据，会变成一个RDD，然后对这一个RDD

//根据每个搜索词出现的次数进行排序，然后获得排名前三的热点搜索词

JavaPairDStream<String, Integer> finalDStream = seachWordCountDStream.transformToPair(new Function<JavaPairRDD<String,Integer>, JavaPairRDD<String,Integer>>() {

private static final long serialVersionUID = 1L;

@Override

public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> searchWordCountRDD) throws Exception {

//将热点词和出现的频次进行反转

JavaPairRDD<Integer, String> countSearchWordsRDD = searchWordCountRDD.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {

@Override

public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {

return new Tuple2<Integer, String>(tuple._2, tuple._1);

}

});

//然后执行降序排序

JavaPairRDD<Integer, String> sortedCountSearchWordsRDD = countSearchWordsRDD.sortByKey(false);

//再次进行反转，变成（seachWord,count）的这种格式

JavaPairRDD<String, Integer> sortedSearchWordCountRDD = sortedCountSearchWordsRDD.mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {

@Override

public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception {

return new Tuple2<String, Integer>(tuple._2, tuple._1);

}

});

//然后用take，获取前三名的热点搜索词

List<Tuple2<String, Integer>> hotSearchWordCounts = sortedSearchWordCountRDD.take(3);

for (Tuple2<String, Integer> wordCount : hotSearchWordCounts) {

System.out.println(wordCount._1+":"+wordCount._2);

}

return sortedSearchWordCountRDD;

}

});

//这个无关紧要，只是为了触发job的执行，所以必须有output操作

finalDStream.print();

jsc.start();

jsc.awaitTermination();

jsc.close();

}

}

package com.spark.spark_streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.Seconds

object WindowHotWord {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("WindowHotWord").setMaster("local[2]")

    val ssc = new StreamingContext(conf,Seconds(5))

    val seachLogDStream  = ssc.socketTextStream("hadoop01", 9999)

    val seachWordsDStream = seachLogDStream.map(_.split(" ")(1))

    val seachWordsPairsDStream = seachWordsDStream.map(searchWord =>(searchWord,1))

    val seachWordCountsDStream=seachWordsPairsDStream.reduceByKeyAndWindow(

        (v1:Int,v2:Int) => v1 + v2,Seconds(60), Seconds(10))

    val finalDStream = seachWordCountsDStream.transform(searchWordCountsRDD =>{

      val countSeachWordsRDD = searchWordCountsRDD.map(tuple => (tuple._2,tuple._1))

      val sortedCountSeachWordsRDD =countSeachWordsRDD.sortByKey(false)

      val sortedWordCountsRDD = sortedCountSeachWordsRDD.map(tuple => (tuple._2,tuple._1))

      val top3SearchWordCountsRDD = sortedWordCountsRDD.take(3)

      for(tuple <-top3SearchWordCountsRDD ){

        println(tuple)

      }

      searchWordCountsRDD

    })

    finalDStream.print()

    ssc.start()

    ssc.awaitTermination()

  }

}

Output

意义

print

打印每个batch中的前10个元素，主要用于测试，或者是不需要执行什么output操作时，用于简单的触发一下job

saveAsTextFile(prefix,[suffix])

将每个batch的数据保存到文件中。每个batch的文件命名格式是prefix-time_in_mx[.suffix]

saveAsObjectFile

同上，但是将每个batch的数据一序列化对象的方式，保存到sequenceFile中

saveAsHadoopFile

同上，将数据保存到hadoop文件中

foreachRDD

最常用的output操作，遍历DStream中的每一个产生的RDD，进行处理，可以将每一个RDD中的数据写入到外部存储，比如文件，数据库，缓存等。通常在其中是针对RDD执行action操作的，比如foreach

package com.spark.spark_streaming;

import java.util.ArrayList;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.function.Function;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.sql.DataFrame;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.RowFactory;

import org.apache.spark.sql.hive.HiveContext;

import org.apache.spark.sql.types.DataTypes;

import org.apache.spark.sql.types.StructField;

import org.apache.spark.sql.types.StructType;

import org.apache.spark.streaming.Durations;

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 scala.Tuple2;

/**

 * 与spark sql整合使用，top3热门实时统计

 *

 * @author Administrator

 *

 */

public class Top3HotProduct {

@SuppressWarnings("deprecation")

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("Top3HotProduct");

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

// 首先看一下日志的格式

// username pruduct category

// 其实socket是为了方便测试，企业中使用的是kafka

// 获取输入数据流

JavaReceiverInputDStream<String> productClickLogsDStream = jsc.socketTextStream("localhost", 9999);

// 做一个映射(category_pruduct,1)的格式，可以使用window操作，进行reduceByKey操作，从而统计出每个种类商品的点击次数

JavaPairDStream<String, Integer> categoryPruductPairsDStream = productClickLogsDStream

.mapToPair(new PairFunction<String, String, Integer>() {

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String productClickLog) throws Exception {

return new Tuple2<String, Integer>(

productClickLog.split(" ")[2] + "_" + productClickLog.split(" ")[1], 1);

}

});

// 执行window操作

// 到这里，就可以做到，每隔10秒。对最近60秒的数据，执行reduceByKey操作

// 计算出来这60秒内，每个种类的每个商品的点击次数

JavaPairDStream<String, Integer> categoryProductCountDStream = categoryPruductPairsDStream

.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {

private static final long serialVersionUID = 1L;

@Override

public Integer call(Integer v1, Integer v2) throws Exception {

return v1 + v2;

}

}, Durations.seconds(60), Durations.seconds(10));

// 然后真对每个种类的每个商品的点击次数，foreachRDD，在内部使用sparksql执行top3热门商品的统计

categoryProductCountDStream.foreachRDD(new Function<JavaPairRDD<String, Integer>, Void>() {

private static final long serialVersionUID = 1L;

@Override

public Void call(JavaPairRDD<String, Integer> categoryProductCountsRDD) throws Exception {

// 将该RDD，转换为javaRDD<Row>的格式

JavaRDD<Row> categoryProductCountRowRDD = categoryProductCountsRDD

.map(new Function<Tuple2<String, Integer>, Row>() {

private static final long serialVersionUID = 1L;

@Override

public Row call(Tuple2<String, Integer> categoryProductCount) throws Exception {

String category = categoryProductCount._1.split("_")[0];

String product = categoryProductCount._1.split("_")[1];

Integer count = categoryProductCount._2;

return RowFactory.create(category, product, count);

}

});

// 然后执行dataframe转换

List<StructField> structFields = new ArrayList<StructField>();

structFields.add(DataTypes.createStructField("category", DataTypes.StringType, true));

structFields.add(DataTypes.createStructField("product", DataTypes.StringType, true));

structFields.add(DataTypes.createStructField("click_count", DataTypes.IntegerType, true));

StructType structType = DataTypes.createStructType(structFields);

HiveContext sqlContext = new HiveContext(categoryProductCountsRDD.context());

DataFrame categoryProductCountDF = sqlContext.createDataFrame(categoryProductCountRowRDD, structType);

// 将60秒内的每个种类的，每个商品的点击次数注册为一张零时表

categoryProductCountDF.registerTempTable("product_clock_log");

// 执行sql语句，真对临时表，统计出来每个种类下，点击次数排名前3 的热门商品

DataFrame top3ProductDF = sqlContext.sql("select category,product,click_count from "

+ "( select category,product,click_count,row_number() over (partition by category order by click_count desc) rank from product_clock_log"

+ ") tmp where rank <=3");

// 其实在企业中不是打印，应该是保存到redis缓存中，或者是MySQL中

// 然后配合J2EE系统，进行数据的展示和查询、图形报表

top3ProductDF.show();

return null;

}

});

jsc.start();

jsc.awaitTermination();

jsc.close();

}

}

package com.spark.spark_streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.Seconds

import org.apache.spark.sql.Row

import org.apache.spark.sql.types.StructType

import org.apache.spark.sql.types.StructField

import org.apache.spark.sql.types.StringType

import org.apache.spark.sql.types.IntegerType

import org.apache.spark.sql.hive.HiveContext

object Top3HotProduct {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("Top3HotProduct")

    val ssc = new StreamingContext(conf,Seconds(5))

    val productClickLogsDStream =ssc.socketTextStream("localhost", 9999)

    val categoryProductPairsDStream = productClickLogsDStream.map(productLog => (productLog.split(" ")(2)+"_"+productLog.split(" ")(2),1))

    val categoryProductCountsDStream = categoryProductPairsDStream.

    reduceByKeyAndWindow((v1 :Int ,v2 :Int) => v1 + v2, Seconds(60), Seconds(10))

    categoryProductCountsDStream.foreachRDD(categoryProductCountsRDD =>{

      val categoryProductCountsRowRDD = categoryProductCountsRDD.map(tuple => {

        val category = tuple._1.split("_")(0)

        val product =tuple._1.split(" ")(1)

        val count =tuple._2

        Row(category,product,count)

      })

      val structType = StructType(Array(

         StructField("category",StringType,true),

         StructField("product",StringType,true),

         StructField("click_count",IntegerType,true)

      ))

      val hiveContext = new HiveContext(categoryProductCountsRDD.context)

      val categpryProductCountDF =hiveContext.createDataFrame(categoryProductCountsRowRDD, structType)

      categpryProductCountDF.registerTempTable("product_click_log")

     val top3ProductDF = hiveContext.sql("select category,product,click_count from "

+ "( select category,product,click_count,row_number() over (partition by category order by click_count desc) rank from product_clock_log"

+ ") tmp where rank <=3")

      top3ProductDF.show()

    })

  }

}

JavaStreamingContextFactory jscf = new JavaStreamingContextFactory (){

public JavaStreamingContext create(){

   JavaStreamingContext  jsc = new JavaStreamingContext (..);

JavaDSream<String> lines =jscf.socketTextStream(...);

Jscf.chechpoint(checkPointDirectory);

Return jscf;

}

};

JavaStreamingContext context  = JavaStreamingContext.getOrCreate(checkpointDirectory,contextFatory);

Context.start();

context.awaitTermina();

    def functionToCtreateContext():StreamingContext ={

Val ssc = new StreamingContext(...)

Val lines = ssc.socketTextStream(...)

Ssc.checkpoint(checkpointDirectory)

ssc

}   

Val context = StreamingContext.getOrCreate(checkpointDirectory,functionToCtreateContext_)

Conntext.start()

context.awaitTermination()

部署场景

Worker失效

Driver失效

Spark1.0，没有WAL

1.可靠receiver，可以保证数据零丢失

2.不可靠receiver，可以保证零丢失数据

3.至少一次的语义

1.不可靠receiver，缓存的数据全部丢失

2.任何receiver，过去接收的所有的数据全部丢失

3.没有容错语义

Spark1.2，开启了WAL

1.可靠receiver，零数据丢失

2.至少一次的语义

1.可靠receiver和文件，零数据丢失