devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spark3.x入门到精通-阶段五(SparkStreaming详解原理&java&scala双语实战)_spark3 支持java语法嘛

作者：weixin_40725706 | 2024-07-14 13:23:40

赞

踩

spark3 支持java语法嘛

SparkStreaming

简介

Spark Streaming 是 Spark 的一个子模块，用于快速构建可扩展，高吞吐量，高容错的流处理程序。具有以下特点：

  • 通过高级 API 构建应用程序，简单易用；
  • 支持多种语言，如 Java，Scala 和 Python；
  • 良好的容错性，Spark Streaming 支持快速从失败中恢复丢失的操作状态；
  • 能够和 Spark 其他模块无缝集成，将流处理与批处理完美结合；
  • Spark Streaming 可以从 HDFS，Flume，Kafka，Twitter 和 ZeroMQ 读取数据，也支持自定义数据源。

DStream

Spark Streaming 提供称为离散流 (DStream) 的高级抽象，用于表示连续的数据流。 DStream 可以从来自 Kafka，Flume 和 Kinesis 等数据源的输入数据流创建，也可以由其他 DStream 转化而来。在内部，DStream 表示为一系列 RDD。 

Spark & Storm & Flink 

storm 和 Flink 都是真正意义上的流计算框架，但 Spark Streaming 只是将数据流进行极小粒度的拆分，拆分为多个批处理，使得其能够得到接近于流处理的效果，但其本质上还是批处理（或微批处理）。

开始使用

hello world

java版本

  1. public class JavaHelloWord {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         //当使用Local模式启动Spark时，master URL必须为"local[n]"，且"n"的值必须大于"receivers"的数量：
  5.         //否则会出现Expecting 1 replicas with only 0 处理
  6.         sparkConf.setMaster("local[2]")
  7.                 .setAppName("JavaHelloWord");
  8.  
  9.         //Durations.seconds(1)表示每一秒拉去一次数据
  10.         //每一秒就是一个batch，每一个batch里面会有多个rdd
  11.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  12.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9999);
  13.  
  14.         //对于RDD里面每一个元素处理，这里的每一个元素是一行数据，其实中间的操作就是对于RDD里面元素的处理
  15.         JavaDStream<String> wordDStream = socketTextStream.flatMap(new FlatMapFunction<String, String>() {
  16.             @Override
  17.             public Iterator<String> call(String s) throws Exception {
  18.                 System.out.println(s);
  19.                 return Arrays.asList(s.split(" ")).iterator();
  20.             }
  21.         });
  22.  
  23.         JavaPairDStream<String, Integer> wordCountDStream = wordDStream.mapToPair(new PairFunction<String, String, Integer>() {
  24.             @Override
  25.             public Tuple2<String, Integer> call(String word) throws Exception {
  26.                 return new Tuple2<String, Integer>(word, 1);
  27.             }
  28.         });
  29.  
  30.         JavaPairDStream<String, Integer> resultDStream = wordCountDStream.reduceByKey(new Function2<Integer, Integer, Integer>() {
  31.             @Override
  32.             public Integer call(Integer v1, Integer v2) throws Exception {
  33.                 return v1 + v2;
  34.             }
  35.         });
  36.  
  37.         resultDStream.foreachRDD(
  38.                 //这里对于每一batch所有的RDD进行分布式的发送到集群其他机器上面去执行
  39.                 rdd->{
  40.                     rdd.foreach(new VoidFunction<Tuple2<String, Integer>>() {
  41.                         //对于RDD里面每一个分区里面的数据进行分布式计算
  42.                         @Override
  43.                         public void call(Tuple2<String, Integer> result) throws Exception {
  44.                             System.out.println(result);
  45.                         }
  46.                     });
  47.                 }
  48.         );
  49.  
  50.         javaStreamingContext.start();
  51.         javaStreamingContext.awaitTermination();
  52.         javaStreamingContext.close();
  53.     }
  54. }

打包后执行(下面是client提交，就会在提交的机器上面启动Driver,那么打印的时候就会直接在提交的窗口打印数据)

  1. bin/spark-submit \
  2. --master yarn \
  3. --deploy-mode client \
  4. --driver-memory 512m \
  5. --executor-memory 512m \
  6. --executor-cores 1 \
  7. --num-executors 2 \
  8. --queue default \
  9. --class com.zhang.one.javasparkstreaming.JavaHelloWord \
  10. /home/bigdata/spark/spark-yarn/original-sparkstart-1.0-SNAPSHOT.jar

