原理与实践全书课件完整版ppt全套教学教程最全电子教案电子讲义(最新).pptx

《原理与实践全书课件完整版ppt全套教学教程最全电子教案电子讲义(最新).pptx》由会员分享，可在线阅读，更多相关《原理与实践全书课件完整版ppt全套教学教程最全电子教案电子讲义(最新).pptx（322页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、第八章Table API&SQL的介绍和使用第二章大数据必备编程知识l案例：动物类（Animal）和鱼类（Fish）l继承关系保证所有动物子类都具有动物类的属性和方法l子类有自己的属性和方法。l除了动物，还有很多其他事物也会移动，使用接口（interface）来抽象“移动”。继承lJava的继承：继承类 extends 实现接口 implements继承publicclassClassAimplementsMoveOverridepublicvoidmove().实现接口publicclassDogextendsAnimalprivateStringdogData;public Dog(Str

2、ing myName,String myDescription,StringmyDogData)this.name=myName;this.description=myDescription;this.dogData=myDogData继承类interfaceclassl重写：子类和父类都定义同名方法，子类的方法会覆盖父类中已有的方法。l重载：多个同名方法，这些方法名字相同、参数不同、返回类型不同。重写与重载publicclassClassAimplementsMoveOverridepublicvoidmove().Override：在子类中重写父类中的同名方法publicclassOver

3、loading/无参数，返回值为int类型publicinttest()System.out.println(test);return1;/有一个参数publicvoidtest(inta)System.out.println(test+a);/有两个参数和一个返回值publicStringtest(inta,Strings)System.out.println(test+a+s);returna+s;同名方法重载：一个类中多个方法都名为test，但是参数类型和返回值类型不同。l案例：Java中的List和ArrayListlArrayList是一个泛型类，List是一个泛型接口lArrayL

4、ist泛型是一种集合容器，可以向这个集合容器中添加String、Double以及其他各类数据类型。没必要创建StringArrayList、DoubleArrayList等类。泛型iststrList=newArrayList();ListdoubleList=newLinkedList();l类名后面加上 l类内部的一些属性和方法都可以使用泛型Tl泛型规范：T 代表一般的任何类。E 代表元素（Element）或异常（Exception）。K 或KEY代表键（Key）。V 代表值（Value），通常与K一起配合使用。Java泛型类publicclassMyArrayList privatein

5、tsize;Telements;publicMyArrayList(intcapacity)this.size=capacity;this.elements=(T)newObjectcapacity;publicvoidset(Telement,intposition)elementsposition=element;OverridepublicStringtoString()Stringresult=;for(inti=0;isize;i+)result+=elementsi.toString();returnresult;l与泛型类类似，使用符号l可以继承并实现这个接口Java泛型接口pu

6、blicinterfaceList.publicListsubList(intfromIndex,inttoIndex);publicclassArrayList implementsList.publicListsubList(intfromIndex,inttoIndex)./返回一个List类型值要实现的子类是泛型的publicclassDoubleListimplementsList.publicListsubList(intfromIndex,inttoIndex)./返回一个List类型值要实现的子类不是泛型的，而是有确定类型的l泛型方法可以存在于泛型类中，也可以存在于普通的类中。

7、l泛型方法的类型E和泛型类中的类型T可以不一样。泛型方法是泛型类的一个成员，泛型方法既可以继续使用类的类型T，也可以自己定义新的类型E。Java泛型方法publicclassMyArrayList./public关键字后的表明该方法是一个泛型方法/泛型方法中的类型E和泛型类中的类型T可以不一样public E processElement(E element).returnE;lJava泛型信息只存在于代码编译阶段，当程序运行到JVM上时，与泛型相关的信息会被擦除。l对于绝大多数应用系统开发者来说影响不太大，对于框架开发者来说，必须要注意。类型擦除ClassstrListClass=newAr

