Flink1.13架构全集| 一文带你由浅入深精通Flink方方面面

<properties>
<flink.version>1.13.0</flink.version>
<java.version>1.8</java.version>
<scala.binary.version>2.12</scala.binary.version>
<slf4j.version>1.7.30</slf4j.version>
</properties>

<dependencies>

<!-- 引入Flink相关依赖-->

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-java</artifactId>
  <version>${flink.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  <version>${flink.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-clients_${scala.binary.version}</artifactId>
  <version>${flink.version}</version>
</dependency>

<!-- 引入日志管理相关依赖-->

<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-api</artifactId>
  <version>${slf4j.version}</version>
</dependency>
<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-log4j12</artifactId>
  <version>${slf4j.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-to-slf4j</artifactId>
  <version>2.14.0</version>
</dependency>
</dependencies>

log4j.rootLogger=error, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

hello flink

hello world
hello spark

hello hadoop

hello hive

package com.liuhao.wordcountP;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class wordcountP {
  public static void main(String[] args) throws Exception {
//        1.获取执行环境
      ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//        2.读取文件数据
      DataSource<String> DataSource = env.readTextFile("input/wordcount.txt");
//        3.处理转换，变成（hello，1）二元组
      FlatMapOperator<String, Tuple2<String, Long>> stringTuple2FlatMapOperator = DataSource
              .flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
                  String[] word = line.split(" ");
                  for (String s : word) {
                      out.collect(Tuple2.of(s, 1l));
                  }
              }).returns(Types.TUPLE(Types.STRING, Types.LONG));
//        4.按照key进行分组
      UnsortedGrouping<Tuple2<String, Long>> groupBy = stringTuple2FlatMapOperator.groupBy(0);
//        5.按照第二个字段进行sum
      AggregateOperator<Tuple2<String, Long>> sum = groupBy.sum(1);
//        6.输出
      sum.print();
  }
}

$ bin/flink run -Dexecution.runtime-mode=BATCH WordCount.jar

package com.liuhao.WordCountL;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
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;

import java.util.Arrays;

public class WordCountL {
public static void main(String[] args) throws Exception {
  // 1. 创建流式执行环境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  // 2. 读取文件
  DataStreamSource<String> lineStream =

env.readTextFile("input/words.txt");
  // 3. 转换、分组、求和，得到统计结果

SingleOutputStreamOperator<Tuple2<String, Long>> sum = 

lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {

String[] words = line.split(" ");

for (String word : words) {

out.collect(Tuple2.of(word, 1L));

}

}).returns(Types.TUPLE(Types.STRING, Types.LONG))

.keyBy(data -> data.f0)

.sum(1);



  // 4. 打印
  sum.print();
  // 5. 执行
  env.execute();
}
}

package com.liuhao.SocketWordCountL ;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
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;

import java.util.Arrays;

public class SocketWordCountL {
  public static void main(String[] args) throws Exception {
    // 1. 创建流式执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2. 读取文本流
    DataStreamSource<String> lineStream = env.socketTextStream("hadoop102", 7777);
    // 3. 转换、分组、求和，得到统计结果

SingleOutputStreamOperator<Tuple2<String, Long>> sum = 

lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {

String[] words = line.split(" ");

for (String word : words) {

out.collect(Tuple2.of(word, 1L));

}

}).returns(Types.TUPLE(Types.STRING, Types.LONG))

.keyBy(data -> data.f0)

.sum(1);


    // 4. 打印
    result.print();
    // 5. 执行
    env.execute();
  }
}

[hadoop@hadoop ~]$ nc -lk 7777

$ tar -zxvf flink-1.13.0-bin-scala_2.12.tgz -C /opt/module/

flink-1.13.0/

flink-1.13.0/log/

flink-1.13.0/LICENSE

flink-1.13.0/lib/

……

$ cd flink-1.13.0/

$ bin/start-cluster.sh 

Starting cluster.

Starting standalonesession daemon on host hadoop100.

Starting taskexecutor daemon on host hadoop100.

$ jps

10589 StandaloneSessionClusterEntrypoint

10697 TaskManagerRunner

10898 Jps

$ bin/stop-cluster.sh 

Stopping taskexecutor daemon (pid: 10680) on host hadoop100.

Stopping standalonesession daemon (pid: 10369) on host hadoop100.

$ cd conf/

$ vim flink-conf.yaml
jobmanager.rpc.address: hadoop100

$ vim workers 
hadoop101
hadoop102

jobmanager.memory.process.size：对JobManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1600M，可以根据集群规模进行适当调整。

taskmanager.memory.process.size：对TaskManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1600M，可以根据集群规模进行适当调整。

taskmanager.numberOfTaskSlots：对每个TaskManager能够分配的Slot数量进行配置，默认为1，可根据TaskManager所在的机器能够提供给Flink的CPU数量决定。所谓Slot就是TaskManager中具体运行一个任务所分配的计算资源。

parallelism.default：Flink任务执行的默认并行度，优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。

关于Slot和并行度的概念，我们会在下一章做详细讲解。

