Apache Flink Streaming(DataStream API)

综述：

• 在Flink中DataStream程序是在数据流上实现了转换的常规程序。

1.示范程序

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class WindowWordCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

}

nc -lk 9999

2.数据源

• 程序从源读取输入。可以通过StreamExecutionEnvironment.addSource(sourceFunction)给程序附上源。
• 在StreamExecutionEnvironment中有一些可访问的预定义的流数据源：                                                                              readTextFile(path)   逐行作为字符串读取文本文件                                                                                                              readFile(fileInputFormat, path)    通过指定的文件输入格式(the specified file input format)读取文件                                    readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)     这是一个被前两个方法内部调用的方法。它基于给定fileInputFormat在path下读取文件，根据提供的watchType，这个源会定期监测(每 interval ms)新数据的路径。
• 基于套接字的  socketTextStream   从套接字读取。元素可以由一个分隔符分开。
• 基于集合的                                                                                                                                                                              fromCollection(Collection)     从Java Java.util.Collection创建一个数据流，集合中的所有元素必须是相同类型的。            fromCollection(Iterator, Class)     从一个迭代器创建一个数据流，类指定迭代器返回的元素的数据类型。                          fromElements(T ...)      从给定的对象的序列创建一个数据流，所有对象必须是相同类型的。                                            fromParallelCollection(SplittableIterator, Class)   在并行执行中，从一个迭代器创建一个数据流，类指定迭代器返回的元素的数据类型。                                                                                                                                                                      generateSequence(from, to)       在给定的时间间隔内，生成的数字序列，并行执行。
• 自定义的          addSource    附上一个新的源函数。例如要从Apache Kafka读取，可以用addSource(new FlinkKafkaConsumer08<>(...))。

3.DataStream Transformations     参照运算符。

4.Data Sinks 数据接收

• .Data Sinks中消耗数据流，把他们发给文件、套接字、外部系统或打印。Flink带有多种封装在操作符中的内置输出格式：   writeAsText() / TextOutputFormat    把元素当做字符串(通过toString()获得)顺序写入                                                     writeAsCsv(...) / CsvOutputFormat    将元组写为逗号分隔值文件。                                                                               print() / printToErr()    在stdout/stderr上输出每个元素的toString()值。                                                                       writeUsingOutputFormat() / FileOutputForma      方法和基类定义文件输出。                                                             writeToSocket   根据SerializationSchema把元素输出到套接字。                                                                                   addSink    调用自定义sink函数。

5.Iterations  迭代次数

• 迭代流程序实现一个阶跃函数并且被嵌入到一个IterativeStream。需要指定的哪一部分流反馈迭代和哪一部分流用split转换或filter转发下游。                                                                                                                                                                  首先,我们定义一个IterativeStream：

  IterativeStream<Integer> iteration = input.iterate();

  然后,我们用一系列的转换指定将在循环内执行的逻辑：

  DataStream<Integer> iterationBody = iteration.map(/* this is executed many times */);

  关闭一个迭代和定义迭代尾巴,调用IterativeStream中closeWith(feedbackStream) 方法。一种常见模式是使用一个filter(过滤器)来分离流的一部分(用于反馈)和流的一部分(向前传播)：

  iteration.closeWith(iterationBody.filter(/* one part of the stream */));
  DataStream<Integer> output = iterationBody.filter(/* some other part of the stream */);

  例如，这是程序是一系列整数不断减去1，直到他们达到零：

  DataStream<Long> someIntegers = env.generateSequence(0, 1000);
  
  IterativeStream<Long> iteration = someIntegers.iterate();
  
  DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
    @Override
    public Long map(Long value) throws Exception {
      return value - 1 ;
    }
  });
  
  DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
    @Override
    public boolean filter(Long value) throws Exception {
      return (value > 0);
    }
  });
  
  iteration.closeWith(stillGreaterThanZero);
  
  DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
    @Override
    public boolean filter(Long value) throws Exception {
      return (value <= 0);
    }
  });

6.执行参数

