Flink（五）【DataStream 转换算子（上）】

前言

        这节注定是一个大的章节，我预估一下得两三天，涉及到的一些东西不懂就重新学，比如 Lambda 表达式，我只知道 Scala 中很方便，但在 Java 中有点发怵了；一个接口能不能 new 来构造对象? 答案是可以的，匿名内部类嘛。但这些好多都是不用不知道的事情。

        不得不感慨还是学习爽啊，不懂就练，再不懂就问；辛苦自己倒也无妨，可是感情就不一样了，不懂就问？等到问的时候人家就要和你 say goodbye 了 。

1、基本转换算子（map/filter/flatMap）

1.1、map

map 已经是非常熟悉的算子了，在 Scala 中、在 Spark 中，map 的特点就是一进一出。

我们只需要基于 DataStream 调用 map()方法就可以进行转换处理。方法需要传入的参数是接口 MapFunction 的实现；返回值类型还是 DataStream，不过泛型（流中的元素类型）可能改变。

下面我们实现提取上一节的 POJO 类 WaterSensor 中 id 字段的功能。

package com.lyh.transform;import com.lyh.bean.WaterSensor;
import function.MyMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class MapDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));// 三种实现方式:// 1. 匿名内部类 实现MapFunction<T,O>接口 T: 输入类型 O: 输出类型SingleOutputStreamOperator<String> map = sensorDS.map(new MapFunction<WaterSensor, String>() {@Overridepublic String map(WaterSensor waterSensor) throws Exception {return waterSensor.getId();}});// 2. lambda 表达式sensorDS.map(sensor -> sensor.getId()); // 如果是一行 可以省去大括号和returnsensorDS.map(sensor -> {return sensor.getId();  // 如果是多行 加个大括号 需要return});sensorDS.map(WaterSensor::getId); // 方法与构造函数引用 JDK8新特性// 3. 定义一个类来实现 MapFunctionsensorDS.map(new MyMapFunctionImpl());map.print();env.execute();}
}

匿名内部类和lambda表达式虽然简单快捷，但是实际开发中如果说我们的的这个类需要应用到多个场景中的话还是最好单独建个包（比如function），然后单独定义这个类：

package function;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;public class MyMapFunctionImpl implements MapFunction<WaterSensor,String> {@Overridepublic String map(WaterSensor waterSensor) throws Exception {return waterSensor.getId();}
}

这么写是最规范的写法，也是良好的开发习惯。 

1.2、filter

filter 转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。

进行 filter 转换之后的新数据流的数据类型与原数据流是相同的。filter 转换需要传入的参数需要实现 FilterFunction 接口，而 FilterFunction 内要实现 filter()方法，就相当于一个返回布尔类型的条件表达式。

查看 FilterFunction 接口的源码，我们发现，这又是一个函数接口（只有一个抽象方法，且被@FunctionalInterface 标注），那么实现就很容易了，我们可以使用 lambda 表达式。

案例：过滤 Id 为 "s1" 的传感器。

package com.lyh.transform;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class FilterDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));// 过滤 "s1" 的三种实现方式:// 1. 匿名内部类 实现FilterFunction<T>接口 T: 要过滤的数据类型sensorDS.filter(new FilterFunction<WaterSensor>() {@Overridepublic boolean filter(WaterSensor waterSensor) throws Exception {return waterSensor.getId().equals("s1");}});// 2. lambda 表达式SingleOutputStreamOperator<WaterSensor> res = sensorDS.filter((waterSensor) -> waterSensor.getId().equals("s1"));// 3. 定义一个类来实现 FilterFunction// 这里省略res.print();env.execute();}
}

运行结果：

WaterSensor{id='s1', ts=1, vc=1}

可以看到，id 为 s1 的传感器被过滤输出了。 

1.3、flatMap