下面用cluster提交(启动的Driver在集群的随机一台启动)

  1. bin/spark-submit \
  2. --master yarn \
  3. --deploy-mode cluster \
  4. --driver-memory 512m \
  5. --executor-memory 512m \
  6. --executor-cores 1 \
  7. --num-executors 2 \
  8. --queue default \
  9. --class com.zhang.one.javasparkstreaming.JavaHelloWord \
  10. /home/bigdata/spark/spark-yarn/original-sparkstart-1.0-SNAPSHOT.jar

下面是查看输出数据的步骤

 

 

scala版本

  1. object ScalaStreaingWc {
  2.   def main(args: Array[String]): Unit = {
  3.     val conf = new SparkConf()
  4.     conf.setAppName("ScalaStreaingWc")
  5.       .setMaster("local[2]")
  6.  
  7.     val sparkStreaming = new StreamingContext(conf, Seconds(1))
  8.     sparkStreaming.socketTextStream("master",9999)
  9.       .flatMap(_.split(" "))
  10.       .map((_,1))
  11.       .reduceByKey(_+_)
  12.       .foreachRDD(rdd=>rdd.foreach(println))
  13.  
  14.     sparkStreaming.start()
  15.     sparkStreaming.awaitTermination()
  16.     sparkStreaming.stop()
  17.   }
  18. }

SparkStreaming用Kafka作为数据源

Receiver模式(由于这种模式在spark3.x没有了所以就不写代码了)

关于上面为什么必须用local[*]图解，因为sparkstreaming会一直占用资源，必须有足够资源的前提它才能启动，下面以Reciver模式原理图解

Direct模式

java&scala双语实战

  • 有点kafka的partition和RDD的partiton有一一对应的关系
  • 能够保证spark仅仅消费一次kafka的数据

现在kafka创建对应的topic 

 ./kafka-topics.sh --bootstrap-server master:9092 --create --topic spark-streaming-topic --partitions 2 --replication-factor 2

创建一个生产者

./kafka-console-producer.sh --broker-list master:9092 --topic spark-streaming-topic

java版本 

  1. public class JavaKafkaDirectStream {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4. //对应kafka里面的分区一一对应
  5.         sparkConf.setMaster("local[2]")
  6.                 .setAppName("JavaKafkaDirectStream");
  7.  
  8.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  9.         Map<String, Object> params = new HashMap<>();
  10.         params.put("bootstrap.servers","master:9092");
  11.         params.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  12.         params.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  13.         params.put("group.id","spark-streaming-group");
  14.         params.put("auto.offset.reset","latest");
  15.         params.put("enable.auto.commit",true);
  16.         List<String> topics = Arrays.asList("spark-streaming-topic");
  17.         JavaInputDStream<ConsumerRecord<String, String>> kafkaDStream = KafkaUtils.createDirectStream(
  18.                 javaStreamingContext,
  19.                 LocationStrategies.PreferConsistent(),
  20.                 ConsumerStrategies.Subscribe(topics, params));
  21.  
  22.         JavaDStream<String> kafkaValueDStream = kafkaDStream.map((Function<ConsumerRecord<String, String>, String>) v1 -> v1.value());
  23.         JavaDStream<String> woldDStream = kafkaValueDStream.flatMap((FlatMapFunction<String, String>) s -> Arrays.asList(s.split(" ")).iterator());
  24.         JavaPairDStream<String, Integer> wordCountDStream = woldDStream.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<String, Integer>(s, 1));
  25.         JavaPairDStream<String, Integer> resultDStream = wordCountDStream.reduceByKey((Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2);
  26.         resultDStream.foreachRDD(rdd->rdd.foreach((VoidFunction<Tuple2<String, Integer>>) result -> System.out.println(result)));
  27.  
  28.         javaStreamingContext.start();
  29.         javaStreamingContext.awaitTermination();
  30.         javaStreamingContext.stop();
  31.     }
  32. }