• StreamExecutionEnvironment包含ExecutionConfig (允许在运行时给job设定具体的配置值)
• 延迟控制  默认情况下,元素并不一个接一个的传输到网络上(这将导致不必要的网络流量)而是缓冲。尽管缓冲可以优化吞吐量，但是当传入的流不够快他可能导致延迟问题。为了控制吞吐量和延迟,您可以使用env.setBufferTimeout(timeoutMillis)去设置一个最大等待时间使缓冲区填满。

  LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  env.setBufferTimeout(timeoutMillis);
  
  env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);

  为了使吞吐量最大化，设置setBufferTimeout(-1)会删除超时并且只会当缓冲都满了才刷新。为了使最小化延迟，设置Timeout为 一个值接近于0(例如5或10 ms)的值。应该避免缓冲区Timeout为0，因为它可能导致性能严重的下降。

 7.调试

• Flink提供通过从IDE中支持本地调试显著地减轻数据分析程序的开发，注入测试数据和结果数据的集合。
• 本地执行环境：

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  
  DataStream<String> lines = env.addSource(/* some source */);
  // build your program
  
  env.execute();

  LocalEnvironment创建和使用。

• 收集数据来源：

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  
  // Create a DataStream from a list of elements
  DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);
  
  // Create a DataStream from any Java collection
  List<Tuple2<String, Integer>> data = ...
  DataStream<Tuple2<String, Integer>> myTuples = env.fromCollection(data);
  
  // Create a DataStream from an Iterator
  Iterator<Long> longIt = ...
  DataStream<Long> myLongs = env.fromCollection(longIt, Long.class);

  目前,收集数据源要求数据类型和迭代器实现Serializable(可串行化)。此外，收集数据源不能并行执行(parallelism(平行)= 1)。

• Iterator Data Sink：迭代数据接收器。Flink还提供了一个sink收集DataStream数据结果进行测试和调试：

  import org.apache.flink.streaming.experimental.DataStreamUtils
  
  DataStream<Tuple2<String, Integer>> myResult = ...
  Iterator<Tuple2<String, Integer>> myOutput = DataStreamUtils.collect(myResult)

Event Time：

1.Time

• Processing time ：处理时间。处理时间是指机器执行相应的操作的系统时间。当流程序运行在处理时间时，所有基于时间的操作(如时间窗口)将使用机器的系统时钟运行各自的操作符。
•  Event time：事件时间 。事件时间是在其生产设备每个事件发生的时间。这个时间通常是嵌入在他们进入Flink之前的记录中，事件的时间戳可以从每个记录提取出。时间的进展取决于数据。
• Ingestion time：摄取时间。摄取时间是事件进入Flink的时间。源操作符的每个记录源的当前时间作为一个时间戳，基于时间的操作(如时间窗)指这个时间戳。

• Flink DataStream程序的第一部分通常设置基本时间特性。该设置定义了数据流源的行为方式(例如，它们是否将分配时间戳)，以及窗口操作应该使用的概念像KeyedStream.timeWindow(Time.seconds(30))。以下示例显示了一个Flink程序，它在每小时时间窗口中合计事件。窗户的行为适应时间特征。

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
  
  // alternatively:
  // env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
  // env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
  DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
  
  stream
      .keyBy( (event) -> event.getUser() )
      .timeWindow(Time.hours(1))
      .reduce( (a, b) -> a.add(b) )
      .addSink(...);

2.事件时间和水印

• 支持事件时间的流处理器需要一种方法来衡量事件时间的进度。例如，当事件时间超过一小时时，需要通知构建每小时窗口的窗口操作员，以便操作员可以关闭正在进行的窗口。事件时间可以独立于处理时间(由挂钟测量)进行。
• Flink中用于衡量事件时间进度的机制是水印。水印作为数据流的一部分流动并带有时间戳(t)。
• 下图显示了具有(逻辑的)时间戳和(内联的)水印的事件流。在该示例中，事件按顺序(关于它们的时间戳)，意味着水印仅是流中的周期性标记。
• 水印对于乱序流是至关重要的，如下图所示，这里的活动不是由他们的时间戳排序。通常，水印是一种声明，在该流中到达某个时间戳的所有事件都应该到达。一旦水印到达操作员，操作员就可以将其内部事件时钟提前到水印的值。