flatMap 操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap 可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。

        同 map 一样，flatMap 也可以使用 Lambda 表达式或者 FlatMapFunction 接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。flatMap 操作会应用在每一个输入事件上面，FlatMapFunction 接口中定义了 flatMap 方法，用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回 0 个、1 个或多个结果数据。因此 flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。所以 flatMap 方法也可以实现 map 方法和 filter 方法的功能，当返回结果是 0 个的时候，就相当于对数据进行了过滤，当返回结果是 1 个的时候，相当于对数据进行了简单的转换操作。

案例 - 如果输入的数据是 sensor_1 ,只打印 vc ,如果输入的是 sensor_2, 既打印 ts 又打印 vc。

package com.lyh.transform;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;// 如果输入的数据是 sensor_1 ,只打印 vc ,如果输入的是 sensor_2, 既打印 ts 又打印 vc
public class FlatMapDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));// 过滤 "s1" 的三种实现方式:// 1. 匿名内部类 实现FilterFunction<T>接口 T: 要过滤的数据类型sensorDS.flatMap(new FlatMapFunction<WaterSensor, String>() {@Overridepublic void flatMap(WaterSensor sensor, Collector<String> out) throws Exception {if (sensor.getId().equals("s1")){out.collect(sensor.getVc()+"");}else if (sensor.getId().equals("s2")){out.collect(sensor.getTs()+"\n"+sensor.getVc());}}});// 2. lambda 表达式SingleOutputStreamOperator<String> res = sensorDS.flatMap((WaterSensor sensor,Collector<String> out) -> {if (sensor.getId().equals("s1")) {out.collect(sensor.getVc() + "");} else if (sensor.getId().equals("s2")) {out.collect(sensor.getTs() + "\n" + sensor.getVc());}}).returns(Types.STRING()); // 需要注明返回类型// 3. 定义一个类来实现 FilterFunction// 这里省略res.print();env.execute();}
}

        注意：这次需要对返回结果的类型进行声明，flatMap 和 map（一进一出）、filter（一进一出，因为返回的结果是一个 boolean 值） 是有点不一样的，因为 flatMap 是可以一进0出、一进一出甚至一进多出的。从上面的代码中我们可以看到， faltMap 方法中是通过采集器输出的，所以就可以通过判断语句实现不同数量的输出。

输出结果：

1
2
2

2、聚合算子（Aggregation）

        直观上看，基本转换算子确实是在“转换”——因为它们都是基于当前数据，去做了处理和输出。而在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的信息。

        聚合对应着我们 MapReduce 中的 reduce 这一步。

2.1、keyBy（按键分区）

        我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区；这个操作就是通过 keyBy 来完成的。

        keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
        基于不同的 key，流中的数据将被分配到不同的分区中去，这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot中进行处理了。

        在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。所以这里 key 如果是 POJO 的话，必须要重写 hashCode()方法。keyBy()方法需要传入一个参数，这个参数指定了一个或一组 key。有很多不同的方法来指定 key：比如对于 Tuple 数据类型，可以指定字段的位置或者多个位置的组合；对于 POJO 类型，可以指定字段的名称（String）；另外，还可以传入 Lambda 表达式或者实现一个键选择器（KeySelector），用于说明从数据中提取 key 的逻辑。

案例 - 根据 WaterSensor 的 id 进行分组（注意：这里是分组不是分区，一个分区可以存在多个组，因为 keyBy 只保证相同组在一个分区）。 

package com.lyh.aggregation;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class KeyByDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));/*** 按照 id 来进行分区* 因为KeyBy不是转换算子,所以它返回一个 键控流,仍然是一个数据流类型*/sensorDS.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor sensor) throws Exception {return sensor.getId();}});// lambda 表达式KeyedStream<WaterSensor, String> keyedStream = sensorDS.keyBy((sensor) -> sensor.getId());keyedStream.print();env.execute();}
}

 运行结果：

2> WaterSensor{id='s1', ts=1, vc=1}
1> WaterSensor{id='s2', ts=2, vc=2}
2> WaterSensor{id='s3', ts=3, vc=3}

