（十）Flink Datastream API 编程指南 算子-1 （转换算子、物理分区、任务链、资源组 、算子和作业）等基本介绍

概览

操作符将一个或多个DataStream转换为一个新的DataStream。程序可以将多个转换组合成复杂的数据流拓扑。

本节将描述基本的转换、应用这些转换后的有效物理分区以及对Flink的 operator chain（链）的深入了解。

DataStream 各种转换算子 Map

DataStream → DataStream
获取一个元素并生成一个元素。将输入流的值加倍的map函数:

DataStream dataStream = //...
dataStream.map(new MapFunction() {
    @Override
    public Integer map(Integer value) throws Exception {
        return 2 * value;
    }
});

flatMap

DataStream → DataStream
接受一个元素并生成0、1或多个元素。一个将句子分割为单词的flatmap函数:

dataStream.flatMap(new FlatMapFunction(){
	@Override
	public void flatMap(String value,Collector out) throw Exception{
		for(String word: value.split(" ")){
			out.collect(word);
		}
	}
})

Filter

DataStream → DataStream
对每个元素求布尔函数值，并保留函数返回true的元素。一种过滤掉零值的过滤器:

SingleOutputStreamOperator filter = dataStream.filter(new FilterFunction() {
            @Override
            public boolean filter(String s) throws Exception {
                return s.length() > 0;
            }
        });

KeyBy

DataStream → KeyedStream
逻辑地将流划分为不相交的分区。具有相同键的所有记录被分配到相同的分区。在内部，keyBy()是通过散列分区实现的。有不同的方法可以指定键。

dataStream.keyBy(value -> value.getSomeKey());
dataStream.keyBy(value -> value.f0)

在以下情况下，类型不能是键:

• 它是POJO类型，但不覆盖hashCode()方法，并依赖于Object.hashCode()实现。
• 它是任何类型的数组。

Reduce

KeyedStream → DataStream
对键控数据流的“滚动”减少。将当前元素与最后减少的值合并并发出新值。

keyedStream.reduce(new ReduceFunction() {
    @Override
    public Integer reduce(Integer value1, Integer value2)
    throws Exception {
        return value1 + value2;
    }
});

Window

KeyedStream → WindowedStream
Windows可以在已经分区的KeyedStreams上定义。Windows根据某些特征(例如，最近5秒内到达的数据)对每个键中的数据进行分组。有关窗口的完整描述，请参阅窗口。

dataStream
  .keyBy(value -> value.f0)
  .window(TumblingEventTimeWindows.of(Time.seconds(5))); 

WindowAll

DataStream → AllWindowedStream
Windows可以在常规的datastream上定义。Windows根据某些特征(例如，最近5秒内到达的数据)对所有流事件进行分组。有关窗口的完整描述，请参阅窗口。

在很多情况下，这是一个非平行变换。所有记录将被收集到窗口operator的一个任务中,也就是说单并行度下处理。

dataStream
  .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

Window Apply

WindowedStream → DataStream
AllWindowedStream → DataStream

将通用函数应用于整个窗口。下面是一个手动求和窗口元素的函数。

如果你正在使用一个窗口转换，你需要使用一个AllWindowFunction代替。

windowedStream.apply(new WindowFunction, Integer, Tuple, Window>() {
    public void apply (Tuple tuple,
            Window window,
            Iterable> values,
            Collector out) throws Exception {
        int sum = 0;
        for (value t: values) {
            sum += t.f1;
        }
        out.collect (new Integer(sum));
    }
});

// applying an AllWindowFunction on non-keyed window stream
allWindowedStream.apply (new AllWindowFunction, Integer, Window>() {
    public void apply (Window window,
            Iterable> values,
            Collector out) throws Exception {
        int sum = 0;
        for (value t: values) {
            sum += t.f1;
        }
        out.collect (new Integer(sum));
    }
});

WindowReduce

WindowedStream → DataStream
对窗口应用函数reduce函数，并返回经过简化的值。

windowedStream.reduce (new ReduceFunction>() {
    public Tuple2 reduce(Tuple2 value1, Tuple2 value2) throws Exception {
        return new Tuple2(value1.f0, value1.f1 + value2.f1);
    }
});