scala版本

  1. object ScalaKafkaDirectStream {
  2.   def main(args: Array[String]): Unit = {
  3.  
  4.     val sparkConf = new SparkConf().setAppName("ScalaKafkaDirectStream").setMaster("local[2]")
  5.     val streamingContext = new StreamingContext(sparkConf, Seconds(5))
  6.  
  7.     val kafkaParams = Map[String, Object](
  8.       /*
  9.        * 指定 broker 的地址清单，清单里不需要包含所有的 broker 地址，生产者会从给定的 broker 里查找其他 broker 的信息。
  10.        * 不过建议至少提供两个 broker 的信息作为容错。
  11.        */
  12.       "bootstrap.servers" -> "master:9092",
  13.       /*键的序列化器*/
  14.       "key.deserializer" -> classOf[StringDeserializer],
  15.       /*值的序列化器*/
  16.       "value.deserializer" -> classOf[StringDeserializer],
  17.       /*消费者所在分组的 ID*/
  18.       "group.id" -> "spark-streaming-group",
  19.       /*
  20.        * 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理:
  21.        * latest: 在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
  22.        * earliest: 在偏移量无效的情况下，消费者将从起始位置读取分区的记录
  23.        */
  24.       "auto.offset.reset" -> "latest",
  25.       /*是否自动提交*/
  26.       "enable.auto.commit" -> (true: java.lang.Boolean)
  27.     )
  28.  
  29.     /*可以同时订阅多个主题*/
  30.     val topics = Array("spark-streaming-topic")
  31.     val stream = KafkaUtils.createDirectStream[String, String](
  32.       streamingContext,
  33.       /*位置策略*/
  34.       /**
  35.        * 位置策略
  36.        * Spark Streaming 中提供了如下三种位置策略，用于指定 Kafka 主题分区与 Spark 执行程序 Executors 之间的分配关系：
  37.           PreferConsistent : 它将在所有的 Executors 上均匀分配分区；
  38.           PreferBrokers : 当 Spark 的 Executor 与 Kafka Broker 在同一机器上时可以选择该选项，它优先将该 Broker 上的首领分区分配给该机器上的 Executor；
  39.           PreferFixed : 可以指定主题分区与特定主机的映射关系，显示地将分区分配到特定的主机.
  40.        */
  41.       PreferConsistent,
  42.       /*订阅主题*/
  43.       /**
  44.        *
  45.        * @param 需要订阅的主题的集合
  46.        * @param Kafka 消费者参数
  47.        * @param offsets(可选): 在初始启动时开始的偏移量。如果没有，则将使用保存的偏移量或 auto.offset.reset 属性的
  48.        *           def Subscribe[K, V](
  49.                        topics: ju.Collection[jl.String],
  50.                        kafkaParams: ju.Map[String, Object],
  51.                        offsets: ju.Map[TopicPartition, jl.Long]): ConsumerStrategy[K, V] = { ... }
  52.        */
  53.       Subscribe[String, String](topics, kafkaParams)
  54.     )
  55.  
  56.     /*打印输入流*/
  57.     stream.map(_.value())
  58.       .flatMap(_.split(" "))
  59.       .map((_,1))
  60.       .reduceByKey(_+_)
  61.       .foreachRDD(rdd=>rdd.foreach(println))
  62.  
  63.     streamingContext.start()
  64.     streamingContext.awaitTermination()
  65.   }
  66. }

常用的transform使用

updateStateByKey

  • 对于每一个key维护一个状态
  • 注意使用它的时候必须开始checkpoint(因为如果数据丢失了，还可以在磁盘中恢复数据)

java版本

  1. public class JavaUpdateStateByKey {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         sparkConf.setMaster("local[2]")
  5.                 .setAppName("JavaUpdateStateByKey");
  6.         //解决权限问题
  7.         System.setProperty("HADOOP_USER_NAME","bigdata");
  8.  
  9.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  10.         javaStreamingContext.checkpoint("hdfs://master:8020/sparkstreamingcheckpoing");
  11.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9998);
  12.  
  13.         JavaDStream<String> wordDStream = socketTextStream.flatMap(new FlatMapFunction<String, String>() {
  14.             @Override
  15.             public Iterator<String> call(String s) throws Exception {
  16.                 return Arrays.asList(s.split(" ")).iterator();
  17.             }
  18.         });
  19.  
  20.         //装换成key,value操作的
  21.         JavaPairDStream<String, Integer> wordCountDStream = wordDStream.mapToPair(new PairFunction<String, String, Integer>() {
  22.             @Override
  23.             public Tuple2<String, Integer> call(String s) throws Exception {
  24.                 return new Tuple2<String, Integer>(s, 1);
  25.             }
  26.         });
  27.  
  28.         //对于每一个key维护一个状态
  29.         wordCountDStream.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>(){
  30.             @Override
  31.             public Optional<Integer> call(List<Integer> newValues, Optional<Integer> state) throws Exception {
  32.                 Integer initData=0;
  33.                 //如果有state了,就拿出这个state的值
  34.                 if(state.isPresent()){
  35.                     initData=state.get();
  36.                 }
  37.                 //这里对于每一batch的新的数据对于state进行累加
  38.                 for (Integer newValue : newValues) {
  39.                     initData+=newValue;
  40.                 }
  41.                 return Optional.of(initData);
  42.             }
  43.         }).foreachRDD(rdd->rdd.foreach((VoidFunction<Tuple2<String, Integer>>) result -> System.out.println(result)));
  44.  
  45.         javaStreamingContext.start();
  46.         javaStreamingContext.awaitTermination();
  47.         javaStreamingContext.stop();
  48.     }
  49. }