$ scp -r ./flink-1.13.0 atguigu@hadoop103:/opt/module
$ scp -r ./flink-1.13.0 atguigu@hadoop104:/opt/module

$ bin/start-cluster.sh 

Starting cluster.

Starting standalonesession daemon on host hadoop100.

Starting taskexecutor daemon on host hadoop101.

Starting taskexecutor daemon on host hadoop102.

[hadoop@hadoop100 flink-1.13.0]$ jps

13859 Jps

13782 StandaloneSessionClusterEntrypoint

[hadoop@hadoop101 flink-1.13.0]$ jps

12215 Jps

12124 TaskManagerRunner

[hadoop@hadoop102 flink-1.13.0]$ jps

11602 TaskManagerRunner

11694 Jps

<build>

   <plugins>

      <plugin>

        <groupId>org.apache.maven.plugins</groupId>

        <artifactId>maven-assembly-plugin</artifactId>

        <version>3.0.0</version>

        <configuration>

          <descriptorRefs>

            <descriptorRef>jar-with-dependencies</descriptorRef>

          </descriptorRefs>

        </configuration>

        <executions>

          <execution>

            <id>make-assembly</id>

            <phase>package</phase>

            <goals>

              <goal>single</goal>

            </goals>

          </execution>

        </executions>

      </plugin>

    </plugins>

</build>

[INFO] -----------------------------------------------------------------------

[INFO] BUILD SUCCESS

[INFO] -----------------------------------------------------------------------

[INFO] Total time: 21.665 s

[INFO] Finished at: 2021-06-01T17:21:26+08:00

[INFO] Final Memory: 141M/770M

[INFO] -----------------------------------------------------------------------

$ bin/start-cluster.sh

$ nc -lk 7777

$ bin/flink run -m hadoop100:8081 -c com.liuhao.WordCountL ./FlinkTutorial-1.0-SNAPSHOT.jar --host hadoop102 --port 7777

$ cat flink-atguigu-taskexecutor-0-hadoop102.out

SLF4J: Class path contains multiple SLF4J bindings.

SLF4J: Found binding in [jar:file:/opt/module/flink-1.13.0/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: Found binding in [jar:file:/opt/module/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.

SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]

(hello,1)

(hello,2)

(flink,1)

(hello,3)

(scala,1)

$ cp ./FlinkTutorial-1.0-SNAPSHOT.jar lib/

$ ./bin/standalone-job.sh start --job-classname com.liuhao.WordCountL

$ ./bin/taskmanager.sh start

$ ./bin/standalone-job.sh stop

$ ./bin/taskmanager.sh stop

$ sudo vim /etc/profile.d/my_env.sh

HADOOP_HOME=/opt/module/hadoop-2.7.5

export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop

export HADOOP_CLASSPATH=`hadoop classpath`

[hadoop@hadoop100 ~]$ start-dfs.sh

[hadoop@hadoop101 ~]$ start-yarn.sh

$ bin/yarn-session.sh -nm test

-d：分离模式，如果你不想让Flink YARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。

-jm(--jobManagerMemory)：配置JobManager所需内存，默认单位MB。

-nm(--name)：配置在YARN UI界面上显示的任务名。

-qu(--queue)：指定YARN队列名。

-tm(--taskManager)：配置每个TaskManager所使用内存。

2022-11-25 15:54:27,069 INFO  org.apache.flink.yarn.YarnClusterDescriptor          [] - YARN application has been deployed successfully.

2022-11-25 15:54:27,070 INFO  org.apache.flink.yarn.YarnClusterDescriptor          [] - Found Web Interface hadoop104:39735 of application 'application_1622535605178_0003'.

JobManager Web Interface: http://hadoop104:39735

$ bin/flink run -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

$ bin/flink run -d -t yarn-per-job -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

$ bin/flink run -m yarn-cluster -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

$ ./bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY

$ ./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>

$ bin/flink run-application -t yarn-application -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

$ ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

$ ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

$ ./bin/flink run-application -t yarn-application	-Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist-dir"	hdfs://myhdfs/jars/my-application.jar

1. 提交应用
2. 启动并提交应用
3. 请求slots
4. 任务启动
5. 注册slots
6. 发出提供slot的指令
7. 提供slots
8. 提交要在slots中执行的任务
9. 交换数据

1. Flink任务提交后，Client向HDFS上传Flink的Jar包和配置
2. 随后向Yarn ResourceManager 提交任务，ResourceManager分配Container资源并通知对应的NodeManager启动
3. ApplicationMaster，ApplicationMaster 启动后加载Flink的Jar包和配置构建环境
4. 然后启动JobManager，之后ApplicationMaster向ResourceManager申请资源启动TaskManager
5. ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager
6. NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager
7. TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。

1.Flink中每一个Taskmanageri都是一个JMM进程，它可能会在独立的线程上执行一个或多个subtask
2.为了控制一个Taskmanageri能接收多少个task, Taskmanager通过task slot来进行控制（一个Taskmanager至少有一个slot）


假如一个TaskManager有三个slot，那么它会将管理的内存平均分成三份，每个slot独自占据一份。这样一来，我们在slot上执行一个子任务时，相当于划定了一块内存“专款专用”，就不需要跟来自其他作业的任务去竞争内存资源了。所以现在我们只要2个TaskManager，就可以并行处理分配好的5个任务了。

3.默认情况下，Fink允许子任务共享slot，即使它们是不同任务的子任务。这样的结果是，一个slot可以保存作业的整个管道。
4.Task Slot是静态的概念，是指Taskmanager具有的并发执行能力

1. 所有的Flink程序都是由三部分组成的：Source、 Transformation和Sink
2. Source负责读取数据源，Transformation利用各种算子进行处理加工，Sink负责输出
3. 在运行时，Flink上运行的程序会被映射成 “逻辑数据流”（ dataflows),它包含了这三部分
4. 每一个dataflow以一个或多个Sources开始以一个或多个sinks结束。dataflow类以于任意的有向无环图（DAG)
5. 在大部分情况下，程序中的转换运算（ transformations）跟 dataflow中的算子

