HDFS数据源、DStream的持久化存储_ 2|学习笔记

开发者学堂课程【大数据实时计算框架 Spark 快速入门：HDFS 数据源、DStream 的持久化存储_ 2】学习笔记，与课程紧密联系，让用户快速学习知识。

课程地址：

https://developer.aliyun.com/learning/course/100/detail/1723

HDFS 数据源、DStream 的持久化存储_ 2

​

package com. shsxt. study,streaming;

import java. util. Arrays;[

public class  UpdateStateByKeyWordcount {

public static void main(String[]args){

SparkConfconfen—new Sparkconf(). setApplame (“ IpdateStateyWordcount ”).setMas  Javastreaming  Context jsscsc=new JavaStreamingContext (conf, Durations-seconds(5) js jssc.  checkpoint (".");

 JavaReceivprInputDStream <String>lines=jssc. socketTextStream ("node24", 8888)]

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

private static final long serial  VersionUID =1L;

@Override

public Iterable<String>call(String line) throws Exception{

return Arrays.asList(line, split(""));

JavaPairDStreamsString ,Integer>pairds.mapIoPair(new Pair function<String, String, Integer/O

private static final long serial  VersionUID =1L;

@Override

publicTuple2 <String, Integer>call(String word) throws Exception{return newTuple2<String, Integer>(word,1);

}

});

JavaPainDStreawKtrin ,Integer)  wordcounts - pudatestated $Key(new Function? list(integer), Optional.

​​

private static final long se  rialVersionUID =1L;

//实际上，对于每个单词，每次 batch 计算的时候，都会调用这个函数，第一个参数 values 相当于这个 batch 中

//这个 key 对应的新的一组值，可能有多个，可能 2 个 1，（xuruyun，1）（xuru yyun,1），那么这个 values 就是（1，1）

//那么第二个参数表示的是这个 key 之前的状态，我们看类型 Integer 就知道了，这里是泛型自己指定的。