scala版本

  1. object ScalaUpdateStateByKey {
  2.   def main(args: Array[String]): Unit = {
  3.     val conf = new SparkConf()
  4.     conf.setMaster("local[2]")
  5.       .setAppName("ScalaUpdateStateByKey")
  6.     //解决权限问题（主要一点要在得到sparkStreaming前配置）
  7.     System.setProperty("HADOOP_USER_NAME", "bigdata")
  8.     val sparkStreaming = new StreamingContext(conf, Seconds(1))
  9.     sparkStreaming.checkpoint("hdfs://master:8020/scalasparkstreamingcheckpoing");
  10.  
  11.     sparkStreaming.socketTextStream("master",9998)
  12.       .flatMap(_.split(" "))
  13.       .map((_,1))
  14.       .updateStateByKey((values:Seq[Int],state:Option[Int])=>{
  15.         var res: Int = state.getOrElse(0)
  16.         for (elem <- values) {
  17.           res=res+elem
  18.         }
  19.         Option(res)
  20.       }).foreachRDD(rdd=>rdd.foreach(println))
  21.  
  22.     sparkStreaming.start()
  23.     sparkStreaming.awaitTermination()
  24.     sparkStreaming.stop()
  25.   }
  26. }

transform

使用它是因为DStream只能和DStream执行join操作，当我们想和其他的RDD使用的时候就需要用到它

下面我们写一个事实黑名单点击过滤案例

java版本实现

  1. public class JavaBlackList {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         sparkConf.setMaster("local[2]")
  5.                 .setAppName("JavaHelloWord");
  6.  
  7.         //定义要过滤的黑名单的数据
  8.         List<Tuple2<String, Boolean>> blackList = new ArrayList<>();
  9.         blackList.add(new Tuple2<>("zhangsan",true));
  10.         //得到每一批的数据的时候，我们为了测试把批次间隔调大一点
  11.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(5));
  12.         JavaPairRDD<String, Boolean> blackListRDD = javaStreamingContext.sparkContext().parallelizePairs(blackList);
  13.  
  14.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9998);
  15.         JavaPairDStream<String, String> socketTupleUserMsg = socketTextStream.mapToPair(new PairFunction<String, String, String>() {
  16.             @Override
  17.             public Tuple2<String, String> call(String s) throws Exception {
  18.                 String[] socketTextStringItem = s.split(" ");
  19.                 //因为假的输入的是name msg的形式
  20.                 String userName = socketTextStringItem[0];
  21.                 String msg = socketTextStringItem[1];
  22.                 return new Tuple2<String, String>(userName, msg);
  23.             }
  24.         });
  25.  
  26.         //第一个参数是装换成的RDD，第二个是要返回什么RDD
  27.         socketTupleUserMsg.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
  28.             @Override
  29.             public JavaRDD<String> call(JavaPairRDD<String, String> socketKeyValue) throws Exception {
  30.                 //对于事实得到的用户数据和黑名单的数据进行左外连接的操作，
  31.                 JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> allUser = socketKeyValue.leftOuterJoin(blackListRDD);
  32.                 //过滤出黑名单的数据
  33.                 JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filterUser = allUser.filter(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() {
  34.                     @Override
  35.                     public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> v1) throws Exception {
  36.                         //这里是Optional的特殊处理,isPresent表示这个Optional是否有值
  37.                         //get()如果有值，并且左外连接以后匹配到了名字那么就是黑名单的数据我们直接过滤掉
  38.                         if (v1._2._2.isPresent() && v1._2._2.get()) {
  39.                             return false;
  40.                         }
  41.                         return true;
  42.                     }
  43.                 });
  44.  
  45.                 //对于连接的数据，我们只要之前输入的数据就行，那么就通过map进行装换处理
  46.  
  47.                 return filterUser.map(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, String>() {
  48.                     @Override
  49.                     public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> v1) throws Exception {
  50.                         return v1._2._1;
  51.                     }
  52.                 });
  53.             }
  54.             //得到最后过滤出来的结果
  55.         }).foreachRDD(rdd->rdd.foreach(new VoidFunction<String>() {
  56.             @Override
  57.             public void call(String res) throws Exception {
  58.                 System.out.println(res);
  59.             }
  60.         }));
  61.  
  62.  
  63.         javaStreamingContext.start();
  64.         javaStreamingContext.awaitTermination();
  65.         javaStreamingContext.stop();
  66.     }
  67. }