1.Streamgraph：是根据用户通过Stream API编写的代码生成的最初的图。用来表示程序的拓扑结构。
2.Jobgraph: Streamgraph经过优化后生成了Jobgraph，提交给Jobmanager的数据结构。主要的优化为，将多个符合条件的节点chain在一起作为一个节点Execution Graph: Jobmanager根据Jobgraph生成
3.ExecutiongraphExecution Graph是 Job Graphi的并行化版本，是调度层最核心的数据结构。
4.物理执行图：Jobmanager根据Executiongraph对Job进行调度后，在各个Taskmanager上部署Task后形成的“ 图”，并不是一个具体的数据结构。

1. 特定算子的子任务（ subtask）的个数被称之为其并行度（ parallelism)般情况下，一个 stream的并行度，可以认为就是其所有算子中最大的并行度。
2. 一个程序中，不同的算子可能具有不同的并行度
3. 算子之间传输数据的形式可以是 one-to-one( (forwarding）的模式也可以是 redistributing的模式，具体是哪一种形式，取決于算子的种类
4. One-to-one: stream维护着分区以及元素的顺序（比如 Sources和map之间）。这意味着map算子的子任务看到的元素的个数以及顺序跟 Source算子的子任务生产的元素的个数、顺序相同。map、 fliter、flatmap等算子都是one-to-one的对应关系
5. Redistributing: stream的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如keyby基于 hash Code重分区、而broadcast和rebalance会随机重新分区，这些算子都会引起distributer过程，而redistribute过程就类似于Spark中的shuffle过程。

bin/flink run –p 2 –c com.atguigu.wc.StreamWordCount 
./FlinkTutorial-1.0-SNAPSHOT.jar

1. Flink采用了一种称为任务链的优化技术，可以在特定条件下减少本地通信的开销。为了满足任务链的要求 ，必须将两个或多个算子设为相同的并行度，并通过本地转发（ ocal forward）的方式进行连接
2. 相同并行度的one-to-one操作， Flink这样相连的算子链接在一起形成一个task，原来的算子成为里面的subtask并行度相同、并且是one-to-one操作，两个条件缺一不可

这种模式下，数据流维护着分区以及元素的顺序。比如图中的source和map算子，source算子读取数据之后，可以直接发送给map算子做处理，它们之间不需要重新分区，也不需要调整数据的顺序。这就意味着map 算子的子任务，看到的元素个数和顺序跟source 算子的子任务产生的完全一样，保证着“一对一”的关系。map、filter、flatMap等算子都是这种one-to-one的对应关系。

这种关系类似于Spark中的窄依赖。

在这种模式下，数据流的分区会发生改变。比图中的map和后面的keyBy/window算子之间（这里的keyBy是数据传输算子，后面的window、apply方法共同构成了window算子）,以及keyBy/window算子和Sink算子之间，都是这样的关系。

每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的下游目标任务。例如，keyBy()是分组操作，本质上基于键（key）的哈希值（hashCode）进行了重分区；而当并行度改变时，比如从并行度为2的window算子，要传递到并行度为1的Sink算子，这时的数据传输方式是再平衡（rebalance），会把数据均匀地向下游子任务分发出去。这些传输方式都会引起重分区（redistribute）的过程，这一过程类似于Spark中的shuffle。

.map(word -> Tuple2.of(word, 1L)).disableChaining();

.map(word -> Tuple2.of(word, 1L)).startNewChain()

env.socketTextStream().flatMap(…).keyBy(0).sum(1).print();

bin/flink run –p 2 –c com.liuhao.WordCountL

./FlinkTutorial-1.0-SNAPSHOT.jar

这是根据用户通过 DataStream API编写的代码生成的最初的DAG图，用来表示程序的拓扑结构。这一步一般在客户端完成。

逻辑流图中的节点，完全对应着代码中的四步算子操作：

源算子Source（socketTextStream()）→扁平映射算子Flat Map(flatMap()) →分组聚合算子Keyed Aggregation(keyBy/sum()) →输出算子Sink(print())

StreamGraph经过优化后生成的就是作业图（JobGraph），这是提交给 JobManager 的数据结构，确定了当前作业中所有任务的划分。主要的优化为: 将多个符合条件的节点链接在一起合并成一个任务节点，形成算子链，这样可以减少数据交换的消耗。JobGraph一般也是在客户端生成的，在作业提交时传递给JobMaster。

分组聚合算子（Keyed Aggregation）和输出算子Sink(print)并行度都为2，而且是一对一的关系，满足算子链的要求，所以会合并在一起，成为一个任务节点。

JobMaster收到JobGraph后，会根据它来生成执行图（ExecutionGraph）。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。与JobGraph最大的区别就是按照并行度对并行子任务进行了拆分，并明确了任务间数据传输的方式。

JobMaster生成执行图后， 会将它分发给TaskManager；各个TaskManager会根据执行图部署任务，最终的物理执行过程也会形成一张“图”，一般就叫作物理图（Physical Graph）。这只是具体执行层面的图，并不是一个具体的数据结构。

物理图主要就是在执行图的基础上，进一步确定数据存放的位置和收发的具体方式。有了物理图，TaskManager就可以对传递来的数据进行处理计算了。

所以我们可以看到，程序里定义了四个算子操作：源（Source）->转换（flatmap）->分组聚合（keyBy/sum）->输出（print）；合并算子链进行优化之后，就只有三个任务节点了；再考虑并行度后，一共有5个并行子任务，最终需要5个线程来执行。

获取执行环境（execution environment）
读取数据源（source）
定义基于数据的转换操作（transformations）
定义计算结果的输出位置（sink）
触发程序执行（execute）

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment

 		.createRemoteEnvironment(

  			"host", // JobManager主机名

  			1234, // JobManager进程端口号

  		"path/to/jarFile.jar" // 提交给JobManager的JAR包
  		);

// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();

// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

bin/flink run -Dexecution.runtime-mode=BATCH ...

在提交作业时，增加execution.runtime-mode参数，指定值为BATCH。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

env.execute();

DataStream<String> stream = env.addSource(...);

import java.sql.Timestamp;

public class Event {

  public String user;

  public String url;

  public Long timestamp;

  public Event() {

  }

  public Event(String user, String url, Long timestamp) {

    this.user = user;

    this.url = url;

    this.timestamp = timestamp;

  }

 

  @Override

  public String toString() {

    return "Event{" +

        "user='" + user + '\'' +

        ", url='" + url + '\'' +

        ", timestamp=" + new Timestamp(timestamp) +

         '}';

  }

}

类是公有（public）的

 有一个无参的构造方法

 所有属性都是公有（public）的

 所有属性的类型都是可以序列化的

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

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.setParallelism(1);


  ArrayList<Event> clicks = new ArrayList<>();

  clicks.add(new Event("Mary","./home",1000L));

  clicks.add(new Event("Bob","./cart",2000L));


  DataStream<Event> stream = env.fromCollection(clicks);

	stream.print();

  env.execute();

}

DataStreamSource<Event> stream2 = env.fromElements(

   new Event("Mary", "./home", 1000L),

   new Event("Bob", "./cart", 2000L)

);

DataStream<String> stream = env.readTextFile("clicks.csv");

 参数可以是目录，也可以是文件；还可以从hdfs目录下读取，使用路径hdfs://...；

 路径可以是相对路径，也可以是绝对路径；

 相对路径是从系统属性user.dir获取路径: idea下是project的根目录, standalone模式下是集群节点根目录；

DataStream<String> stream = env.socketTextStream("localhost", 7777);

<dependency>

 <groupId>org.apache.flink</groupId>

 <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>

 <version>${flink.version}</version>

</dependency>

public class SourceKafka {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    Properties properties = new Properties();

    properties.setProperty("bootstrap.servers", "hadoop102:9092");

    properties.setProperty("group.id", "consumer-group");

    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    properties.setProperty("auto.offset.reset", "latest");

    DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>(

        "clicks",

        new SimpleStringSchema(),

        properties

    ));

    stream.print("Kafka");

    env.execute();

  }

}