8、rayList().getClass();ClassintListClass=newArrayList().getClass();/输出：classjava.util.ArrayListSystem.out.println(strListClass);/输出：classjava.util.ArrayListSystem.out.println(intListClass);/输出：trueSystem.out.println(strListClass.equals(intListClass);泛型擦除：无法区别strListClass和intListClass这两个类型l适合进行并行计算的一种编

9、程范式l非函数式编程：创建中间变量，分步执行l函数式编程：与数学表达式更相似实现单个函数，将零到多个输入转换成零到多个输出。比如，add()将两个输入转化为一个输出。将多个函数连接起来，实现所需业务逻辑。比如，将add()、multiply()连接到一起。函数式编程addResult=x+yresult=addResult*z非函数式编程result=add(x,y).multiply(z)函数式编程lLambda表达式被一些编程语言用来实现函数式编程。l一个箭头符号-，两边连接着输入参数和函数体。Lambda表达式(parameters)-bodyJava的Lambda表达式的语法规则/接收

10、2个int类型参数，返回它们的和(intx,inty)-x+y/接收1个String类型参数，将其输出到控制台，不返回任何值(Strings)-System.out.print(s);/参数为圆半径，返回圆面积，返回值为double类型(doubler)-doublepi=3.1415;returnr*r*pi;几个JavaLambda表达式案例l输入参数：接收零到多个输入参数程序员可以提供输入类型，也可以不提供类型，让代码根据上下文去推断参数可以放在圆括号()中，多个参数通过英文逗号,隔开l函数体：可以有一到多行语句函数体有多行内容，必须使用花括号 输出的类型与所需要的类型相匹配Java L

11、ambda表达式lLambda表达式本质是一种接口，它要实现一个函数式接口（Functional Interface）中的虚方法l函数式接口是一种接口，并且它只有一个虚方法。lFunctionalInterface 注解函数式接口FunctionalInterfaceinterfaceAddInterfaceTadd(Ta,Tb);publicstaticclassMyAddimplementsAddInterfaceOverridepublicDoubleadd(Doublea,Doubleb)returna+b;如果没有Lambda表达式(Integera,Integerb)-a+b;使用

12、Lambda表达式lJava 8之后推出的，专注于对集合（Collection）对象的操作。l右侧案例：数据先经过stream()方法被转换为一个Stream类型，后经过filter()、map()、collect()等处理逻辑，生成我们所需的输出。各个操作之间使用英文点号.来连接，这种方式被称作链式调用（Method Chaining）。l链式调用：将多个函数连接起来。lFlink的API是面向数据集或数据流的操作。这些操作分布在大数据集群的多个节点上，并行地分布式执行。Java Stream APIListstrings=Arrays.asList(abc,bc,12345,efg,abc

13、d,jkl);Listlengths=strings.stream().filter(string-!string.isEmpty().map(s-s.length().collect(Collectors.toList();lengths.forEach(s)-System.out.println(s);第九章Flink的部署和配置部署模式Standalone集群：集群内只部署FlinkHadoop YARN集群：兼容Hadoop生态，可以部署Hadoop、Hive、Spark、FlinkKubernetes集群：兼容各类容器l至少一个Master，至少一个WorkerlMaster：Dis

14、patcher、ResourceManager和JobManagerlWorker：TaskManagerl安装好Java、配置好SSH免密码访问l配置Flink主目录的conf/flink-conf.yaml和conf/slaves两个文件l将Flink主目录部署到每个节点的相同路径，或者部署到所有节点都可访问的共享目录l启动这个集群：bin/start-cluster.sh Standalone集群一个Standalone集群拓扑样例$./bin/flinkrun-m192.168.0.1:8081./examples/batch/WordCount.jar提交作业：使用针对Hadoop的

15、Flink，设置HADOOP_CONF_DIR和YARN_CONF_DIR 等与Hadoop相关的配置启动好YARN集群三种作业提交方式：Per-Job：每次向YARN提交一个作业，YARN为这个作业单独分配资源，基于这些资源启动一个Flink集群，作业运行结束，相应的资源会被释放。Session：在YARN上启动一个长期运行的Flink集群，用户可以向这个集群提交多个作业。Application：在Per-Job模式上做了一些优化。Hadoop YARNlClient将作业提交给YARN的ResourceManagerlYARN为这个作业生成一个Application Master以运行Fi

16、nk Master，里面运行这运行着JobManager和Flink-YARN ResourceManager lJobManager会根据本次作业所需资源向Flink-YARN ResourceManager申请Slot资源 lFlink-YARN ResourceManager会向YARN申请所需的资源作为TaskManagerlTaskManager将这些Slot注册到Flink-YARN ResourceManagerlJobManager将作业的计算任务部署到各TaskManager上 YARN Per-JobYARNPer-Job模式l在Client上，使用bin/yarn-ses

17、sion.sh启动一个YARN Session，Flink向YARN ResourceManager申请一个Application Masterl用户提交一个作业，作业被发送给DispatcherlDispatcher会启动针对该作业的JobManagerlJobManager向Flink-YARN ResourceManager上申请所需资源，启动TaskManagerlTaskManager注册lJobManager将计算任务部署到各TaskManager上l如果用户提交下一个作业，那么Dispatcher启动新的JobManager，新的JobManager负责新作业的资源申请和任务调度

18、。YARN SessionYARNSession模式lPer-Job和Session模式作业提交的过程依赖Client，main()方法在Client上执行，main()方法会将作业的各个依赖下载到本地，生成JobGraph，并将依赖以及JobGraph发送到Flink集群,负载很重。lApplication模式允许main()方法在JobManager上执行，这样可以分担Client的压力 YARN Applicationl两种方式：Kubernetes和Kubernets原生lKubernetes原生Session模式：使用bin/kubernetes-session.sh启动一个Kube

19、rnetes SessionKubernetes相关组件将进行初始化，生成Flink Master（Dispatcher、Flink-Kubernetes ResourceManager）用户提交作业，申请资源，将作业运行到TaskManager上lKubernetes原生Application模式KubernetesKubernetes原生Session模式l使用Key-Value来设置参数，很多进程会读取这个文件l从Flink官网下载的Flink主目录里的flink-conf.yaml有一些默认设置，针对单机环境，在自己环境中使用，需要修改这个文件conf/flink-conf.yaml

20、l安装Java，设置$JAVA_HOMEl类加载：将Java的.class文件加载到JVM虚拟机l一个Flink作业主要加载两种类：Java Classpath：JDK核心类库和Flink主目录下lib文件夹中的类用户类（User Code）：用户编写的应用作业中的类l类加载策略：子类优先（Child-first）：优先加载用户编写的应用作业中的类，Flink默认使用Child-first父类优先（Parent-first）：优先加载Java Classpath中的类Java和类加载l并行度（parallelism）：每个算子都会被切分为parallelism个子任务，分布到多个Slot上。根

21、据计算规模大小，调整并行度。l如果作业开启了算子链和槽位共享，一个Slot上运行着一个作业所有算子组成的流水线（Pipeline），这个作业需要parallelism个Slot。l槽位划分：将TaskManager划分成多少个Slot。使用 taskmanager.numberOfTaskSlots 设置默认值为1，Standalone集群官方建议将参数值配置为与CPU核心数相等或成比例 配置taskmanager.numberOfTaskSlots没有绝对的准则：每个TaskManager下有一个Slot，该Slot会独立运行在一个JVM进程中。隔离性好。每个TaskManager下有多个S

22、lot，那么多个Slot同时运行在一个JVM进程中。多个Slot可以共享资源，隔离性差。并行度和槽位划分l堆区（Heap）:JVM虚拟化之后的内存，存储Java对象实例，使用Java垃圾回收（Garbage Collection，GC）机制来清理内存中的不再使用的对象。l某个时间点，必须进行一次Full GC，Full GC影响Java应用的性能。l悖论：Heap越大，Full GC时间越长。Heap太小，会出现OutOfMemoryError异常。l堆外（Off-Heap）：直接由操作系统管理的内存，适合读写操作频繁的场景。使用、监控和调试更复杂。Full GC影响性能时，可以考虑Off-H

23、eap。Java内存lJVM进程：Master、TaskManagerlFlink占用的内存（Total Flink Memory）：Flink可以使用的内存JVM HeapOff-heap MemoryFlink Managed Memory：TaskManager个别场景使用 Direct Memory：网络通信缓存使用的内存 lJVM相关内存（JVM Specific Memory）：Java程序都需要的一块内存区域lFlink将内存管理部分做了封装，用户在绝大多数情况下不用关注数据到底是如何写入内存的Flink内存模型lMaster占用内存不大l最简单的配置方法：设置Total Pro

24、cess Memory（jobmanager.memory.process.size），根据默认分配比例，将内存分配给各个模块。设置Total Flink Memory（jobmanager.memory.flink.size），如果Master进程需要管理多个作业，需要增大这个配置。Master的内存配置 lTaskManager涉及数据处理，对内存的需求很高l根据是否为Flink框架所用，分为Flink框架和非Flink框架，框架使用的内存，用户任务无法使用（Framework Heap，Framework Off-heap）用户任务使用的内存（Task Heap、Task Off-hea

25、p、Flink Managed Memory和Network）l根据堆区非堆区，分为Heap和Off-heapTask Heap：用户程序内存Flink Managed Memory：流处理下RocksDB的State Backend，批处理下排序、中间数据缓存等Network：Netty缓存TaskManager的内存配置 l粗粒度的配置方法：配置Total Process Memory或Total Flink Memory两者中的任意一个。各个子模块根据默认比例获得相应的内存。Total Process Memory：整个进程的内存，适合容器化部署方式。（taskmanager.memor

26、y.process.size）Total Flink Memory：Flink可用内存，适合Standalone集群方式。（taskmanager.memory.flink.size）l细粒度的配置方法同时配置Task Heap和Flink Managed Memory两个内存。这两个内存服务于具体的计算任务。（taskmanager.memory.task.heap.size和taskmanager.memory.managed.size）l以上3个参数不要同时配置，否则会引起冲突。TaskManager的内存配置 lFlink会将一部分数据写到本地磁盘，比如：日志信息、RocksDB数据等

27、。lio.tmp.dirs 配置了本地磁盘读写位置，默认会使用JVM的参数java.io.tmpdir lLinux下一般为/tmp磁盘lFlink使用算子链将尽可能多的上、下游算子链接到一起。上、下游算子会被捆绑到一起，作为一个线程执行。可以提高资源利用率。l如果两个算子不做算子链，算子间数据通信存在序列化和反序列化，通信成本高。l数据不发生交换，才可以进行算子链。l两个算子负载都很高，不应该进行算子链。l默认开启算子链。env.disableOperatorChaining()：关闭算子链。lstartNewChain()：对特定算子开启新的算子链。算子链和槽位共享lFlink默认开启了槽

28、位共享：从Source到Sink的所有算子子任务组成的Pipline可以共享一个Slot。l也可以手动设置槽位共享组（Slot Sharing Group）。槽位共享整个作业并行度为2，从Source到Sink所有算子共享一个Slot。给WindowAggreagtion设置SlotSharingGroup，该算子及之后的算子被划分到另外的Slotstream.timeWindow(.).sum(.).slotSharingGroup(“A”);l生产环境中，一般使用命令行管理作业，名为 flink，放在Flink主目录下的bin目录下。l功能：提交、取消、罗列当前作业，获取作业信息，设置Sa

29、vepoint。l会从conf/flink-conf.yaml里读取配置信息。l使用方式：lACTION包括run（提交作业）、stop（取消作业）等。OPTIONS为一些预置的选项，ARGUMENTS是用户传入的参数。命令行工具$./bin/flinkOPTIONSARGUMENTSl模板：l需要使用Maven对用户代码打包，得到JAR包。l使用WordCount的例子：l使用-c参数，设置程序入口类：l在命令行中用-p选项设置这个作业的并行度提交作业$./bin/flinkrunOPTIONSARGUMENTS$./bin/flinkrun./examples/streaming/Word

30、Count.jar$./bin/flinkrun./examples/streaming/WordCount.jar-corg.apache.flink.streaming.examples.wordcount.WordCount$./bin/flinkrun-p2./examples/streaming/WordCount.jarl传入参数，参数会写入main方法的参数String中l使用-m选项来设置向具体哪个Master提交提交作业$./bin/flinkrun./examples/streaming/WordCount.jar-corg.apache.flink.streaming.e

31、xamples.wordcount.WordCount-input/tmp/a.log-output/tmp/b.log$./bin/flinkrun-mmyJMHost:8081./examples/streaming/WordCount.jarl罗列当前作业：l触发一个作业执行Savepoint，savepointDirectory为目录：l关停一个Flink作业：l从一个Savepoint恢复一个应用作业：管理作业$./bin/flinklist$./bin/flinksavepointsavepointDirectory$./bin/flinkcancel$./bin/flinkrun

32、-sOPTIONSl确认已经在机器上安装好Hadoop，配置好环境变量HADOOP_CONF_DIR：l将Hadoop依赖包添加到Flink中，两种方法：在环境变量中添加Hadoop Classpath，Flink从Hadoop Classpath中读取所需依赖包。将所需的Hadoop JAR包添加到Flink主目录下的lib目录中。与Hadoop集成HADOOP_CONF_DIR=/path/to/etc/hadooplHADOOP_CLASSPATH存储Hadoop相关类的路径lhadoop是Hadoop提供的命令，hadoop classpath返回Hadoop所有相关的JAR包和依赖l

33、下面的命令会添加环境变量：lFlink启动时会从$HADOOP_CLASSPATH中寻找所需依赖包。直接使用已经安装的Hadoop。lFlink所需要的依赖和已经安装的Hadoop提供的依赖有可能发生冲突。添加Hadoop ClasspathexportHADOOP_CLASSPATH=hadoopclasspathlFlink主目录下有一个lib目录，专门存放各类第三方的依赖包。l可以将Hadoop依赖包添加到这个目录下：从Flink官网下载预打包的Hadoop依赖包：一些通用的Hadoop版本从源码编译：一些定制的Hadoop将Hadoop添加到lib文件夹下第六章状态和检查点什么是有状态

34、的计算有状态计算的潜在场景数据去重：需要记录哪些数据已经流入过应用，当新数据流入时，根据已流入数据去重检查输入流是否符合某个特定模式：之前流入的数据以状态的形式缓存下来对一个窗口内的数据进行聚合分析，比如分析一小时内某项指标75分位值或99分位值Flink分布式计算，一个算子有多个算子子任务状态可以被理解为某个算子子任务在当前实例上的一个变量，变量记录了数据流的历史信息，新数据流入，可以结合历史信息来进行计算接收输入流/获取对应状态/更新状态状态管理的难点要解决问题：实时性，延迟不能太高数据不丢不重、恰好计算一次，尤其发生故障恢复后程序的可靠性要高，保证7*24小时稳定运行难点不能将状态直接交

35、由内存，因为内存空间有限用持久化的系统备份状态，出现故障时，如何从备份中恢复需要考虑扩展到多个节点时的伸缩性Flink解决了上述问题，提供有状态的计算APIlManaged State 和 Raw State托管状态（Managed State）是由Flink管理的，Flink帮忙存储、恢复和优化原生状态（Raw State）是开发者自己管理的，需要自己序列化 lManaged State又细分为Keyed State和Operator StateFlink的几种状态类型ManagedStateRawState状态管理方式FlinkRuntime托管，自动存储、自动恢复、自动伸缩用户自己管理状

36、态数据结构Flink提供的常用数据结构，如ListState、MapState等字节数组：byte使用场景绝大多数Flink函数用户自定义函数lKeyed State是KeyedStream上的状态，每个Key共享一个状态lOperator State每个算子子任务共享一个状态Keyed State和Operator StateKeyedState相同Key的数据可以访问、更新这个状态OperatorState流入这个算子子任务的所有数据可以访问、更新这个状态lKeyed State和Operator State都是基于本地的，每个算子子任务维护着自身的状态，不能访问其他算子子任务的状态l具体

37、的实现层面，Keyed State需要重写Rich Function函数类，Operator State需要实现CheckpointedFunction等接口Keyed State和Operator StateKeyedStateOperatorState适用算子类型只适用于KeyedStream上的算子可以用于所有算子状态分配每个Key对应一个状态一个算子子任务对应一个状态创建和访问方式重写RichFunction，通过里面的RuntimeContext访问实现CheckpointedFunction等接口横向扩展状态随着Key自动在多个算子子任务上迁移有多种状态重新分配的方式支持的数据结构

38、ValueState、ListState、MapState等ListState、BroadcastState等l修改Flink应用的并行度：每个算子的并行算子子任务数发生了变化，整个应用需要关停和启动一些算子子任务l某份在原来某个算子子任务的状态需要平滑更新到新的算子子任务上lFlink的Checkpoint可以辅助状态数据在算子子任务之间迁移算子子任务生成快照（Snapshot）保存到分布式存储上子任务重启后，相应的状态在分布式存储上重建（Restore）lKeyed State与Operator State的横向扩展方式稍有不同横向扩展问题lFlink提供了封装好的数据结构供我们使用，包括

39、ValueState、ListState等l主要有：ValueState：单值MapState：Key-Value对ListState：列表ReducingState和AggregatingState：合并lKeyed State由于跟Key绑定，Key自动分布到不同算子子任务，Keyed State也可以根据Key分发到不同算子子任务上Keyed Statel实现RichFunction函数类，比如RichFlatMapFunctionl创建StateDescriptor，StateDescriptor描述状态的名字和状态的数据结构，每种类型的状态有对应的StateDescriptorl通过

40、StateDescriptor，从RuntimeContext中获取状态l调用状态提供的方法，获取状态，更新数据Keyed State/创建StateDescriptorMapStateDescriptorbehaviorMapStateDescriptor=newMapStateDescriptor(behaviorMap,Types.STRING,Types.INT);/通过StateDescriptor获取运行时上下文中的状态behaviorMapState=getRuntimeContext().getMapState(behaviorMapStateDescriptor);MapSt

41、ate：UV get(UK key)void put(UK key,UV value)boolean contains(UK key)l案例：统计电商用户行为UserBehavior场景下，某个用户（userId）下某种用户行为（behavior）的数量Keyed State/*MapStateFunction继承并实现RichFlatMapFunction*两个泛型分别为输入数据类型和输出数据类型*/publicstaticclassMapStateFunctionextendsRichFlatMapFunctionUserBehavior,Tuple3/指向MapState的句柄priva

42、teMapStatebehaviorMapState;Overridepublicvoidopen(Configurationconfiguration)/创建StateDescriptorMapStateDescriptorbehaviorMapStateDescriptor=newMapStateDescriptor(behaviorMap,Types.STRING,Types.INT);/通过StateDescriptor获取运行时上下文中的状态behaviorMapState=getRuntimeContext().getMapState(behaviorMapStateDescrip

43、tor);OverridepublicvoidflatMap(UserBehaviorinput,CollectorTuple3out)throwsExceptionintbehaviorCnt=1;/behavior有可能为pv、cart、fav、buy等/判断状态中是否有该behaviorif(behaviorMapState.contains(input.behavior)behaviorCnt=behaviorMapState.get(input.behavior)+1;/更新状态behaviorMapState.put(input.behavior,behaviorCnt);out.

44、collect(Tuple3.of(input.userId,input.behavior,behaviorCnt);使用MapState记录某个behavior下的数量 lUserBehavior案例l先基于userId进行keyByl再使用有状态的MapStateFunction进行处理Keyed Stateenv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStreamuserBehaviorStream=./生成一个KeyedStreamKeyedStreamkeyedStream=userBehavior

45、Stream.keyBy(user-user.userId);/在KeyedStream上进行flatMapDataStreamTuple3behaviorCountStream=keyedStream.flatMap(newMapStateFunction();l状态：算子子任务的本地数据在Checkpoint过程时写入存储，这个过程被称为备份（Snapshot）初始化或重启一个Flink作业时，以一定逻辑从存储中读出并变为算子子任务的本地数据，这个过程被称为重建（Restore）lKeyed State开箱即用：数据划分基于Key，Snapshot和Restore过程可以基于Key在多个算

46、子子任务之间做数据迁移lOperator State每个算子子任务管理自己的状态，流入到这个算子子任务上的所有数据可以访问和修改Operator State故障重启后，数据流中某个元素不一定流入重启前的算子子任务上需要根据具体业务场景设计Snapshot和Restore的逻辑使用CheckpointedFunction接口类Operator StatelFlink定期执行Checkpoint，会将状态数据Snapshot到存储上l每次执行Snapshot，会调用snapshotState()方法，因此我们要实现一些Snapshot逻辑，比如将哪些状态持久化linitializeState()在

47、算子子任务初始化状态时调用，有两种被调用的可能：整个Flink作业第一次执行，状态数据需要初始化一个默认值Flink作业遇到故障重启，基于之前已经持久化的状态恢复lListState/UnionListStatelBroadcastStateOperator StatepublicinterfaceCheckpointedFunction/Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化voidsnapshotState(FunctionSnapshotContextcontext)throwsException;/初始化时会调用这个方法，向

48、本地状态中填充数据voidinitializeState(FunctionInitializationContextcontext)throwsException;CheckpointedFunction源码l状态以列表的形式序列化并存储l单个状态为S，每个算子子任务有零到多个状态，共同组成一个列表 ListStateS，Snapshot时将这些状态以列表形式写入存储l包含所有状态的大列表，当作业重启时，将这个大列表重新分布到各个算子子任务上lListState：将大列表按照Round-Ribon模式均匀分布到各个算子子任务上，每个算子子任务得到的是大列表的子集lUnionListState：

49、将大列表广播给所有算子子任务l应用场景：Source上保存流入数据的偏移量，Sink上对输出数据做缓存Operator State ListState、UnionListStateOperator State 使用方法l重点实现snapshotState()和initializeState()两个方法l在initializeState()方法里初始化并获取状态注册StateDescriptor，指定状态名字和数据类型从FunctionInitializationContext中获取OperatorStateStore，进而获取Operator Statel在snapshotState()方法里

50、实现一些业务逻辑基于ListState实现可缓存的Sink/重写CheckpointedFunction中的snapshotState()/将本地缓存Snapshot到存储上OverridepublicvoidsnapshotState(FunctionSnapshotContextcontext)throwsException/将之前的Checkpoint清理checkpointedState.clear();for(Tuple2element:bufferedElements)/将最新的数据写到状态中checkpointedState.add(element);/重写Checkpointe