这次我们把并行度分为2，当我们调用 Sink 算子-print 的时候，它就会按照分区并行对数据进行输出，同时，它保证了相同 key 的数据在同一个分区中；当然，多个组也可以在同一个分区的，因为它只保证一个组的数据在同一个分区，没说不同组不可以在同一个分区；比如 hash值%分区数 结果相同的多个组也会被分到同一个组中。

2.2、sum/min/max/minBy/maxBy

有了按键分组的数据流 KeyedStream，我们就可以基于它进行聚合操作了。在 Flink 中，聚合算子和 keyBy 是成对出现的，因为把具有相同结构的数据放在一起（keyBy）再做聚合运算（聚合算子sum...）才是有意义的。就像我们 Hive 中经常把 groupBy 和 聚合函数放在一起。

  •  sum()：在输入流上，对指定的字段做叠加求和的操作。
  •  min()：在输入流上，对指定的字段求最小值。
  •  max()：在输入流上，对指定的字段求最大值。
  •  minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。
  •  maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。

所以在 Flink 中，聚合算子是必须在 KeyBy 之后才能出现的，它做的是组内的聚合。

package com.lyh.aggregation;import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SimpleAggregationDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s1", 10L, 5),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));KeyedStream<WaterSensor, String> keyedStream = sensorDS.keyBy((sensor) -> sensor.getId());/*** sum 的两种参数类型:*  1.按照位置进行聚合:* 之前我们对于元组类型Tuple2是直接可以使用位置来进行聚合的 但是POJO 类型是不允许的*  2.按照字段名进行聚合:*  对于 POJO 类型我们需要传入它的字段名*/
//        keyedStream.sum("vc").print();
//        keyedStream.min("vc").print();/*** max 与 maxBy 的区别*  -max 只会取比较字段的最大值，非比较字段保留第一次的值*  -maxBy 如果该数据对象的比较字段是新的最大值 直接取该数据对象*/keyedStream.max("vc").print();keyedStream.maxBy("vc").print();env.execute();}
}

这里要注意的就是 max/maxBy、min/minBy 的区别了。

输入数据：
WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=10, vc=12}
WaterSensor{id='s2', ts=2, vc=2}
WaterSensor{id='s3', ts=3, vc=3}max("vc"):
WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=1, vc=12}    //这里只替换比较字段的值
WaterSensor{id='s2', ts=2, vc=2}
WaterSensor{id='s3', ts=3, vc=3}maxBy("vd"):
WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=10, vc=12}    //这里发现vc有新的最大值 直接替换整个对象
WaterSensor{id='s2', ts=2, vc=2}
WaterSensor{id='s3', ts=3, vc=3}

2.3、reduce （规约聚合）

reduce 同样必须先 keyBy，再对相同组内的数据进行规约聚合，reduce 方法的特点是每个 key 的第一条数据来的时候不会立即执行 reduce 方法，而是存起来，直接输出；需要注意的是reduce的输出类型必须和输入类型一致。

reduce 方法需要传入一个实现 ReduceFunction 接口的对象，我们查看源码可以发现，这又双叒叕是一个函数接口，所以我们依然用lambda表达式这样最简洁。

我们用 reduce 实现 maxBy 的功能：

package com.lyh.aggregation;import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class ReduceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s1", 10L, 5),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));KeyedStream<WaterSensor, String> keyedStream = sensorDS.keyBy((sensor) -> sensor.getId());/*** reduce 同样必须在 keyBy 之后使用* 两两聚合 输入类型必须=输出类型*/SingleOutputStreamOperator<WaterSensor> reduce = keyedStream.reduce((sensor1, sensor2) -> new WaterSensor(sensor1.id, sensor2.ts, sensor1.getVc() + sensor2.getVc()));reduce.print();env.execute();}
}

输出结果：

WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=10, vc=6}
WaterSensor{id='s2', ts=2, vc=2}
WaterSensor{id='s3', ts=3, vc=3}

 显然利用 reduce 算子我们可以更加灵活地对数据进行处理，比如我们修该上面的代码：

SingleOutputStreamOperator<WaterSensor> reduce = keyedStream.reduce((sensor1, sensor2) -> new WaterSensor(sensor1.id, sensor1.ts+sensor2.ts, sensor1.getVc() + sensor2.getVc()));

这样我们就可以对来的数据的每个字段进行累加。

什么是有状态计算？

比如这里我们 reduce 会把每次处理的结果（状态）存下来，而且它自己会来维护，所以当相同 key 的数据再次来的时候，它才能对新旧数据再次进行聚合。

我们 reduce(value1,value2) 中，value1 指的就是之前的计算结果，也就是状态。而 value2 指的就是新来的数据。 

3、用户自定义函数（UDF）

有点像我们 Hive 的 UDF ，更加灵活。

3.1、函数类

对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类，例如 MapFunction、FilterFunction、ReduceFunction 等。

前面对于 POJO 类型的数据，我们用匿名内部类、lambda 和 单独一个实现类都实现过。

案例

这里我们再写一个简单的自定义函数 - FilterFuctionImpl ，我们希望可以通过自定义的字段名来对传进来的数据进行过滤（之前我们是写死只能过滤某个特定 id 的 WaterSensor 对象）。

package com.lyh.udf;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;public class FilterFunctionImpl implements FilterFunction<WaterSensor> {public String sensor_id;public FilterFunctionImpl(String id) {this.sensor_id = id;}@Overridepublic boolean filter(WaterSensor sensor) throws Exception {return this.sensor_id.equals(sensor.getId());}
}

我们修改上面 1.2 中的 FilterDemo 代码：

SingleOutputStreamOperator<WaterSensor> res = sensorDS.filter(new FilterFunctionImpl("s2"));

运行结果：

WaterSensor{id='s2', ts=2, vc=2}

3.2、富函数

        “富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction 等。

        既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

生命周期的概念在编程中其实非常重要，到处都有体现。例如：对于 C 语言来说，我们需要手动管理内存的分配和回收，也就是手动管理内存的生命周期。分配内存而不回收，会造成内存泄漏，回收没有分配过的内存，会造成空指针异常。而在 JVM 中，虚拟机会自动帮助我们管理对象的生命周期。对于前端来说，一个页面也会有生命周期。数据库连接、网络连接以及文件描述符的创建和关闭，也都形成了生命周期。所以生命周期的概念在编程中是无处不在的，需要我们多加注意。

Rich Function 有生命周期的概念。典型的生命周期方法有：

  • open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调用。所以像文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在 open()方法中完成。。
  • close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。

需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如 RichMapFunction 中的 map()，在每条数据到来后都会触发一次调用。

我们可以看到，RichMapFunction 是一个抽象类，它继承了 AbstractRichFunction 这个抽象类和 MapFunction 这个函数接口。

而 AbstractRichFunction 这个抽象类又为它提供了好多关于生命周期、上下文的管理方法。 

案例

我们先用有界数据流来看一下这些方法是什么时候被调用的，这里先不去设置并行度（本地环境默认等于机器的cpu核数）。

package com.lyh.udf;import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class RichFunctionDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<Integer> ds = env.fromElements(1, 2, 3, 4);SingleOutputStreamOperator<Integer> map = ds.map(new RichMapFunction<Integer, Integer>() {@Overridepublic Integer map(Integer value) throws Exception {return value + 1;}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);RuntimeContext runtimeContext = getRuntimeContext();    // 上下文对象int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();    // 子任务的编号(类似于我们线程的编号 从0开始)String taskNameWithSubtasks = runtimeContext.getTaskNameWithSubtasks();System.out.println("子任务线程编号: "+indexOfThisSubtask+" 启动, 子任务名称: "+taskNameWithSubtasks+"调用 open()");}@Overridepublic void close() throws Exception {super.close();RuntimeContext runtimeContext = getRuntimeContext();    // 上下文对象int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();    // 子任务的编号(类似于我们线程的编号 从0开始)String taskNameWithSubtasks = runtimeContext.getTaskNameWithSubtasks();System.out.println("子任务线程编号: "+indexOfThisSubtask+" 启动, 子任务名称: "+taskNameWithSubtasks+"调用 close()");}});map.print();env.execute();}
}

 运行结果：