3.并行流中的水印

• 水印或直接在源函数生成。每个源函数的平行子任务通常独立生成的水印。这些水印定义特定并行来源的事件时间。
• 某些元素可能违反水印条件，这意味着即使在水印(t)发生之后，也会出现更多具有时间戳t'<= t的元素。某些元素可以任意延迟，从而无法指定某个事件时间戳的所有元素将发生的时间。

4.生成时间戳/水印

• 使用事件时间,流程序需要设置相应的时间特性。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

• 了处理事件时间，Flink需要知道事件的时间戳，这意味着流中的每个元素都需要分配其事件时间戳。时间戳分配与生成水印密切相关，水印告诉系统事件时间的进展。时间戳和水印都指定为毫秒。有两种方法可以分配时间戳并生成水印：
  1. 直接在数据流源中。
  2. 通过时间戳分配器/水印生成器：在Flink中，时间戳分配器还定义要发出的水印。
• 数据流源可以直接为它们生成的元素分配时间戳，也可以发出水印。完成此操作后，不需要时间戳分配器。请注意，如果使用时间戳分配器，则源将提供的任何时间戳和水印都将被覆盖。要直接在源中为每一个元素分配一个时间戳，必须使用SourceContext上的collectWithTimestamp(...)方法，为了生成水印,源必须调用emitWatermark(Watermark)方法。下面是一个(非检查点)源的简单示例，它分配时间戳并生成水印：

  @Override
  public void run(SourceContext<MyType> ctx) throws Exception {
      while (/* condition */) {
          MyType next = getNext();
          ctx.collectWithTimestamp(next, next.getEventTimestamp());
  
          if (next.hasWatermarkTime()) {
              ctx.emitWatermark(new Watermark(next.getWatermarkTime()));
          }
      }
  }

• 时间戳分配器获取流并生成带时间戳和水印的新流。如果原始流已经有时间戳和/或水印，则时间戳分配器会覆盖它们。

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
  DataStream<MyEvent> stream = env.readFile(
          myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
          FilePathFilter.createDefaultFilter(), typeInfo);
  
  DataStream<MyEvent> withTimestampsAndWatermarks = stream
          .filter( event -> event.severity() == WARNING )
          .assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks());
  
  withTimestampsAndWatermarks
          .keyBy( (event) -> event.getGroup() )
          .timeWindow(Time.seconds(10))
          .reduce( (a, b) -> a.add(b) )
          .addSink(...);

  AssignerWithPeriodicWatermarks 分配时间戳并定期生成水印

State & Fault Tolerance:

1.Keyed State和Operator State

• Flink有两种基本的状态：Keyed State和Operator State。
• 键控状态总是相对于键，只能用于KeyedStream上的函数和运算符。
• 与操作符状态(或非键状态),每个操作符绑定到一个并行算子实例状态。

2.The Broadcast State Pattern

• 使用State描述运算符状态，其在运算符的并行任务中恢复时均匀分布，或者联合，整个状态用于初始化已恢复的并行任务。

3.检查点

• Flink中的每个函数和运算符都可以是有状态的。有状态函数通过各个元素/事件的处理存储数据，使状态成为任何类型的更精细操作的关键构建模块。
• 为了能使状态容错，Flink需要checkpoint状态。检查点允许Flink恢复流中的状态和位置，从而为应用程序提供与无故障执行相同的语义。