第一个参数topic，定义了从哪些主题中读取数据。可以是一个topic，也可以是topic列表，还可以是匹配所有想要读取的topic的正则表达式。当从多个topic中读取数据时，Kafka连接器将会处理所有topic的分区，将这些分区的数据放到一条流中去。

第二个参数是一个DeserializationSchema或者KeyedDeserializationSchema。Kafka消息被存储为原始的字节数据，所以需要反序列化成Java或者Scala对象。上面代码中使用的SimpleStringSchema，是一个内置的DeserializationSchema，它只是将字节数组简单地反序列化成字符串。DeserializationSchema和KeyedDeserializationSchema是公共接口，所以我们也可以自定义反序列化逻辑。

第三个参数是一个Properties对象，设置了Kafka客户端的一些属性。

 run()方法：通过运行时上下文对象SourcContext循环生成数据，并发送到流中；

cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

public class ClickSource implements SourceFunction<Event> {

   private Boolean running = true;

   @Override

   public void run(SourceContext<Event> sourceContext) throws Exception {

     Random random = new Random();

     String[] users = {"Mary", "Bob"};

     String[] urls = {"./home", "./cart", “./fav”, “./prod?id=1”, “./prod?id=2”};

     while (running) {

       sourceContext.collect(new Event(

           users[random.nextInt(users.length)],

           urls[random.nextInt(urls.length)],

           Calendar.getInstance().getTimeInMillis()

       ));

     }

   }