子任务线程编号: 4 启动, 子任务名称: Map -> Sink: Print to Std. Out (5/16)#0调用 open()
子任务线程编号: 14 启动, 子任务名称: Map -> Sink: Print to Std. Out (15/16)#0调用 open()
子任务线程编号: 7 启动, 子任务名称: Map -> Sink: Print to Std. Out (8/16)#0调用 open()
子任务线程编号: 5 启动, 子任务名称: Map -> Sink: Print to Std. Out (6/16)#0调用 open()
子任务线程编号: 8 启动, 子任务名称: Map -> Sink: Print to Std. Out (9/16)#0调用 open()
子任务线程编号: 6 启动, 子任务名称: Map -> Sink: Print to Std. Out (7/16)#0调用 open()
子任务线程编号: 2 启动, 子任务名称: Map -> Sink: Print to Std. Out (3/16)#0调用 open()
子任务线程编号: 13 启动, 子任务名称: Map -> Sink: Print to Std. Out (14/16)#0调用 open()
子任务线程编号: 15 启动, 子任务名称: Map -> Sink: Print to Std. Out (16/16)#0调用 open()
子任务线程编号: 9 启动, 子任务名称: Map -> Sink: Print to Std. Out (10/16)#0调用 open()
子任务线程编号: 11 启动, 子任务名称: Map -> Sink: Print to Std. Out (12/16)#0调用 open()
子任务线程编号: 12 启动, 子任务名称: Map -> Sink: Print to Std. Out (13/16)#0调用 open()
子任务线程编号: 0 启动, 子任务名称: Map -> Sink: Print to Std. Out (1/16)#0调用 open()
子任务线程编号: 10 启动, 子任务名称: Map -> Sink: Print to Std. Out (11/16)#0调用 open()
子任务线程编号: 3 启动, 子任务名称: Map -> Sink: Print to Std. Out (4/16)#0调用 open()
子任务线程编号: 1 启动, 子任务名称: Map -> Sink: Print to Std. Out (2/16)#0调用 open()
12> 4
11> 3
10> 2
13> 5
子任务线程编号: 0 启动, 子任务名称: Map -> Sink: Print to Std. Out (1/16)#0调用 close()
子任务线程编号: 13 启动, 子任务名称: Map -> Sink: Print to Std. Out (14/16)#0调用 close()
子任务线程编号: 4 启动, 子任务名称: Map -> Sink: Print to Std. Out (5/16)#0调用 close()
子任务线程编号: 7 启动, 子任务名称: Map -> Sink: Print to Std. Out (8/16)#0调用 close()
子任务线程编号: 1 启动, 子任务名称: Map -> Sink: Print to Std. Out (2/16)#0调用 close()
子任务线程编号: 11 启动, 子任务名称: Map -> Sink: Print to Std. Out (12/16)#0调用 close()
子任务线程编号: 10 启动, 子任务名称: Map -> Sink: Print to Std. Out (11/16)#0调用 close()
子任务线程编号: 8 启动, 子任务名称: Map -> Sink: Print to Std. Out (9/16)#0调用 close()
子任务线程编号: 5 启动, 子任务名称: Map -> Sink: Print to Std. Out (6/16)#0调用 close()
子任务线程编号: 14 启动, 子任务名称: Map -> Sink: Print to Std. Out (15/16)#0调用 close()
子任务线程编号: 2 启动, 子任务名称: Map -> Sink: Print to Std. Out (3/16)#0调用 close()
子任务线程编号: 6 启动, 子任务名称: Map -> Sink: Print to Std. Out (7/16)#0调用 close()
子任务线程编号: 12 启动, 子任务名称: Map -> Sink: Print to Std. Out (13/16)#0调用 close()
子任务线程编号: 15 启动, 子任务名称: Map -> Sink: Print to Std. Out (16/16)#0调用 close()
子任务线程编号: 3 启动, 子任务名称: Map -> Sink: Print to Std. Out (4/16)#0调用 close()
子任务线程编号: 9 启动, 子任务名称: Map -> Sink: Print to Std. Out (10/16)#0调用 close()

 我们再用无界数据流进行模拟，这次设置并行度为 2，修改上面的代码：