操作符：

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

  转换	描述
  Map DataStream → DataStream	采用一个元素并生成一个元素。一个map函数，它将输入流的值加倍： DataStream<Integer> dataStream = //... dataStream.map(new MapFunction<Integer, Integer>() { @Override public Integer map(Integer value) throws Exception { return 2 * value; } });
  FlatMap DataStream→DataStream	采用一个元素并生成零个，一个或多个元素。将句子分割为单词的flatmap函数： dataStream.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { for(String word: value.split(" ")){ out.collect(word); } } });
  Filter DataStream → DataStream	计算每个元素的布尔函数，并保留函数返回true的元素。过滤掉零值的过滤器： dataStream.filter(new FilterFunction<Integer>() { @Override public boolean filter(Integer value) throws Exception { return value != 0; } });
  KeyBy DataStream → KeyedStream	逻辑上将流分区为不相交的分区。具有相同密钥key的记录都分配给同一分区。在内部KeyBy()是通过哈希分区实现的。有不同的方式去指定keys。 此转换返回KeyedStream，其中包括使用键控状态所需的KeyedStream。 dataStream.keyBy("someKey") // Key by field "someKey" dataStream.keyBy(0) // Key by the first element of a Tuple 下列类型不能成为key： 1.是一个POJO类型但是没有重写hashCode()方法并且依赖于Object.hashCode()实现。 2.是任何类型的数组。
  Reduce KeyedStream → DataStream	键控数据流上的“滚动”减少。将当前元素与最后一个减少的值组合并发出新值。   reduce函数，用于创建部分和的流： keyedStream.reduce(new ReduceFunction<Integer>() { @Override public Integer reduce(Integer value1, Integer value2) throws Exception { return value1 + value2; } });
  Flod KeyedStream → DataStream	具有初始值的键控数据流上的“滚动”折叠。将当前元素与最后折叠的值组合并发出新值。  折叠函数，当应用于序列（1,2,3,4,5）时，发出序列“start-1”，“start-1-2”，“start-1-2-3”,. .. DataStream<String> result = keyedStream.fold("start", new FoldFunction<Integer, String>() { @Override public String fold(String current, Integer value) { return current + "-" + value; } });
  Aggregations KeyedStream → DataStream	在键控数据流上滚动聚合。min和minBy之间的差异是min返回最小值，而minBy返回该字段中具有最小值的元素（max和maxBy相同）。 keyedStream.sum(0); keyedStream.sum("key"); keyedStream.min(0); keyedStream.min("key"); keyedStream.max(0); keyedStream.max("key"); keyedStream.minBy(0); keyedStream.minBy("key"); keyedStream.maxBy(0); keyedStream.maxBy("key");
  Window KeyedStream → WindowedStream	可以在已经分区的KeyedStream上定义Windows。Windows根据某些特征（例如，在最后5秒内到达的数据）对每个key中的数据进行分组。 dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
  WindowALL DataStream → AllWindowedStream	Windows可以在常规DataStream上定义。Windows根据某些特征（例如，在最后5秒内到达的数据）对所有流事件进行分组。 警告：在许多情况下，这是非并行转换。所有记录将收集在windowAll运算符的一个任务中。 dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
  Window Apply WindowedStream→DataStream  AllWindowedStream→DataStream	将一般功能应用于整个窗口。下面是一个手动求和窗口元素的函数。 注意：如果您正在使用windowAll转换，则需要使用AllWindowFunction。 windowedStream.apply (new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() { public void apply (Tuple tuple, Window window, Iterable<Tuple2<String, Integer>> values, Collector<Integer> 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<Tuple2<String,Integer>, Integer, Window>() { public void apply (Window window, Iterable<Tuple2<String, Integer>> values, Collector<Integer> out) throws Exception { int sum = 0; for (value t: values) { sum += t.f1; } out.collect (new Integer(sum)); } });
  Window Reduce WindowedStream→DataStream	将reduce函数应用于window并返回减少后的值。 windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>>() { public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception { return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1); } });
  Window Flod WindowedStream → DataStream	将功能折叠功能应用于window并返回折叠值。示例函数应用于序列（1,2,3,4,5）时，将序列折叠为字符串“start-1-2-3-4-5”： windowedStream.fold("start", new FoldFunction<Integer, String>() { public String fold(String current, Integer value) { return current + "-" + value; } });
  Aggregations on windows WindowedStream → DataStream	聚合窗口的内容。min和minBy之间的差异是min返回最小值，而minBy返回该字段中具有最小值的元素（max和maxBy相同）。 windowedStream.sum(0); windowedStream.sum("key"); windowedStream.min(0); windowedStream.min("key"); windowedStream.max(0); windowedStream.max("key"); windowedStream.minBy(0); windowedStream.minBy("key"); windowedStream.maxBy(0); windowedStream.maxBy("key");
  Union DataStream *→DataStream	两个或多个数据流的联合，创建包含来自所有流的所有元素的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次元素。 dataStream.union(otherStream1, otherStream2, ...);
  Window Join DataStream，DataStream→DataStream	在给定密钥和公共窗口上连接两个数据流。 dataStream.join(otherStream) .where(<key selector>).equalTo(<key selector>) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new JoinFunction () {...});
  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	在给定密钥和公共窗口上对两个数据流进行Cogroup。 dataStream.coGroup(otherStream) .where(0).equalTo(1) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new CoGroupFunction () {...});
  Connect DataStream,DataStream → ConnectedStreams	“连接”两个保留其类型的数据流。连接允许两个流之间共享状态。 DataStream<Integer> someStream = //... DataStream<String> otherStream = //... ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);
  CoMap，CoFlatMap ConnectedStreams → DataStream	类似于连接数据流上的map和flatMap。 connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() { @Override public Boolean map1(Integer value) { return true; } @Override public Boolean map2(String value) { return false; } }); connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() { @Override public void flatMap1(Integer value, Collector<String> out) { out.collect(value.toString()); } @Override public void flatMap2(String value, Collector<String> out) { for (String word: value.split(" ")) { out.collect(word); } } });
  Split DataStream → SplitStream	根据一些标准把一个流拆分成多个流。 SplitStream<Integer> split = someDataStream.split(new OutputSelector<Integer>() { @Override public Iterable<String> select(Integer value) { List<String> output = new ArrayList<String>(); if (value % 2 == 0) { output.add("even"); } else { output.add("odd"); } return output; } });
  Select SplitStream → DataStream	从拆分流中选择一个或多个流。 SplitStream<Integer> split; DataStream<Integer> even = split.select("even"); DataStream<Integer> odd = split.select("odd"); DataStream<Integer> all = split.select("even","odd");
  Iterate DataStream → IterativeStream → DataStream	通过将一个运算符的输出重定向到某个先前的运算符，在流中创建“反馈”循环。这对于定义不断更新模型的算法特别有用。以下代码以流开头并连续应用迭代体。大于0的元素将被发送回反馈通道，其余元素将向下游转发。 IterativeStream<Long> iteration = initialStream.iterate(); DataStream<Long> iterationBody = iteration.map (/*do something*/); DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){ @Override public boolean filter(Integer value) throws Exception { return value > 0; } }); iteration.closeWith(feedback); DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){ @Override public boolean filter(Integer value) throws Exception { return value <= 0; } });
  Extract Timestamps DataStream → DataStream	从记录中提取时间戳，以便使用使用事件时间语义的窗口。 stream.assignTimestamps (new TimeStampExtractor() {...});