   @Override

   public void cancel() {

     running = false;

   }

 }

env.addSource(new ClickSource())

public class SourceCustom {

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

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.setParallelism(1);

  	//有了自定义的source function，调用addSource方法

  DataStreamSource<Event> stream = env.addSource(new ClickSource());

  stream.print("SourceCustom");

  env.execute();

}

}

Java元组类型（TUPLE）：这是Flink内置的元组类型，是Java API的一部分。最多25个字段，也就是从Tuple0~Tuple25，不支持空字段

Scala 样例类及Scala元组：不支持空字段

行类型（ROW）：可以认为是具有任意个字段的元组,并支持空字段

POJO：Flink自定义的类似于Java bean模式的类

Flink支持所有的Java类和Scala类。不过如果没有按照上面POJO类型的要求来定义，就会被Flink当作泛型类来处理。Flink会把泛型类型当作黑盒，无法获取它们内部的属性；它们也不是由Flink本身序列化的，而是由Kryo序列化的。

在这些类型中，元组类型和POJO类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为Flink的POJO类型。

 类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；

 类有一个公共的无参构造方法；

 类中的所有字段是public且非final的；或者有一个公共的getter和setter方法，这些方法需要符合Java bean的命名规范。

所以我们看到，之前的UserBehavior，就是我们创建的符合Flink POJO定义的数据类型。

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})

public class TransMap {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

     env.setParallelism(1);

 

    DataStreamSource<Event> stream = env.fromElements(

        new Event("Mary", "./home", 1000L),

        new Event("Bob", "./cart", 2000L)

    );

 

    // 传入匿名类，实现MapFunction

    stream.map(new MapFunction<Event, String>() {

      @Override

      public String map(Event e) throws Exception {

        return e.user;

      }

    });

 

    // 传入MapFunction的实现类

    stream.map(new UserExtractor()).print();

 

    env.execute();

  }

  public static class UserExtractor implements MapFunction<Event, String> {

    @Override

    public String map(Event e) throws Exception {

      return e.user;

    }

  }

}

public class TransFilter {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

 

    DataStreamSource<Event> stream = env.fromElements(

        new Event("Mary", "./home", 1000L),

        new Event("Bob", "./cart", 2000L)

    );

 

    // 传入匿名类实现FilterFunction

    stream.filter(new FilterFunction<Event>() {

      @Override

      public boolean filter(Event e) throws Exception {

        return e.user.equals("Mary");

      }

    });

    // 传入FilterFunction实现类

    stream.filter(new UserFilter()).print();  

    env.execute();

  }

  public static class UserFilter implements FilterFunction<Event> {

    @Override

    public boolean filter(Event e) throws Exception {

      return e.user.equals("Mary");

    }

  }

}

public class TransFlatmap {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    DataStreamSource<Event> stream = env.fromElements(

        new Event("Mary", "./home", 1000L),

        new Event("Bob", "./cart", 2000L)

    );

    stream.flatMap(new MyFlatMap()).print();

    env.execute();

  }

  public static class MyFlatMap implements FlatMapFunction<Event, String> {

    @Override

    public void flatMap(Event value, Collector<String> out) throws Exception {

      if (value.user.equals("Mary")) {

        out.collect(value.user);

      } else if (value.user.equals("Bob")) {
        out.collect(value.user);

        out.collect(value.url);

      }

    }

  }

}

public class TransKeyBy {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

 

    DataStreamSource<Event> stream = env.fromElements(

        new Event("Mary", "./home", 1000L),

        new Event("Bob", "./cart", 2000L)

    );

 

    // 使用Lambda表达式

    KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user);

 

    // 使用匿名类实现KeySelector

    KeyedStream<Event, String> keyedStream1 = stream.keyBy(new KeySelector<Event, String>() {

      @Override

      public String getKey(Event e) throws Exception {

        return e.user;

      }

    });

    env.execute();

  }

}

 sum()：在输入流上，对指定的字段做叠加求和的操作。

 min()：在输入流上，对指定的字段求最小值。

 max()：在输入流上，对指定的字段求最大值。

 minBy()：与min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而minBy()则会返回包含字段最小值的整条数据。

 maxBy()：与max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。

public class TransAggregation {

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

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

   env.setParallelism(1);

   DataStreamSource<Event> stream = env.fromElements(

       new Event("Mary", "./home", 1000L),

       new Event("Bob", "./cart", 2000L)

   );

   stream.keyBy(e -> e.user).max("timestamp");   // 指定字段名称

   env.execute();

 }

}

public interface ReduceFunction<T> extends Function, Serializable {

  T reduce(T value1, T value2) throws Exception;

}

