本文由简悦simpread转码原文地址flink常用海量资源

本文由简悦SimpRead转码，原文地址本课时我们主要介绍本文由简悦SimpRead转码，原文地址本课时我们主要介绍Flink的DataSet和DataStream的API，并模拟了实时计算的场景，详细讲解了DataStream常用的API的使用。在前面的课程中，曾经提到过，Flink很重要的一个特点是“流批一体”，然而事实上Flink并没有完全做到所谓的“流批一体”，即编写一套代码，可以同时支持流式计算场景和批量计算的场景。目前截止1.10版本依然采用了DataSet和DataStream两套API来适配不同的应用场景。在官网或者其他网站上，都可以找到目前Flink支持两套API和一些应用场景，但大都缺少了“为什么ApacheFlink器学习等。尤其是在流式计算方面，Flink实现了计算引擎级别的流批一体。那么对于普通开发者而言，如果使用原生的Flink的核心实现类则在flink-streaming-java这个模块。在上述两张图中，我们分别打开DataSet和在上述两张图中，我们分别打开DataSet和DataStream这两个类，可以发现，二者支持的API都非常丰富且十分类似，比如常用的map、filter、join等常见的transformation函数。我们在前面的课时中讲过Flink的编程模型，对于DataSet而言，SourceJava集合；而DataStream的Source部分则一般是消息中间件比如Kafka以下面我们详细讲解DataStreamAPI的使用。我们先来回顾一下Flink的编程模型，在之前的课时中提到过，Flink程序的基础构建模块是（Steams）和转换（ansformations），每一个数据流起始于一个或多个Souc多个Sin。数据流类似于有向无环图（G）。在第02课时中模仿了一个流式计算环境，我们选择监听一个本地的Socket端口，并且使用Flink中的滚动窗口，每5秒打印一次计算结果。在第02课时中模仿了一个流式计算环境，我们选择监听一个本地的Socket端口，并且使用Flink中的滚动窗口，每5秒打印一次计算结果。在本课时中，我们利用Flink提供的自定义Source功能来实现一个自定义的实时数据源，具体实现如publicclassMyStreamingSourceprivatebooleanisRunning=*重写run*@param*@throwspublicvoidrun(SourceContext<Item>ctx)throwsExceptionItemitem=publicvoidcancel()isRunning=Itemitem=publicvoidcancel()isRunning=privateIteminti=newItemitem=newreturnclassprivateStringprivateIntegerItem()publicStringgetName()returnvoidsetName(StringprivateIntegerItem()publicStringgetName()returnvoidsetName(S=privateIntegergetId()returnvoidsetId(Integerid)this.id=publicStringtoString()return"Item{""+name+'\''",id="+idclassStreamingDemopublicstatic",id="+idclassStreamingDemopublicstaticvoidmain(String[]args)throwsExceptionStreamExecutionEnvironmentenvDataStreamSource<MyStreamingSource.Item>textenv.addSource(newDataStream<MyStreamingSource.Item>item=(MapFunction<MyStreamingSource.Item,value->StringjobName="userdefinedstreaming在自定义的数据源中，实现了Flink中的SourceFunction接口，同时实现了其中的run方法，在方法中每隔一秒钟随机发送一个自定义的Item可以直接运行mainDataStreamAPIDataStreamAPIMapclassStreamingDemopublicstaticvoidmain(String[]args)throwsExceptionStreamExecutionEnvironmentenv=MapFunction<MyStreamingSource.Item,Object>(){publicObjectmap(MyStreamingSource.Itemitem)throwsExceptionStringjobName="user我们只取出每个ItemStringjobName="user我们只取出每个Item的name注意，Map算子是最常用的算子之一，官网中的表述是对一个DataStream进行映射，每次进行转换都会调用MapFunction函数。从源DataStream到目标DataStream的转换过程中，返回的是SingleOutputStreamOperator。当然了，我们也可以在重写的map函数中使用lambda表达式。SingleOutputStreamOperator<Object>mapItems=item->甚至，还可以自定义自己的Map函数。通过重写MapFunction或RichMapFunctionmapclassStreamingDemopublicstaticvoidmain(String[]args)throwsExceptionStreamExecutionEnvironmentenvclassStreamingDemopublicstaticvoidmain(String[]args)throwsExceptionStreamExecutionEnvironmentenvDataStreamSource<MyStreamingSource.Item>items=SingleOutputStreamOperator<String>mapItems=StringjobName="userdefinedstreamingstaticclassMyMapFunctionpublicStringmap(MyStreamingSource.ItemreturnRichMapFunctionopen、close等函数方法，重写这些方法还能实现更为复杂的FlatMap接受一个元素，返回零到多个元素。FlatMap和Map候，FlatMap会将列表“平铺FlatMap接受一个元素，返回零到多个元素。FlatMap和Map候，FlatMap会将列表“平铺”，也就是以单个元素的形式进行输出。SingleOutputStreamOperator<Object>flatMapItems=FlatMapFunction<MyStreamingSource.Item,Object>()throwsExceptionStringname=上面的程序会把名字逐个输出。我们也可以在FlatMap需要的数据等。顾名思义，Fliter的意思就是过滤掉不需要的数据，每个元素都会被filter函数处理，如果filter函数返回true则保留，否则丢弃。例如，我们只保留id为偶数的那些itemSingleOutputStreamOperator<MyStreamingSource.Item>filterItems=items.filter(newFilterFunction<MyStreamingSource.Item>(){filterlambdafilterItems=item->item.getId()filterlambdafilterItems=item->item.getId()%2==在介绍KyByKyedStea。数据的某种属性或者单纯某个字段进行分组，然后对不同的组进行不同的处理。举个例子，当我们需要描述一个用户画像时，则需要根据用户的不同行为事件进行加权；再比如，我们在监控双十一的交易大盘时，则需要按照商品的品类进行分组，分别计算销售额。KeyByDataStreamKeyedStreamKeyedStreamDataStream，KeyedStream我们在第02课时中讲解的WordCount程序，曾经使用过DataStream<WordWithCount>windowCounts=.flatMap(newFlatMapFunction<String,WordWithCount>()publicflatMap(Stringvalue,Collector<WordWithCount>word:value.split("\\s"))out.collect(newpublicflatMap(Stringvalue,Collector<WordWithCount>word:value.split("\\s"))out.collect(new.timeWindow(Time.seconds(5),KeyBykey进行分组，那么相同分组的数据会被分发到一个subtask上进行处理，在大数据量和key分布不均匀的时非常容易出现数常见的解决方式是把所有数据加上随机前后缀Aggregationssum、max、min等。Aggregations也需要指定一个key进行聚合，官网给出了几个常见的例子：在上面的这几个函数中，max、min、sum会分别返回最大值、最小值和汇总值；而在上面的这几个函数中，max、min、sum会分别返回最大值、最小值和汇总值；而minBy和maxmaxByStreamExecutionEnvironmentenv=Listdata=newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(new从上图中可以看到，我们希望按照uple3果如我们所料，的确是按照第三个元素大小依次进行的打印，但是结果却出现了一个这样的元素(0,1,2)从上图中可以看到，我们希望按照uple3果如我们所料，的确是按照第三个元素大小依次进行的打印，但是结果却出现了一个这样的元素(0,1,2)，这在我们的源数据中并不存在。我们在FlinkThedifferencebetweenminandminByisthatminreturnstheminimumvalue,minByreturnstheelementthathastheminimumvalueinthisfield(sameformaxand文档中说：min和minBy的区别在于，min会返回我们制定字段的最大值，minBy会返回对应的元素（max和maxBy同理）。网上很多资料也这么写：minminByminminBykey，min和minBy都会返回整个元素，只是min会根据用户指定的字段取最小值，并且把这个值保存在对应的位置，而对于其他的字段，并不能保证其数值正确。max和maxBy同理。事实上，对于Aggegations函数，Flink实际生产环境中应该尽量避免在一个无限流上使用Aggegations。而且，对于同一个yedSteam，只能调用一次Aggregationdata.add(newDataStreamSource<MyStreamingSource.Item>items=StringjobName="userdefinedstreamingReduce函数的原理是，会在每一个分组的yedStReduce函数的原理是，会在每一个分组的yedSteam分组聚合。Listdata=newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newdata.add(newDataStreamSource<Tuple3<Integer,Integer,Integer>>items=SingleOutputStreamOperator<Tuple3<Int