• 以下转换可用于元组的数据流：

  转换	描述
  Project DataStream→DataStream	从元组中选择字段的子集。 DataStream<Tuple3<Integer, Double, String>> in = // [...] DataStream<Tuple2<String, Integer>> out = in.project(2,0);

• 链接两个后续转换意味着将它们共同定位在同一个线程中以获得更好的性能。如果可能的话，Flink默认链操作符（例如，两个后续的map转换）。

 5.视窗windows

• Windows是处理无限流的核心。Windows将流拆分为有限大小的“buckets(桶)”，我们可以在其上应用计算。
• 窗口Flink程序的一般结构如下所示。第一个片段指的是键控流，而第二个片段指的是非键控流。

  键控Windows:
  
  stream
         .keyBy(...)               <-  keyed versus non-keyed windows
         .window(...)              <-  required: "assigner"
        [.trigger(...)]            <-  optional: "trigger" (else default trigger)
        [.evictor(...)]            <-  optional: "evictor" (else no evictor)
        [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
        [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
         .reduce/aggregate/fold/apply()      <-  required: "function"
        [.getSideOutput(...)]      <-  optional: "output tag"
  
  非键控Windows:
  
  stream
         .windowAll(...)           <-  required: "assigner"
        [.trigger(...)]            <-  optional: "trigger" (else default trigger)
        [.evictor(...)]            <-  optional: "evictor" (else no evictor)
        [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
        [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
         .reduce/aggregate/fold/apply()      <-  required: "function"
        [.getSideOutput(...)]      <-  optional: "output tag"
  
  在上面，方括号（[...]）中的命令是可选的。这表明Flink允许您以多种不同方式自定义窗口逻辑，以便最适合您的需求。

• 窗口生命周期：只要属于此窗口的第一个元素到达，就会创建一个窗口，当时间(事件或处理时间)超过其结束时间戳加上用户指定允许迟到时，窗口将被完全删除。Flink保证仅删除基于时间的窗口而不是其他类型。
• 翻滚窗口( tumbling windows)：翻滚窗口分配器分配每个元素给指定窗口大小的窗口。例如，如果指定大小为5分钟的翻滚窗口，则将评估当前窗口，并且每五分钟将启动一个新窗口，如下图所示。 

  DataStream<T> input = ...;
  
  // tumbling event-time windows
  input
      .keyBy(<key selector>)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .<windowed transformation>(<window function>);
  
  // tumbling processing-time windows
  input
      .keyBy(<key selector>)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
      .<windowed transformation>(<window function>);
  
  // daily tumbling event-time windows offset by -8 hours.
  input
      .keyBy(<key selector>)
      .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
      .<windowed transformation>(<window function>);

• 滑动窗口(Sliding Windows)：滑动窗口分配器分配元素以固定长度的窗口。与翻滚窗口分配器类似，窗口大小由窗口大小参数配置。附加的窗口滑动参数控制滑动窗口的启动频率。                                                                                                                   

  DataStream<T> input = ...;
  
  // sliding event-time windows
  input
      .keyBy(<key selector>)
      .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
      .<windowed transformation>(<window function>);
  
  // sliding processing-time windows
  input
      .keyBy(<key selector>)
      .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
      .<windowed transformation>(<window function>);
  
  // sliding processing-time windows offset by -8 hours
  input
      .keyBy(<key selector>)
      .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
      .<windowed transformation>(<window function>);

• 会话窗口(Session Windows)：在会话窗口中按活动会话分配组中的元素。与翻滚窗口和滑动窗口相比，会话窗口不重叠并且没有固定的开始和结束时间。相反，当会话窗口在一段时间内没有接收到元素时，即当发生不活动的间隙时，会关闭会话窗口。   

  DataStream<T> input = ...;
  
  // event-time session windows with static gap
  input
      .keyBy(<key selector>)
      .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
      .<windowed transformation>(<window function>);
      
  // event-time session windows with dynamic gap
  input
      .keyBy(<key selector>)
      .window(EventTimeSessionWindows.withDynamicGap((element) -> {
          // determine and return session gap
      }))
      .<windowed transformation>(<window function>);
  
  // processing-time session windows with static gap
  input
      .keyBy(<key selector>)
      .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
      .<windowed transformation>(<window function>);
      
  // processing-time session windows with dynamic gap
  input
      .keyBy(<key selector>)
      .window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {
          // determine and return session gap
      }))
      .<windowed transformation>(<window function>);

• window join：window join连接两个共享公共密钥并位于同一窗口中的流的元素。一般用法可概括如下：

  stream.join(otherStream)
      .where(<KeySelector>)
      .equalTo(<KeySelector>)
      .window(<WindowAssigner>)
      .apply(<JoinFunction>)
  

  关于语义的一些注释：

• 两个流的元素的成对组合的创建表现得像内部连接，意味着如果它们没有来自要连接的另一个流的对应元素，则不会发出来自一个流的元素。
• 那些加入的元素将在其时间戳中包含仍位于相应窗口中的最大时间戳。例如，[5, 10)具有其边界的窗口将导致连接的元素具有9作为其时间戳。
• 异步IO访问外部数据：与数据库的异步交互意味着单个并行函数实例可以同时处理许多请求并同时接收响应。这样，可以通过发送其他请求和接收响应来覆盖等待时间。至少，等待时间在多个请求上摊销。这导致大多数情况下流量吞吐量更高。

Streaming Connectors：

• Flink内置了一些基本数据源和接收器，并且始终可用。