filter function

Scala

object DataStreamTransformationApp {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentfilterFunction(env)env.execute("DataStreamTransformationApp")}def filterFunction(env: StreamExecutionEnvironment): Unit = {val data=env.addSource(new CustomNonParallelSourceFunction)data.map(x=>{println("received:" + x)x}).filter(_%2 == 0).print().setParallelism(1)}}

数据源选择之前的任意一个数据源即可。

这里的map中没有做任何实质性的操作，filter中将所有的数都对2取模操作，打印结果如下:

received:1
received:2
2
received:3
received:4
4
received:5
received:6
6
received:7
received:8
8

说明map中得到的所有的数据，而在filter中进行了过滤操作。

Java

    public static void filterFunction(StreamExecutionEnvironment env) {DataStreamSource<Long> data = env.addSource(new JavaCustomParallelSourceFunction());data.setParallelism(1).map(new MapFunction<Long, Long>() {@Overridepublic Long map(Long value) throws Exception {System.out.println("received:"+value);return value;}}).filter(new FilterFunction<Long>() {@Overridepublic boolean filter(Long value) throws Exception {return value % 2==0;}}).print().setParallelism(1);}

需要先使用data.setParallelism(1)然后再进行map操作，否则会输出多次。因为我们用的是JavaCustomParallelSourceFunction()，而当我们使用JavaCustomNonParallelSourceFunction时，默认就是并行度1，可以不用设置。

Union Function

Scala

  def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment//    filterFunction(env)unionFunction(env)env.execute("DataStreamTransformationApp")}def unionFunction(env: StreamExecutionEnvironment): Unit = {val data01 = env.addSource(new CustomNonParallelSourceFunction)val data02 = env.addSource(new CustomNonParallelSourceFunction)data01.union(data02).print().setParallelism(1)}

Union操作将两个数据集综合起来，可以一同处理，上面打印输出如下：

Java

    public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
//        filterFunction(environment);unionFunction(environment);environment.execute("JavaDataStreamTransformationApp");}public static void unionFunction(StreamExecutionEnvironment env) {DataStreamSource<Long> data1 = env.addSource(new JavaCustomNonParallelSourceFunction());DataStreamSource<Long> data2 = env.addSource(new JavaCustomNonParallelSourceFunction());data1.union(data2).print().setParallelism(1);}

Split Select Function

Scala

split可以将一个流拆成多个流，select可以从多个流中进行选择处理的流。

def splitSelectFunction(env: StreamExecutionEnvironment): Unit = {val data = env.addSource(new CustomNonParallelSourceFunction)val split = data.split(new OutputSelector[Long] {override def select(value: Long): lang.Iterable[String] = {val list = new util.ArrayList[String]()if (value % 2 == 0) {list.add("even")} else {list.add("odd")}list}})split.select("odd","even").print().setParallelism(1)}

可以根据选择的名称来处理数据。

Java

public static void splitSelectFunction(StreamExecutionEnvironment env) {DataStreamSource<Long> data = env.addSource(new JavaCustomNonParallelSourceFunction());SplitStream<Long> split = data.split(new OutputSelector<Long>() {@Overridepublic Iterable<String> select(Long value) {List<String> output = new ArrayList<>();if (value % 2 == 0) {output.add("odd");} else {output.add("even");}return output;}});split.select("odd").print().setParallelism(1);}

Apache Flink 零基础入门（十六）Flink DataStream transformation相关推荐

Apache Flink 零基础入门（六）Flink核心概念
Flink程序是在分布式集合上实现转换的常规程序(例如filtering, mapping, updating state, joining, grouping, defining windows, ...
Apache Flink 零基础入门【转】
Apache Flink 零基础入门(一):基础概念解析 Apache Flink 零基础入门(二):DataStream API 编程转载于:https://www.cnblogs.com/dav ...
Apache Flink 零基础入门（二十）Flink部署与作业的提交
之前我们都是基于Idea在本地进行开发,这种方式很适合开发以及测试,但是开发完之后,如何提交到服务器中运行? Flink单机部署方式本地开发和测试过程中非常有用,只要把代码放到服务器直接运行. 前置 ...
Apache Flink 零基础入门（一）：基础概念解析
Apache Flink 的定义.架构及原理 Apache Flink 是一个分布式大数据处理引擎,可对有限数据流和无限数据流进行有状态或无状态的计算,能够部署在各种集群环境,对各种规模大小的数据进行 ...
Apache Flink 零基础入门（三）编写最简单的helloWorld
实验环境 JDK 1.8 IDE Intellij idea Flink 1.8.1 实验内容创建一个Flink简单Demo,可以从流数据中统计单词个数. 实验步骤首先创建一个maven项目,其中 ...
Apache Flink 零基础入门（十五）Flink DataStream编程（如何自定义DataSource）
数据源可以通过StreamExecutionEnvironment.addSource(sourceFunction)方式来创建,Flink也提供了一些内置的数据源方便使用,例如readTextFil ...
Apache Flink 零基础入门（二十）Flink kafka connector
内置source和sink 内置source包括从文件读取,从文件夹读取,从socket中读取.从集合或者迭代器中读取.内置的sink包括写文件.控制台输出.socket 内置connectors A ...
Apache Flink 零基础入门（十九）Flink windows和Time操作
Time类型在Flink中常用的Time类型: 处理时间摄取时间事件时间处理时间是上图中,最后一步的处理时间,表示服务器中执行相关操作的处理时间.例如一些算子操作时间,在服务器上面的时间. ...
Apache Flink 零基础入门（十八）Flink Table APISQL
什么是Flink关系型API? 虽然Flink已经支持了DataSet和DataStream API,但是有没有一种更好的方式去编程,而不用关心具体的API实现?不需要去了解Java和Scala的具体 ...

Apache Flink 零基础入门（十六）Flink DataStream transformation

filter function

Scala

Java

Union Function

Scala

Java

Split Select Function

Scala

Java

Apache Flink 零基础入门（十六）Flink DataStream transformation相关推荐

最新文章

热门文章