scala版本实现

  1. object ScalaBlackList {
  2.   def main(args: Array[String]): Unit = {
  3.     val conf = new SparkConf()
  4.     conf.setMaster("local[2]")
  5.       .setAppName("ScalaBlackList")
  6.  
  7.     //定义用户黑名单
  8.     val blackUser = List(
  9.       ("a", true)
  10.     )
  11.     val sparkStreaming = new StreamingContext(conf, Seconds(2))
  12.     val context: SparkContext = sparkStreaming.sparkContext
  13.     val blackUserRDD: RDD[(String, Boolean)] = context.parallelize(blackUser)
  14.  
  15.     val socketDS: ReceiverInputDStream[String] = sparkStreaming.socketTextStream("master", 9998)
  16.     socketDS.map(item => {
  17.       val socketText: Array[String] = item.split(" ")
  18.       val userName: String = socketText(0)
  19.       val msg: String = socketText(1)
  20.       (userName, msg)
  21.     }).transform(rdd => {
  22.       val allUser: RDD[(String, (String, Option[Boolean]))] = rdd.leftOuterJoin(blackUserRDD)
  23.       //对于所有的数据进行过滤
  24.       allUser.filter(item=>{
  25.         //表示在黑名单里面没有数据的时候就是填充一个false作为默认值
  26.         !item._2._2.getOrElse(false)
  27.       }).map(item=>{
  28.         (item._1,item._2._1)
  29.       })
  30.     }).foreachRDD(rdd=>rdd.foreach(println))
  31.  
  32.     sparkStreaming.start()
  33.     sparkStreaming.awaitTermination()
  34.     sparkStreaming.stop()
  35.   }
  36. }

window(滑动窗口)

案例

实时统计每10秒统计最近60秒的最多的搜索词top3

java版本

  1. public class JavaWinddow {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         sparkConf.setAppName("JavaWinddow")
  5.                 .setMaster("local[2]");
  6.  
  7.         //每一秒得到一份batch数据，里面其实一堆RDD
  8.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  9.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9998);
  10.         JavaPairDStream<String, Integer> wordCount = socketTextStream.mapToPair(new PairFunction<String, String, Integer>() {
  11.             @Override
  12.             public Tuple2<String, Integer> call(String s) throws Exception {
  13.                 return new Tuple2<String, Integer>(s, 1);
  14.             }
  15.         });
  16.  
  17.         //这里使用窗口的特性，每10秒统计最近60秒的数据,也就是上面的数据会一直累积，到达条件以后往下面执行
  18.         //第一个是reduceByKey的操作函数，
  19.         // 第二个值是窗口大小
  20.         //第三个值是滑动距离
  21.         JavaPairDStream<String, Integer> reduceWindowValue = wordCount.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
  22.             @Override
  23.             public Integer call(Integer v1, Integer v2) throws Exception {
  24.                 return v1 + v2;
  25.             }
  26.         }, Durations.seconds(60), Durations.seconds(10));
  27.  
  28.         JavaPairDStream<Integer, String> integerStringJavaPairDStream = reduceWindowValue.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
  29.             @Override
  30.             public Tuple2<Integer, String> call(Tuple2<String, Integer> initData) throws Exception {
  31.                 return new Tuple2<Integer, String>(initData._2, initData._1);
  32.             }
  33.         });
  34.         //装换成RDD使用RDD的操作,对于60秒的数据进行排序
  35.         JavaPairDStream<Integer, String> sortDS = integerStringJavaPairDStream.transformToPair(rdd -> rdd.sortByKey(false));
  36.         
  37.         sortDS.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
  38.             @Override
  39.             public Tuple2<String, Integer> call(Tuple2<Integer, String> res) throws Exception {
  40.                 return new Tuple2<String, Integer>(res._2, res._1);
  41.             }
  42.         }).foreachRDD(rdd->rdd.take(3).forEach(
  43.                 item->{
  44.                     System.out.println(item);
  45.                 }
  46.         ));
  47.  
  48.  
  49.         javaStreamingContext.start();
  50.         javaStreamingContext.awaitTermination();
  51.         javaStreamingContext.stop();
  52.     }
  53. }