Union

DataStream* → DataStream
将两个或多个数据流合并，创建一个包含所有数据流中所有元素的新数据流。注意:如果你将一个数据流与它自身结合，你将得到结果流中的每个元素两次。

dataStream.union(otherStream1, otherStream2, ...);

Window Join

DataStream,DataStream → DataStream
将给定键上的两个数据流和一个公共窗口连接起来。

dataStream.join(otherStream)
    .where().equalTo()
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...});

for example:
# socket数据源
SingleOutputStreamOperator map1 = env.socketTextStream(ip, 9998).map((MapFunction) Integer::parseInt);
        SingleOutputStreamOperator map2 = env.socketTextStream(ip, 9997).map((MapFunction) Integer::parseInt);
        
        # 滚动窗口join
        DataStream windowJoin = map1.join(map2)
                .where((KeySelector) integer -> integer)
                .equalTo((KeySelector) integer -> integer)
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .apply(Integer::sum);

		# 滑动窗口join
        DataStream windowJoinOfSlid = map1.join(map2).where((KeySelector) integer -> integer)
                .equalTo((KeySelector) integer -> integer)
                .window(SlidingEventTimeWindows.of(Time.seconds(3), Time.seconds(3), Time.seconds(3)))
                .apply(Integer::sum);

Interval Join

KeyedStream,KeyedStream → DataStream
在给定的时间间隔内将两个键化流的元素e1和e2用一个通用键连接起来，因此e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound.

// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
keyedStream.intervalJoin(otherKeyedStream)
    .between(Time.milliseconds(-2), Time.milliseconds(2)) // lower and upper bound
    .upperBoundExclusive(true) // optional
    .lowerBoundExclusive(true) // optional
    .process(new IntervalJoinFunction() {...});

Window CoGroup

DataStream,DataStream → DataStream
将给定键和公共窗口上的两个数据流进行Cogroups。

dataStream.coGroup(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new CoGroupFunction () {...});

Connect

DataStream,DataStream → ConnectedStream
连接”两个保持其类型的数据流。连接允许两个流之间共享状态。

DataStream someStream = //...
DataStream otherStream = //...

ConnectedStreams connectedStreams = someStream.connect(otherStream);

CoMap, CoFlatMap

ConnectedStream → DataStream
类似于连接数据流上的map和flatMap

connectedStreams.map(new CoMapFunction() {
    @Override
    public Boolean map1(Integer value) {
        return true;
    }

    @Override
    public Boolean map2(String value) {
        return false;
    }
});
connectedStreams.flatMap(new CoFlatMapFunction() {

   @Override
   public void flatMap1(Integer value, Collector out) {
       out.collect(value.toString());
   }

   @Override
   public void flatMap2(String value, Collector out) {
       for (String word: value.split(" ")) {
         out.collect(word);
       }
   }
});

Iterate

DataStream → IterativeStream → ConnectedStream
通过将一个operator的输出重定向到之前的某个operator，在流中创建一个“feedback”循环。这对于定义不断更新模型的算法特别有用。下面的代码从一个流开始，并不断地应用迭代体。大于0的元素被送回反馈通道，其余的元素被转发到下游。

IterativeStream iteration = initialStream.iterate();
DataStream iterationBody = iteration.map ();
DataStream feedback = iterationBody.filter(new FilterFunction(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value > 0;
    }
});
iteration.closeWith(feedback);
DataStream output = iterationBody.filter(new FilterFunction(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value <= 0;
    }
});

Physical Partitioning

Flink还通过以下函数对转换后的流分区进行低级控制(如果需要的话)。

自定义分区

DataStream → DataStream
使用用户定义的Partitioner为每个元素选择目标任务。

dataStream.partitionCustom(partitioner, "someKey");
dataStream.partitionCustom(partitioner, 0);
dataStream.partitionCustom(partitioner, new KeySelector());

for example:
        DataStream partitionCustom = map1.partitionCustom(new MyPartition(), (KeySelector) Integer::longValue);
public static class MyPartition implements Partitioner {
        public MyPartition() {
        }

        @Override
        public int partition(Long key, int numPartitions) {
            if (key % numPartitions == 0) {
                return 0;
            } else {
                return 1;
            }
        }
    }