public class TransReduce {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    // 这里的ClickSource()使用了之前自定义数据源小节中的ClickSource()

    env.addSource(new ClickSource())

        // 将Event数据类型转换成元组类型

        .map(new MapFunction<Event, Tuple2<String, Long>>() {

          @Override

          public Tuple2<String, Long> map(Event e) throws Exception {

            return Tuple2.of(e.user, 1L);

          }

        })

        .keyBy(r -> r.f0) // 使用用户名来进行分流

        .reduce(new ReduceFunction<Tuple2<String, Long>>() {

          @Override

          public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {

            // 每到一条数据，用户pv的统计值加1

            return Tuple2.of(value1.f0, value1.f1 + value2.f1);

          }

        })

        .keyBy(r -> true) // 为每一条数据分配同一个key，将聚合结果发送到一条流中去

        .reduce(new ReduceFunction<Tuple2<String, Long>>() {

          @Override

          public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {

           // 将累加器更新为当前最大的pv统计值，然后向下游发送累加器的值

            return value1.f1 > value2.f1 ? value1 : value2;

          }

        })

        .print();

    env.execute();

  }

}

public class TransFunctionUDF {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

DataStream<String> clicks = env.readTextFile("clicks.csv");    

DataStream<String> stream = clicks.filter(new FlinkFilter());

 stream.print();

env.execute();

  }

  public static class FlinkFilter implements FilterFunction<String> {

    @Override

    public boolean filter(String value) throws Exception {

      return value.contains("home");

    }

  }

DataStream<String> stream = clicks.filter(new FilterFunction<String>() {

@Override

public boolean filter(String value) throws Exception {

  return value.contains("home");

}

});

DataStream<String> clicks = env.readTextFile("clicks.csv");

DataStream<String> stream = clicks.filter(new KeyWordFilter("home"));

public static class KeyWordFilter implements FilterFunction<String> {

  private String keyWord;

  KeyWordFilter(String keyWord) { this.keyWord = keyWord; }

  @Override

  public boolean filter(String value) throws Exception {

    return value.contains(this.keyWord);

  }

}

public class TransFunctionLambda {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    DataStreamSource<Event> clicks = env.fromElements(

        new Event("Mary", "./home", 1000L),

        new Event("Bob", "./cart", 2000L)

    );

    //map函数使用Lambda表达式，不需要进行类型声明

    DataStream<String> stream = clicks.map(event -> event.url);

    stream.print();

	  env.execute();

  }

}

DataStream<String> stream = clicks
.map(event -> Tuple2.of(event.user, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

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

public class RichFunctionExample {

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

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.setParallelism(2);

  env

      .fromElements(1,2,3,4)

      .map(new RichMapFunction<Integer, Integer>() {

        @Override

        public void open(Configuration parameters) throws Exception {

          super.open(parameters);

          System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");

        }

        @Override

        public Integer map(Integer integer) throws Exception {

          return integer + 1;

        }

        @Override

        public void close() throws Exception {

          super.close();

          System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
        }

      })

      .print();

  env.execute();

}

}

public class MyFlatMap extends RichFlatMapFunction<IN, OUT>> {

@Override

public void open(Configuration configuration) {

  // 做一些初始化工作

  // 例如建立一个和MySQL的连接

}

@Override
public void flatMap(IN in, Collector<OUT out) {
  // 对数据库进行读写
}
@Override
public void close() {
  // 清理工作，关闭和MySQL数据库的连接。
}
}

public class ShuffleExample {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4).setParallelism(1);

    stream.shuffle().print().setParallelism(2);

    env.execute();

  }

}

stream.addSink(new SinkFunction(…));

 行编码：StreamingFileSink.forRowFormat（basePath，rowEncoder）。

 批量编码：StreamingFileSink.forBulkFormat（basePath，bulkWriterFactory）。
 

public class SinkFile {

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

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

   env.setParallelism(1);

   DataStream<String> input = env.fromElements("hello world", "hello flink");

   final StreamingFileSink<String> sink = StreamingFileSink

       .forRowFormat(new Path("./output"), 

new SimpleStringEncoder<String>("UTF-8")).withRollingPolicy(

           DefaultRollingPolicy.builder()

             .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))

             .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))

             .withMaxPartSize(1024 * 1024 * 1024)

             .build())

       .build();

   input.addSink(sink);

   env.execute();

 }

}

至少包含15分钟的数据

最近5分钟没有收到新的数据

文件大小已达到1 GB

public class SinkKafka {

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

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

   env.setParallelism(1);

   Properties properties = new Properties();

   properties.put("bootstrap.servers", "hadoop102:9092");

   DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");

   stream

       .addSink(new FlinkKafkaProducer<String>(

           "clicks",

           new SimpleStringSchema(),

           properties

       ));

   env.execute();

 }

}

bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --topic clicks

<dependency>

 <groupId>org.apache.bahir</groupId>

 <artifactId>flink-connector-redis_2.11</artifactId>

 <version>1.0</version>

</dependency>

public class SinkRedis {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("hadoop200").build();

    env.addSource(new ClickSource())