DataStreamSource<String> ds = env.socketTextStream("hadoop102",9999);SingleOutputStreamOperator<Integer> map = ds.map(new RichMapFunction<String, Integer>() {@Overridepublic Integer map(String value) throws Exception {return Integer.parseInt(value) + 1;}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);System.out.println("子任务线程编号: "+getRuntimeContext().getIndexOfThisSubtask()+" 启动, 子任务名称: "+getRuntimeContext().getTaskNameWithSubtasks()+"调用 open()");}@Overridepublic void close() throws Exception {super.close();System.out.println("子任务线程编号: "+getRuntimeContext().getIndexOfThisSubtask()+" 启动, 子任务名称: "+getRuntimeContext().getTaskNameWithSubtasks()+"调用 close()");}});

结论

RichXXXFunction：富函数

  • 多了生命周期管理方法：

    • open（）：每个子任务，在启动时调用一次

    • close（）：每个子任务，在结束时调用一次

      • 如果是flink程序异常挂掉，不会调用 close（）

      • 如果是正常调用 flink 的 cancel 命令或者web ui 端的 cancel ，可以 close（）

  • 多了运行时上下文
    • 可以获取一些运行时的环境信息，比如 子任务编号、名称 ......


以后我们开发的时候，如果 Flink 程序启动时需要执行一次，关闭时需要执行一次的场景下就可以使用这些富函数了。

总结

        剩下的部分下次完成。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/144120.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Vue模板语法

Vue模板语法

模板语法有两大类&#xff1a; 1.插值语法 2.指令语法 让我为大家介绍一下吧&#xff01; 一、插值语法 功能:用于解析标签体内容。 写法: {{xxx}}&#xff0c;xxx是js表达式&#xff0c;且可以直接读取到data中的所有属性。 举个例子&#xff1a; <!DOCTYPE html> &l…
阅读更多...
idea maven 构建本地jar包及pom文件

idea maven 构建本地jar包及pom文件