scala版本

  1. object ScalaWindow {
  2.   def main(args: Array[String]): Unit = {
  3.     val conf = new SparkConf()
  4.     conf.setAppName("ScalaWindow")
  5.       .setMaster("local[2]")
  6.  
  7.     val sparkStreaming = new StreamingContext(conf, Seconds(1))
  8.     val socketDS: ReceiverInputDStream[String] = sparkStreaming.socketTextStream("master", 9998)
  9.     val reduceRDD: DStream[(String, Int)] = socketDS.map((_, 1))
  10.       .reduceByKeyAndWindow((num1:Int,num2:Int)=>num2+num1, Seconds(60), Seconds(10))
  11.  
  12.     reduceRDD.map(item=>{
  13.       (item._2,item._1)
  14.     }).transform(rdd=>rdd.sortByKey(false)).foreachRDD(rdd=>rdd.take(3).foreach(println))
  15.  
  16.     sparkStreaming.start()
  17.     sparkStreaming.awaitTermination()
  18.     sparkStreaming.stop()
  19.   }
  20. }

foreachRDD详解

案例

实时处理10秒统计一次最近60秒的搜索按搜索的次数倒叙

  1. public class JavaWinddow {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         sparkConf.setAppName("JavaWinddow")
  5.                 .setMaster("local[2]");
  6.  
  7.         //每一秒得到一份batch数据，里面其实一堆RDD
  8.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  9.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9998);
  10.         JavaPairDStream<String, Integer> wordCount = socketTextStream.mapToPair(new PairFunction<String, String, Integer>() {
  11.             @Override
  12.             public Tuple2<String, Integer> call(String s) throws Exception {
  13.                 return new Tuple2<String, Integer>(s, 1);
  14.             }
  15.         });
  16.  
  17.         //这里使用窗口的特性，每10秒统计最近60秒的数据,也就是上面的数据会一直累积，到达条件以后往下面执行
  18.         //第一个是reduceByKey的操作函数，
  19.         // 第二个值是窗口大小
  20.         //第三个值是滑动距离
  21.         JavaPairDStream<String, Integer> reduceWindowValue = wordCount.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
  22.             @Override
  23.             public Integer call(Integer v1, Integer v2) throws Exception {
  24.                 return v1 + v2;
  25.             }
  26.         }, Durations.seconds(60), Durations.seconds(10));
  27.  
  28.         JavaPairDStream<Integer, String> integerStringJavaPairDStream = reduceWindowValue.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
  29.             @Override
  30.             public Tuple2<Integer, String> call(Tuple2<String, Integer> initData) throws Exception {
  31.                 return new Tuple2<Integer, String>(initData._2, initData._1);
  32.             }
  33.         });
  34.         //装换成RDD使用RDD的操作,对于60秒的数据进行排序
  35.         JavaPairDStream<Integer, String> sortDS = integerStringJavaPairDStream.transformToPair(rdd -> rdd.sortByKey(false));
  36.  
  37.         sortDS.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
  38.             @Override
  39.             public Tuple2<String, Integer> call(Tuple2<Integer, String> res) throws Exception {
  40.                 return new Tuple2<String, Integer>(res._2, res._1);
  41.             }
  42.         }).foreachRDD(rdd -> rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
  43.             @Override
  44.             public void call(Iterator<Tuple2<String, Integer>> res) throws Exception {
  45.                 //得到连接，每一个连接处理一个分区的数据
  46.                 Jedis jedis = RedisHelper.getJedis();
  47.                 Map<String, String> resMap = new HashMap<>();
  48.                 while (res.hasNext()) {
  49.                     Tuple2<String, Integer> value = res.next();
  50.                     String oldValue = resMap.get(value._1);
  51.                     if (oldValue == null) {
  52.                         oldValue = "0";
  53.                     }
  54.                     Integer newValue = Integer.valueOf(oldValue) + value._2;
  55.                     resMap.put(value._1, newValue + "");
  56.                 }
  57.                 if(!resMap.isEmpty()){
  58.                     jedis.del("hotWordTop3");
  59.                     jedis.hset("hotWordTop3", resMap);
  60.                 }
  61.                 //把连接还回去
  62.                 RedisHelper.removeJedis(jedis);
  63.             }
  64.         }));
  65.  
  66.  
  67.         javaStreamingContext.start();
  68.         javaStreamingContext.awaitTermination();
  69.         javaStreamingContext.stop();
  70.     }
  71. }