        .addSink(new RedisSink<Event>(conf, new MyRedisMapper()));

    env.execute();

	}

}

 JFlinkJedisConfigBase：Jedis的连接配置

 RedisMapper：Redis映射类接口，说明怎样将数据转换成可以写入Redis的类型

public static class MyRedisMapper implements RedisMapper<Event> {

  @Override

  public String getKeyFromData(Event e) {

    return e.user;

  }

  @Override

  public String getValueFromData(Event e) {

    return e.url;

  }

  @Override

  public RedisCommandDescription getCommandDescription() {

    return new RedisCommandDescription(RedisCommand.HSET, "clicks");

  }

}

<dependency>

  <groupId>org.apache.flink</groupId>

  <artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>

  <version>${flink.version}</version>

</dependency>

public class SinkToEs {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    ArrayList<HttpHost> httpHosts = new ArrayList<>();

    httpHosts.add(new HttpHost("hadoop102", 9200, "http"));

    ElasticsearchSink.Builder<Event> esBuilder = new ElasticsearchSink.Builder<>(

        httpHosts,

        new ElasticsearchSinkFunction<Event>() {

          @Override

          public void process(Event event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {

            HashMap<String, String> data = new HashMap<>();

            data.put(event.user, event.url);

            IndexRequest indexRequest = Requests

                .indexRequest()

                .index("clicks")

               .type("type")

                .source(data);

            requestIndexer.add(indexRequest);

          }

        }

    );

    DataStream<Event> stream = env.fromElements(new Event("Mary", "./home", 1000L));

    stream.addSink(esBuilder.build());

    env.execute();

  }

}

 httpHosts：连接到的Elasticsearch集群主机列表

 elasticsearchSinkFunction：这并不是我们所说的SinkFunction，而是用来说明具体处理逻辑、准备数据向Elasticsearch发送请求的函数

<dependency>

 <groupId>org.apache.flink</groupId>

 <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>

 <version>${flink.version}</version>

</dependency>

<dependency>

 <groupId>mysql</groupId>

 <artifactId>mysql-connector-java</artifactId>

 <version>5.1.47</version>

</dependency>

mysql> create table clicks(

  -> user varchar(20) not null,

  -> url varchar(100) not null);

public class SinkToMySQL {

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

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.setParallelism(1);

  env

      .fromElements(

          new Event("Mary", "./home", 1000L),

          new Event("Bob", "./cart", 2000L)

      )

      .addSink(

          JdbcSink.sink(

              "INSERT INTO clicks (user, url) VALUES (?, ?)",

              (statement, r) -> {

                statement.setString(1, r.user);

                statement.setString(2, r.url);

              },

              JdbcExecutionOptions.builder()

                  .withBatchSize(1000)

                  .withBatchIntervalMs(200)

                  .withMaxRetries(5)

                  .build(),

              new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()

                  .withUrl("jdbc:mysql://localhost:3306/test")

                  // 使用MySQL 5.7的话，没有cj

                  .withDriverName("com.mysql.cj.jdbc.Driver")

                  .withUsername("username")

                  .withPassword("password")

                  .build()

          )

      );

  env.execute();

}

}

  stream.addSink(new MySinkFunction<String>());

事件时间，是指每个事件在对应的设备上发生的时间，也就是数据生成的时间。

数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录的“时间戳”（Timestamp）。

在事件时间语义下，我们对于时间的衡量，就不看任何机器的系统时间了，而是依赖于数据本身。由于分布式系统中网络传输延迟的不确定性，实际应用中我们要面对的数据流往往是乱序的。在这种情况下，就不能简单地把数据自带的时间戳当作时钟了，而需要用另外的标志来表示事件时间进展，在Flink中把它叫作事件时间的“水位线”（Watermarks）。

水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据

水位线主要的内容是一个时间戳，用来表示当前事件时间的进展

水位线是基于数据的时间戳生成的

水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进

水位线可以通过设置延迟，来保证正确处理乱序数据

一个水位线Watermark(t)，表示在当前流中事件时间已经达到了时间戳t, 这代表t之前的所有数据都到齐了，之后流中不会出现时间戳t’ ≤ t的数据

水位线是Flink流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对乱序数据的正确处理。

public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(

WatermarkStrategy<T> watermarkStrategy)

DataStream<Event> stream = env.addSource(new ClickSource());

DataStream<Event> withTimestampsAndWatermarks = 

stream.assignTimestampsAndWatermarks(*<watermark strategy>*);

public interface WatermarkStrategy<T> 

  extends TimestampAssignerSupplier<T>,

      WatermarkGeneratorSupplier<T>{

  @Override

  TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);

  @Override

  WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);

}

TimestampAssigner：主要负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。

WatermarkGenerator：主要负责按照既定的方式，基于时间戳生成水位线。

在WatermarkGenerator接口中，主要又有两个方法：onEvent()和onPeriodicEmit()。

onEvent：每个事件（数据）到来都会调用的方法，它的参数有当前事件、时间戳，以及允许发出水位线的一个WatermarkOutput，可以基于事件做各种操作