1、设置模块build 本地输出路径 <build><defaultGoal>compile</defaultGoal><resources><resource><directory>${basedir}/src/main/resources</directory><includes><include>**/**</include></includes>…
阅读更多...
腾讯云服务器可用区是什么意思？

腾讯云服务器可用区是什么意思？

腾讯云服务器可用区是什么意思&#xff1f;云服务器可用区如何选择&#xff1f;可用区是指在同一个地域内电力和网络相互独立的区域&#xff0c;可用区可以做到故障隔离&#xff0c;所以可用区存在的意义在于构建高可用、高容灾应用&#xff0c;将应用部署在不同可用区内&#…
阅读更多...
ChatkBQA：一个基于大语言模型的知识库问题生成-检索框架11.13

ChatkBQA：一个基于大语言模型的知识库问题生成-检索框架11.13

ChatkBQA&#xff1a;一个基于大语言模型的知识库问题生成-检索框架 摘要1 引言3 准备工作4 方法4.1 ChatKBQA概述4.2 在LLMS上进行高效微调4.3 用微调LLMS生成逻辑形式4.4 实体和关系的非监督检索4.5 可解释查询执行 摘要 基于知识的问答&#xff08;KBQA&#xff09;旨在从大…
阅读更多...
Windows Server 2012 R2系统服务器远程桌面服务多用户登录配置分享

Windows Server 2012 R2系统服务器远程桌面服务多用户登录配置分享

Windows Server 2012系统在没有安装远程多界面的情况下&#xff0c;最多只能同时运行2个远程桌面&#xff0c;如果是有多个技术员、合伙人同时操作或是像游戏开发需要用到多界面&#xff0c;但是没有安装就很不方便&#xff0c;今天飞飞来和你们分享Windows server 2012R2系统远…
阅读更多...
C++ opencv基本用法【学习笔记（九）】

C++ opencv基本用法【学习笔记（九）】

这篇博客为修改过后的转载&#xff0c;因为没有转载链接&#xff0c;所以选了原创 文章目录 一、vs code 结合Cmake debug1.1 配置tasks.json1.2 配置launch.json 二、图片、视频、摄像头读取显示2.1 读取图片并显示2.2 读取视频文件并显示2.3 读取摄像头并写入文件 三、图片基…
阅读更多...
通过注释来埋点

通过注释来埋点

目录 开始 插件编写 功能一 功能二 功能三 合并功能 运行代码 总结 这篇文章主要讲如何根据注释&#xff0c;通过babel插件自动地&#xff0c;给相应函数插入埋点代码&#xff0c;在实现埋点逻辑和业务逻辑分离的基础上&#xff0c;配置更加灵活 这篇文章想要达到的效…
阅读更多...
2023最受推荐的五款项目管理工具

2023最受推荐的五款项目管理工具

1、进度猫 进度猫是国内一款轻量级项目管理工具&#xff0c;适用于实时协作的团队。 以甘特图为向导&#xff0c;基于任务清单todolist&#xff0c;支持多用户协作&#xff1b; 甘特图显示具体任务清单、时间和任务的进度&#xff1b; 对未完成任务、已完成任务进行分类管…
阅读更多...
Vue3源码reactive和readonly对象嵌套转换，及实现shallowReadonly

Vue3源码reactive和readonly对象嵌套转换，及实现shallowReadonly

前言 官方文档中对reactive的描述&#xff1a; 响应式转换是“深层”的&#xff1a;它会影响到所有嵌套的属性。一个响应式对象也将深层地解包任何 ref 属性&#xff0c;同时保持响应性。 官方文档中对readonly的描述: 只读代理是深层的&#xff1a;对任何嵌套属性的访问都将是…
阅读更多...
深入了解域名与SSL证书的关系

深入了解域名与SSL证书的关系

在如今数字化的世界里&#xff0c;网络安全成为我们关注的重要议题之一。为了确保数据在网络上传输的安全性&#xff0c;我们通常会采取各种安全措施&#xff0c;其中最常用的就是SSL证书。然而&#xff0c;很多人并不了解SSL证书是如何与域名相互关联的。 首先&#xff0c;我…
阅读更多...
[C/C++] 数据结构 链表OJ题:相交链表(寻找两个链表的相交起始结点)

[C/C++] 数据结构 链表OJ题:相交链表(寻找两个链表的相交起始结点)

题目描述: 给你两个单链表的头节点 headA 和 headB &#xff0c;请你找出并返回两个单链表相交的起始节点。如果两个链表不存在相交节点&#xff0c;返回 null 。 图示两个链表在节点 c1 开始相交&#xff1a; 题目数据 保证 整个链式结构中不存在环。 注意&#xff0c;函数返…
阅读更多...
HTML简单介绍

HTML简单介绍

且视他人之疑目如盏盏鬼火&#xff0c;大胆地去你的夜路。 目录 1.网页 2.Web标准 3.HTML 3.1HTML结构 3.2HTML标签​编辑 4.标签介绍 4.1排版标签 4.2文本格式化标签 4.3媒体标签 4.3.1图片标签 4.3.2 音频标签 4.3.3视频标签 5.相对路径 6.链接标签 6.1target属…
阅读更多...
图论15-有向图-环检测+度数+欧拉回路

图论15-有向图-环检测+度数+欧拉回路

文章目录 1. 有向图设计1.1 私有变量标记是否有向1.2 添加边的处理&#xff0c;双向变单向1.3 删除边的处理&#xff0c;双向变单向1.4 有向图的出度和入度 2 有向图的环检测2.1 普通的算法实现换检测2.2 拓扑排序中的环检测 3 欧拉回路 1. 有向图设计 1.1 私有变量标记是否有…
阅读更多...
燃气管网监测系统|全面保障燃气安全

燃气管网监测系统|全面保障燃气安全

根据新华日报的报道&#xff0c;2023年上半年&#xff0c;我国共发生了294起燃气事故&#xff0c;造成了57人死亡和190人受伤&#xff0c;燃气事故的发生原因有很多&#xff0c;其中涉及到燃气泄漏、设备故障等因素。因此&#xff0c;加强燃气安全管理&#xff0c;提高城市的安…
阅读更多...
电磁场与电磁波part1--矢量分析

电磁场与电磁波part1--矢量分析

目录 1、方向导数 2、散度定理&#xff08;高斯定理&#xff09; 3、散度与旋度的比较 4、旋度定理&#xff08;斯托克斯定理&#xff09; 5、关于点乘、叉乘、梯度、散度、旋度的计算 ~~~~~~~~~~~~~~~~~~~~~~~~ 确认过眼神&#xff0c;是我看不懂的 ~~~~~~~~~~~~~~~~…
阅读更多...
IDEA没有Add Framework Support解决办法

IDEA没有Add Framework Support解决办法

点击File—>Settings 点击第一个设置快捷键 点击apply和ok即可 我们要点击一下项目&#xff0c;再按快捷键ctrlk 即可
阅读更多...
C++入门(2)—函数重载、引用

C++入门(2)—函数重载、引用

目录 一、函数重载 1、参数类型不同 2、参数个数不同 3、参数顺序不同 4、 链接中如何区分函数重载 二、引用 1、规则 2、特征 3、使用场景 做参数 做返回值 4、常引用 5、传值、传引用效率比较 6、引用和指针的区别 接上一小节C入门(1)—命名空间、缺省参数 一…
阅读更多...
Python 如何实现适配器设计模式？什么是适配器（Adapter）设计模式？

Python 如何实现适配器设计模式？什么是适配器（Adapter）设计模式？

什么是适配器设计模式&#xff1f; 适配器&#xff08;Adapter&#xff09;设计模式是一种结构型设计模式&#xff0c;它允许接口不兼容的类之间进行合作。适配器模式充当两个不兼容接口之间的桥梁&#xff0c;使得它们可以一起工作&#xff0c;而无需修改它们的源代码。 主要…
阅读更多...
C++ 基础

C++ 基础

准备工具Vscode或者Clion或者Dev C或者Vs studio 和 MSYS2 是C跨平台的重要工具链. 基础一 准备工作安装MSYS2软件 创建文件 一、基本介绍1.1C源文件1.2 代码注释1.3变量与常量1.3.1变量1.3.2 常量1.3.3 二者的区别&#xff1a; 1.4 关键字和标识符 二、数据类型2.1 基本数据类…
阅读更多...
C/C++ 实现获取硬盘序列号

C/C++ 实现获取硬盘序列号

获取硬盘的序列号、型号和固件版本号&#xff0c;此类功能通常用于做硬盘绑定或硬件验证操作&#xff0c;通过使用Windows API的DeviceIoControl函数与物理硬盘驱动程序进行通信&#xff0c;发送ATA命令来获取硬盘的信息。 以下是该程序的主要功能和流程&#xff1a; 定义常量…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023