放入redis

  1.         <dependency>
  2.             <groupId>redis.clients</groupId>
  3.             <artifactId>jedis</artifactId>
  4.             <version>3.1.0</version>
  5.         </dependency>

如果没有redis想快点使用的话下面提供docker安装

  1. sudo yum install docker
  2. sudo systemctl start docker
  3. sudo systemctl enable docker

时间同步操作(有些时间不一样会有问题)

  1. sudo yum install -y ntpdate
  2. sudo ntpdate 120.24.81.91
  3.  
  4. #冲突时使用这个
  5. sudo sudo systemctl stop ntp

镜像加速(由于自己提供的经常过期，可以自己去找下)

  1. sudo systemctl daemon-reload
  2. sudo systemctl restart docker

启动一个redis容器并且开机自启

sudo docker run -itd --name spark-redis -p 6379:6379  --restart=always  redis:6.0

创建一个redis的工具类

  1. public class RedisHelper {
  2.  
  3.     private static JedisPool jedisPool=null;
  4.  
  5.     static {
  6.  
  7.         GenericObjectPoolConfig config = new GenericObjectPoolConfig();
  8.         config.setMaxIdle(10);
  9.         config.setMaxTotal(20);
  10.         config.setMinIdle(5);
  11.         jedisPool = new JedisPool(config, "node1");
  12.     }
  13.  
  14.     public static Jedis getJedis() {
  15.         Jedis resource = jedisPool.getResource();
  16.         return resource;
  17.     }
  18.  
  19.     public static void removeJedis(Jedis jedis){
  20.         jedis.close();
  21.     }
  22. }

结合SparkSql实战

案例要求

每10秒统计最近60秒用户点击top3的数据

  1. public class JavaStreamingSql {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SparkConf sparkConf = new SparkConf();
  4.         sparkConf.setAppName("JavaStreamingSql")
  5.                 .setMaster("local[2]");
  6.         //第一步得到sparkStreaming，我们可以认为DStream就是1s的一个RDD，但是事件上是产生了一个batch的数据
  7.         //一个batch里面有很多RDD
  8.         JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
  9.         JavaReceiverInputDStream<String> socketTextStream = javaStreamingContext.socketTextStream("master", 9998);
  10.  
  11.         //得到的点击数据进行map
  12.         JavaPairDStream<String, Integer> wordCountDStream = socketTextStream.mapToPair(new PairFunction<String, String, Integer>() {
  13.             //得到的数据是每个用户名
  14.             @Override
  15.             public Tuple2<String, Integer> call(String userName) throws Exception {
  16.                 return new Tuple2<String, Integer>(userName, 1);
  17.             }
  18.         });
  19.  
  20.         //第二步开窗并求聚合操作
  21.         JavaPairDStream<String, Integer> reduceByKeyAndWindowDStream = wordCountDStream.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
  22.             @Override
  23.             public Integer call(Integer v1, Integer v2) throws Exception {
  24.                 return v1 + v2;
  25.             }
  26.         }, Durations.seconds(60), Durations.seconds(10));
  27.  
  28.         //第三步根据得到的RDD使用SparkSql进行分析
  29.         reduceByKeyAndWindowDStream.foreachRDD(rdd->{
  30.             //根据RDD得到Row
  31.             JavaRDD<Row> userClickRowCount = rdd.map(new Function<Tuple2<String, Integer>, Row>() {
  32.                 @Override
  33.                 public Row call(Tuple2<String, Integer> rddItem) throws Exception {
  34.                     return RowFactory.create(
  35.                             rddItem._1,
  36.                             rddItem._2
  37.                     );
  38.                 }
  39.             });
  40.  
  41.             //定义Row的元数据
  42.             List<StructField> rowMeta = new ArrayList<>();
  43.             rowMeta.add(DataTypes.createStructField("userName",DataTypes.StringType,true));
  44.             rowMeta.add(DataTypes.createStructField("clickNum",DataTypes.IntegerType,true));
  45.             StructType rowMetastructType = DataTypes.createStructType(rowMeta);
  46.  
  47.             //使得Row和structType建立联系
  48.             SQLContext sqlContext = new SQLContext(javaStreamingContext.sparkContext());
  49.             Dataset<Row> userClickDataFrame = sqlContext.createDataFrame(userClickRowCount, rowMetastructType);
  50.             userClickDataFrame.registerTempTable("userClickData");
  51.  
  52.             //这里开始对于每10秒统计最近60秒的数据
  53.             Dataset<Row> res = sqlContext.sql(
  54.                     "select userName,clickNum from " +
  55.                             "(" +
  56.                             "select userName,clickNum,row_number() over (order by clickNum desc) as rank  from " +
  57.                             "userClickData" +
  58.                             ") where rank<=3"
  59.             );
  60.  
  61.             res.show();
  62.         });
  63.  
  64.         javaStreamingContext.start();
  65.         javaStreamingContext.awaitTermination();
  66.         javaStreamingContext.stop();
  67.     }
  68. }