onPeriodicEmit：周期性调用的方法，可以由WatermarkOutput发出水位线。周期时间为处理时间，可以调用环境配置的.setAutoWatermarkInterval()方法来设置，默认为200ms。

env.getConfig().setAutoWatermarkInterval(60 * 1000L);

stream.assignTimestampsAndWatermarks(

    WatermarkStrategy.<Event>forMonotonousTimestamps()

        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {

          @Override

          public long extractTimestamp(Event element, long recordTimestamp) {

            return element.timestamp;

          }

        })

);

public class WatermarkExample {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    env

        .addSource(new ClickSource())

        // 插入水位线的逻辑

        .assignTimestampsAndWatermarks(

             // 针对乱序流插入水位线，延迟时间设置为5s

              WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))

			 .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {

                  // 抽取时间戳的逻辑

                  @Override

                  public long extractTimestamp(Event element, long recordTimestamp) {

                    return element.timestamp;

                  }

                })
        )
        .print();
    env.execute();
  }

}

import com.atguigu.bean.Event;

import org.apache.flink.api.common.eventtime.*;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

 

// 自定义水位线的产生

public class CustomPeriodicWatermarkExample {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    env

        .addSource(new ClickSource())

        .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())

        .print();

    env.execute();

  }

 

  public static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {

    @Override

    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {

      return new SerializableTimestampAssigner<Event>() {

        @Override

        public long extractTimestamp(Event element, long recordTimestamp) {

          return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段

        }

      };

    }

    @Override

    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {

      return new CustomBoundedOutOfOrdernessGenerator();

    }

  }

 

  public static class CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator<Event> {

    private Long delayTime = 5000L; // 延迟时间

    private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳

    @Override

    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {

      // 每来一条数据就调用一次

      maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳

    }

    @Override

    public void onPeriodicEmit(WatermarkOutput output) {
      // 发射水位线，默认200ms调用一次

      output.emitWatermark(new Watermark(maxTs - delayTime - 1L));

    }

  }

}

public class PunctuatedGenerator implements WatermarkGenerator<Event> {

  @Override

  public void onEvent(Event r, long eventTimestamp, WatermarkOutput output) {

// 只有在遇到特定的itemId时，才发出水位线

    if (r.user.equals("Mary")) {

      output.emitWatermark(new Watermark(r.timestamp - 1));

    }

  }

  @Override

  public void onPeriodicEmit(WatermarkOutput output) {

    // 不需要做任何事情，因为我们在onEvent方法中发射了水位线

  }

}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import org.apache.flink.streaming.api.watermark.Watermark;

import java.sql.Timestamp;

import java.util.Calendar;

import java.util.Random;

 

public class EmitWatermarkInSourceFunction {

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

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    env.addSource(new ClickSource()).print();

    env.execute();

  }

 

  // 泛型是数据源中的类型

  public static class ClickSource implements SourceFunction<Event> {

    private boolean running = true;

    @Override

    public void run(SourceContext<Event> sourceContext) throws Exception {

      Random random = new Random();

      String[] userArr = {"Mary", "Bob", "Alice"};

      String[] urlArr  = {"./home", "./cart", "./prod?id=1"};

      while (running) {

        long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时间戳

        String username = userArr[random.nextInt(userArr.length)];

        String url    = urlArr[random.nextInt(urlArr.length)];

        Event event = new Event(username, url, currTs);

        // 使用collectWithTimestamp方法将数据发送出去，并指明数据中的时间戳的字段

        sourceContext.collectWithTimestamp(event, event.timestamp);

        // 发送水位线

        sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));

        Thread.sleep(1000L);

      }
    }

    @Override

    public void cancel() {

      running = false;

    }

  }

}

（1）上游并行子任务发来不同的水位线，当前任务会为每一个分区设置一个“分区水位线”（Partition Watermark），这是一个分区时钟；而当前任务自己的时钟，就是所有分区时钟里最小的那个。

（2）当有一个新的水位线（第一分区的4）从上游传来时，当前任务会首先更新对应的分区时钟；然后再次判断所有分区时钟中的最小值，如果比之前大，说明事件时间有了进展，当前任务的时钟也就可以更新了。这里要注意，更新后的任务时钟，并不一定是新来的那个分区水位线，比如这里改变的是第一分区的时钟，但最小的分区时钟是第三分区的3，于是当前任务时钟就推进到了3。当时钟有进展时，当前任务就会将自己的时钟以水位线的形式，广播给下游所有子任务。

（3）再次收到新的水位线（第二分区的7）后，执行同样的处理流程。首先将第二个分区时钟更新为7，然后比较所有分区时钟；发现最小值没有变化，那么当前任务的时钟也不变，也不会向下游任务发出水位线。

（4）同样道理，当又一次收到新的水位线（第三分区的6）之后，第三个分区时钟更新为6，同时所有分区时钟最小值变成了第一分区的4，所以当前任务的时钟推进到4，并发出时间戳为4的水位线，广播到下游各个分区任务。

水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完之前所有数据”为标准来确定自己的时钟，就可以保证窗口处理的结果总是正确的。10.3 窗口（Window）