Random Partitioning

DataStream → DataStream
按均匀分布随机划分元素.

dataStream.shuffle();

Rescaling

DataStream → DataStream
循环地将元素分区到下游ioperator的子集。如果您希望使用管道，例如，从一个源的每个并行实例分散到几个映射器的子集，以分发负载，但又不希望balance()致完全的再平衡，那么这是很有用的。这将只需要本地数据传输，而不是通过网络传输数据，这取决于其他配置值，比如taskmanager的槽位数。

上游操作向其发送元素的下游操作子集取决于上游和下游操作的并行度。例如，如果上游操作的并行度为2，而下游操作的并行度为6，那么一个上游操作将元素分配给三个下游操作，而另一个上游操作将分配给其他三个下游操作。

另一方面，如果下游操作的并行度为2，而上游操作的并行度为6，那么三个上游操作将分配给一个下游操作，而其他三个上游操作将分配给另一个下游操作。

在不同的并行度不是彼此的倍数的情况下，一个或多个下游操作将有不同数量的来自上游操作的输入。

请看图，上面例子中的连接模式的可视化:

dataStream.rescale();

Broadcasting

DataStream → DataStream
向每个分区广播元素。

dataStream.broadcast();

任务链接和资源组

链接两个后续转换意味着将它们放在同一个线程中以获得更好的性能。如果可能的话，默认使用Flink 链算子(例如，两个后续映射转换)。如果需要，API提供了对链接的细粒度控制:

如果你想在整个作业中禁用链，请使用StreamExecutionEnvironment.disableOperatorChaining()。对于更细粒度的控制，可以使用以下功能。注意，这些函数只能在DataStream转换之后使用，因为它们引用了前面的转换。例如，你可以使用someStream.map(…). startnewchain()，但你不能使用someStream.startNewChain()。

资源组是Flink中的一个槽位，参见slots。如果需要，可以手动隔离操作符在不同的槽位。

Start New Chain

开始一个新的链，从这个算子开始。这两个映射器将被链接起来，而过滤器将不会被链接到第一个映射器。

someStream.filter(...).map(...).startNewChain().map(...);

Disable Chaining

不要链映射算子。

someStream.map(...).disableChaining();

设置槽位共享组(Set Slot Sharing Group)

设置操作对应的槽位共享组。Flink将把具有相同槽位共享组的操作放到同一个槽位，而在其他槽位中保留不具有槽位共享组的操作。这可以用来隔离插槽。如果所有输入操作都在同一个槽位共享组内，则输入操作继承该槽位共享组。默认槽位共享组的名称为“default”，可以通过调用slotSharingGroup(“default”)显式地将操作放入该组。

someStream.filter(...).slotSharingGroup("name");

Name And Description（算子和作业顶点描述）

flink中的算子和作业顶点有一个名称和描述。名称和描述都是关于操作符或工作顶点正在做什么的介绍，但是它们的用法不同。

算子和作业顶点的名称将用于web ui、线程名称、日志记录、指标等。作业顶点的名称是根据其中操作符的名称构造的。名称需要尽可能简洁，以避免对外部系统造成巨大压力。

该描述将在执行计划中使用，并在web UI中显示为作业顶点的详细信息。根据作业顶点中算子的描述，构造作业顶点的描述。描述中可以包含操作符的详细信息，便于运行时调试。

someStream.filter(...).setName("filter").setDescription("x in (1, 2, 3, 4) and y > 1");

作业顶点的描述格式默认为树格式字符串。用户可以设置管道。如果他们想将description设置为与以前版本一样的级联格式，则将vertex-description-mode设置为CASCADING。

默认情况下，Flink SQL生成的算子会有一个由算子类型和id组成的名称，以及详细的描述。用户可以设置table.optimizer。如果他们希望像以前的版本一样将name设置为详细描述，则将simple -operator-name-enabled设置为false。

当管道的拓扑结构比较复杂时，用户可以通过集合管道在顶点名称中添加拓扑索引。将顶点-name-include-index-prefix设置为true，这样我们就可以根据日志或度量标签轻松地找到图中的顶点。