输出的结果是 

  1. +--------+--------+
  2. |userName|clickNum|
  3. +--------+--------+
  4. |       c|      28|
  5. |       a|      19|
  6. |       f|      12|
  7. +--------+--------+

scala版本

  1. object ScalaSparkStreamSql {
  2.   def main(args: Array[String]): Unit = {
  3.     val conf = new SparkConf()
  4.     conf.setAppName("ScalaSparkStreamSql")
  5.       .setMaster("local[2]")
  6.  
  7.     val sparkStreaming = new StreamingContext(conf, Seconds(1))
  8.     sparkStreaming.socketTextStream("master", 9998)
  9.       .map((_, 1))
  10.       //每10秒得到最近60秒的数据，60是窗口的大小,10是滑动的长度
  11.       .reduceByKeyAndWindow((num1: Int, num2: Int) => {
  12.         num1 + num2
  13.       }, Seconds(60), Seconds(10))
  14.       .foreachRDD(rdd => {
  15.         //得到Row
  16.         val userClickRow: RDD[Row] = rdd.map(
  17.           mapItem => {
  18.             Row(mapItem._1, mapItem._2)
  19.           }
  20.         )
  21.         //得到数据
  22.         val userClickStructTyoe: StructType = StructType(
  23.           Array(
  24.             StructField("userName", DataTypes.StringType, true),
  25.             StructField("clickNum", DataTypes.IntegerType, true)
  26.           )
  27.         )
  28.  
  29.         //得到DataFrame
  30.         val sparkSession: SparkSession = SparkSession
  31.           .builder()
  32.           .config(conf)
  33.           .getOrCreate()
  34.  
  35.         //开始查询数据
  36.         val userClickDataFrame: DataFrame = sparkSession.createDataFrame(userClickRow, userClickStructTyoe)
  37.         userClickDataFrame.createTempView("userClickData")
  38.  
  39.         sparkSession.sql(
  40.           """
  41.             |select userName,clickNum from
  42.             |(
  43.             |	select userName,clickNum,row_number() over (order by clickNum desc) as rank  from
  44.             |	userClickData
  45.             |) where rank<=3
  46.             |""".stripMargin
  47.         ).show()
  48.  
  49.         //后面可以更具结果的数据自己处理保存到redis或者mysql
  50.       })
  51.  
  52.  
  53.     sparkStreaming.start()
  54.     sparkStreaming.awaitTermination()
  55.     sparkStreaming.stop()
  56.   }
  57. }

输出结果

  1. +--------+--------+
  2. |userName|clickNum|
  3. +--------+--------+
  4. |       a|       9|
  5. |       v|       3|
  6. |       c|       3|
  7. +--------+--------+

缓存与持久化机制

  • sparkstreaming默认对窗口的操作，还有upstateBykey开启的持久化的操作
  • 与RDD不同的是它默认都是会要序列化的

Sparkstreaming容错机制

分区容错

 可靠receiver

SparkStreaming架构原理

sparkStreaming架构原理图解

 接收数据的原理

写数据原理

 

SparkStreaming性能调优 

  • 对于receiver模式的处理，可以调高receiver的数量，提高并行度
  • 对于每一批的RDD进行调优，因为RDD的分区和每一个batch里面的block有关，比如batch的间隔是2s,也就是JavaStreamingContext(sparkConf, Durations.seconds(2))，这里就是设置拉去batch的时间，每一批batch里面有若干的block，每一个block产生的时间是200毫秒，每一个block对应RDD的一个分区，这里调优的地方就是如果你的CPU有8核，但是如果你每一批的数据就是5个block的话，那么就是没有充分的利用cpu我们这个时候就可以对于block产生的时间减少
  • 对于Kafka的Direct模式的话，我们只要设置的并行度和分区一样就可以了
  • 使用CMS垃圾回收器，减少gc的时间，提高batch的处理速度
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/824814
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号