Spark Streaming之UpdateStateByKey算子详解

不念不忘少年蓝@ 2022-12-28 12:48 262阅读 0赞

流处理中，有个状态（state）的概念：

*  无状态的：当前批次处理完之后，数据只与当前批次有关
 *  有状态的：前后批次的数据处理完之后，之间是有关系的

**updateStateByKey解读**

updateStateByKey：返回的是一个新的并且带有状态的DStream，会根据每一个key进行更新，更新的规则是根据自己定义的function来确定的。

updateStateByKey操作允许您在使用新信息不断更新时保持任意状态。要使用它，您必须执行两个步骤：

*  定义状态:状态可以是任意数据类型。
 *  定义状态更新功能:使用函数指定如何使用先前状态和输入流中的新值更新状态。

对于每个batch，Spark都会为每个之前已经存在的key去应用一次state更新函数，无论这个key在 batch中是否有新的数据。  
如果state更新函数返回none，那么key对应的state就会被删除。 当然，对于每个新出现的key，也会执行state更新函数。

**注意，updateStateByKey操作，要求必须开启Checkpoint机制。**

**案例：基于缓存的实时wordcount程序（在实际业务场景中，这个是非常有用的）**

*  定义state；以wordcount为例，value是作为state来处理的，是根据key来更新我们的value的
 *  定义一个state update function：将旧state的值与新state的值根据所定义的规则给相互作用在一起

Java语言实现：

package com.kfk.spark.common;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.streaming.Durations;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/14 * @time : 8:23 下午 */
    public class CommStreamingContext { 
    
        public static JavaStreamingContext getJssc(){ 
            SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("CommStreamingContext");
            return new JavaStreamingContext(conf, Durations.seconds(2));
        }
    }

package com.kfk.spark.stream;
    
    import com.kfk.spark.common.CommStreamingContext;
    import org.apache.spark.api.java.Optional;
    import org.apache.spark.api.java.function.Function2;
    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;
    
    import java.util.Arrays;
    import java.util.List;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/15 * @time : 10:08 下午 */
    public class StreamingUpdateStateByKeyJava { 
        public static void main(String[] args) throws InterruptedException { 
            JavaStreamingContext jssc = CommStreamingContext.getJssc();
    
            // 要使用UpdateStateByKey算子就必须设置一个Checkpoint目录，开启Checkpoint机制
            // 以便于内存数据丢失时，可以从Checkpoint中恢复数据
            jssc.checkpoint("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/sparkCheckpoint");
    
            JavaReceiverInputDStream<String> lines = jssc.socketTextStream("bigdata-pro-m04",9999);
    
            // flatmap
            JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
    
            // map
            JavaPairDStream<String,Integer> pair =  words.mapToPair(word -> new Tuple2<>(word,1));
    
            // 通过spark来维护一份每个单词的全局统计次数
            JavaPairDStream<String,Integer> wordcount = pair.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { 
                @Override
                public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception { 
    
                    Integer newValues = 0;
                    if (state.isPresent()){ 
                        newValues = state.get();
                    }
    
                    for (Integer value : values){ 
                        newValues += value;
                    }
    
                    return Optional.of(newValues);
                }
            });
    
            // lambda表达式写法
    // JavaPairDStream<String,Integer> wordcount = pair.updateStateByKey((values, state) -> { 
    // Integer newValues = 0;
    // if (state.isPresent()){ 
    // newValues = state.get();
    // }
    //
    // for (Integer value : values){ 
    // newValues += value;
    // }
    //
    // return Optional.of(newValues);
    //
    // });
    
            wordcount.print();
    
            jssc.start();
            jssc.awaitTermination();
    
        }
    }

Scala语言实现：

package com.kfk.spark.stream
    
    import com.kfk.spark.common.CommStreamingContextScala
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/16 * @time : 8:04 下午 */
    object StreamingUpdateStateByKeyScala { 
        def main(args: Array[String]): Unit = { 
            val jssc = CommStreamingContextScala.getJssc;
    
            val lines = jssc.socketTextStream("bigdata-pro-m04", 9999)
    
            // 要使用UpdateStateByKey算子就必须设置一个Checkpoint目录，开启Checkpoint机制
            // 以便于内存数据丢失时，可以从Checkpoint中恢复数据
            jssc.checkpoint("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/sparkCheckpoint")
    
            // flatmap
            val words = lines.flatMap(word => word.split(" "))
    
            // map
            val pair = words.map(x => (x,1))
    
            // updateStateByKey
            val wordcount = pair.updateStateByKey((values : Seq[Int], state : Option[Int]) => { 
                var newValue = state.getOrElse(0)
    
                for (value <- values){ 
                    newValue += value
                }
                Option(newValue)
            })
    
            wordcount.print()
    
            jssc.start()
            jssc.awaitTermination()
        }
    
    }

运行结果：

-------------------------------------------
    Time: 1608175348000 ms
    -------------------------------------------
    (hive,2)
    (php,1)
    (python,1)
    (java,6)
    
    -------------------------------------------
    Time: 1608175354000 ms
    -------------------------------------------
    (hive,2)
    (php,1)
    (python,2)
    (java,7)
    (hadoop,1)vb

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！