Flink 流处理特性

支持高吞吐、低延迟、高性能的流处理

支持带有事件时间的窗口（Window）操作

支持有状态计算的 Exactly-once 语义

支持高度灵活的窗口（Window）操作，支持基于 time、count、session，以及 data-driven 的窗口操作

支持具有 Backpressure 功能的持续流模型

支持基于轻量级分布式快照（Snapshot）实现的容错

一个运行时同时支持 Batch on Streaming 处理和 Streaming 处理

Flink 在 JVM 内部实现了自己的内存管理

支持迭代计算

支持程序自动优化：避免特定情况下 Shuffle、排序等昂贵操作，中间结果有必要进行缓存

Flink 基石

Flink 之所以能这么流行，离不开它最重要的四个基石：Checkpoint、State、Time、Window。

首先是 Checkpoint 机制，这是 Flink 最重要的一个特性。Flink 基于Chandy-Lamport算法实现了一个分布式的一致性的快照，从而提供了一致性的语义。Chandy-Lamport 算法实际上在 1985 年的时候已经被提出来，但并没有被很广泛的应用，而 Flink 则把这个算法发扬光大了。

Spark 最近在实现 Continue streaming，Continue streaming 的目的是为了降低它处理的延时，其也需要提供这种一致性的语义，最终采用 Chandy-Lamport 这个算法，说明 Chandy-Lamport 算法在业界得到了一定的肯定。

提供了一致性的语义之后，Flink 为了让用户在编程时能够更轻松、更容易地去管理状态，还提供了一套非常简单明了的 State API，包括里面的有 ValueState、ListState、MapState，近期添加了 BroadcastState，使用 State API 能够自动享受到这种一致性的语义。

除此之外，Flink 还实现了 Watermark 的机制，能够支持基于事件的时间的处理，或者说基于系统时间的处理，能够容忍数据的延时、容忍数据的迟到、容忍乱序的数据。

另外流计算中一般在对流数据进行操作之前都会先进行开窗，即基于一个什么样的窗口上做这个计算。Flink 提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义的窗口。

批处理与流处理

批处理的特点是有界、持久、大量，批处理非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。流处理的特点是无界、实时，流处理方式无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。

在 Spark 生态体系中，对于批处理和流处理采用了不同的技术框架，批处理由 SparkSQL 实现，流处理由 Spark Streaming 实现，这也是大部分框架采用的策略，使用独立的处理器实现批处理和流处理，而 Flink 可以同时实现批处理和流处理。

Flink 是如何同时实现批处理与流处理的呢？答案是，Flink 将批处理（即处理有限的静态数据）视作一种特殊的流处理。

Flink 的核心计算架构是下图中的 Flink Runtime 执行引擎，它是一个分布式系统，能够接受数据流程序并在一台或多台机器上以容错方式执行。

Flink Runtime 执行引擎可以作为 YARN（Yet Another Resource Negotiator）的应用程序在集群上运行，也可以在 Mesos 集群上运行，还可以在单机上运行（这对于调试 Flink 应用程序来说非常有用）。

上图为 Flink 技术栈的核心组成部分，值得一提的是，Flink 分别提供了面向流式处理的接口（DataStream API）和面向批处理的接口（DataSet API）。因此，Flink 既可以完成流处理，也可以完成批处理。Flink 支持的拓展库涉及机器学习（FlinkML）、复杂事件处理（CEP）、以及图计算（Gelly），还有分别针对流处理和批处理的 Table API。

能被 Flink Runtime 执行引擎接受的程序很强大，但是这样的程序有着冗长的代码，编写起来也很费力，基于这个原因，Flink 提供了封装在 Runtime 执行引擎之上的 API，以帮助用户方便地生成流式计算程序。Flink 提供了用于流处理的 DataStream API 和用于批处理的 DataSet API。值得注意的是，尽管 Flink Runtime 执行引擎是基于流处理的，但是 DataSet API 先于 DataStream API 被开发出来，这是因为工业界对无限流处理的需求在 Flink 诞生之初并不大。

DataStream API 可以流畅地分析无限数据流，并且可以用 Java 或者 Scala 等来实现。开发人员需要基于一个叫 DataStream 的数据结构来开发，这个数据结构用于表示永不停止的分布式数据流。

Flink 的分布式特点体现在它能够在成百上千台机器上运行，它将大型的计算任务分成许多小的部分，每个机器执行一部分。Flink 能够自动地确保发生机器故障或者其他错误时计算能够持续进行，或者在修复 bug 或进行版本升级后有计划地再执行一次。这种能力使得开发人员不需要担心运行失败。Flink 本质上使用容错性数据流，这使得开发人员可以分析持续生成且永远不结束的数据（即流处理）。

Flink 运行架构

Flink 程序结构

Flink 程序的基本构建块是流和转换（请注意，Flink 的 DataSet API 中使用的 DataSet 也是内部流 ）。从概念上讲，流是（可能永无止境的）数据记录流，而转换是将一个或多个流作为一个或多个流的操作。输入，并产生一个或多个输出流。

Flink 应用程序结构就是如上图所示：

Source: 数据源，Flink 在流处理和批处理上的 source 大概有 4 类：基于本地集合的 source、基于文件的 source、基于网络套接字的 source、自定义的 source。自定义的 source 常见的有 Apache kafka、RabbitMQ 等，当然你也可以定义自己的 source。

Transformation：数据转换的各种操作，有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select等，操作很多，可以将数据转换计算成你想要的数据。

Sink：接收器，Flink 将转换计算后的数据发送的地点 ，你可能需要存储下来，Flink 常见的 Sink 大概有如下几类：写入文件、打印出来、写入 socket 、自定义的 sink 。自定义的 sink 常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等，同理你也可以定义自己的 sink。

Flink 并行数据流

Flink 程序在执行的时候，会被映射成一个 Streaming Dataflow，一个 Streaming Dataflow 是由一组 Stream 和 Transformation Operator 组成的。在启动时从一个或多个 Source Operator 开始，结束于一个或多个 Sink Operator。

Flink 程序本质上是并行的和分布式的，在执行过程中，一个流(stream)包含一个或多个流分区，而每一个 operator 包含一个或多个 operator 子任务。操作子任务间彼此独立，在不同的线程中执行，甚至是在不同的机器或不同的容器上。operator 子任务的数量是这一特定 operator 的并行度。相同程序中的不同 operator 有不同级别的并行度。

一个 Stream 可以被分成多个 Stream 的分区，也就是 Stream Partition。一个 Operator 也可以被分为多个 Operator Subtask。如上图中，Source 被分成 Source1 和 Source2，它们分别为 Source 的 Operator Subtask。每一个 Operator Subtask 都是在不同的线程当中独立执行的。一个 Operator 的并行度，就等于 Operator Subtask 的个数。上图 Source 的并行度为 2。而一个 Stream 的并行度就等于它生成的 Operator 的并行度。

数据在两个 operator 之间传递的时候有两种模式：

One to One 模式：两个 operator 用此模式传递的时候，会保持数据的分区数和数据的排序；如上图中的 Source1 到 Map1，它就保留的 Source 的分区特性，以及分区元素处理的有序性。

Redistributing （重新分配）模式：这种模式会改变数据的分区数；每个一个 operator subtask 会根据选择 transformation 把数据发送到不同的目标 subtasks,比如 keyBy()会通过 hashcode 重新分区,broadcast()和 rebalance()方法会随机重新分区；

Task 和 Operator chain

Flink的所有操作都称之为Operator，客户端在提交任务的时候会对Operator进行优化操作，能进行合并的Operator会被合并为一个Operator，合并后的Operator称为Operator chain，实际上就是一个执行链，每个执行链会在TaskManager上一个独立的线程中执行。

任务调度与执行

当Flink执行executor会自动根据程序代码生成DAG数据流图；

ActorSystem创建Actor将数据流图发送给JobManager中的Actor；

JobManager会不断接收TaskManager的心跳消息，从而可以获取到有效的TaskManager；

JobManager通过调度器在TaskManager中调度执行Task（在Flink中，最小的调度单元就是task，对应就是一个线程）；

在程序运行过程中，task与task之间是可以进行数据传输的。

Job Client：

主要职责是提交任务, 提交后可以结束进程, 也可以等待结果返回；

Job Client 不是 Flink 程序执行的内部部分，但它是任务执行的起点；

Job Client 负责接受用户的程序代码，然后创建数据流，将数据流提交给 Job Manager 以便进一步执行。执行完成后，Job Client 将结果返回给用户。

JobManager：

主要职责是调度工作并协调任务做检查点；

集群中至少要有一个 master，master 负责调度 task，协调checkpoints 和容错；

高可用设置的话可以有多个 master，但要保证一个是 leader, 其他是standby；

Job Manager 包含 Actor System、Scheduler、CheckPoint三个重要的组件；

JobManager从客户端接收到任务以后, 首先生成优化过的执行计划, 再调度到TaskManager中执行。

TaskManager：

主要职责是从JobManager处接收任务, 并部署和启动任务, 接收上游的数据并处理；

Task Manager 是在 JVM 中的一个或多个线程中执行任务的工作节点；

TaskManager在创建之初就设置好了Slot, 每个Slot可以执行一个任务。

对于 flink 的流任务客户端首先会生成 StreamGraph，接着生成 JobGraph，然后将 jobGraph 提交给 Jobmanager 由它完成 jobGraph 到 ExecutionGraph 的转变，最后由 jobManager 调度执行。

上图所示有一个由 data source、MapFunction和 ReduceFunction 组成的程序，data source 和 MapFunction 的并发度都为 4，而 ReduceFunction 的并发度为 3。一个数据流由 Source-Map-Reduce 的顺序组成，在具有 2 个TaskManager、每个 TaskManager 都有 3 个 Task Slot 的集群上运行。

可以看出 flink 的拓扑生成提交执行之后，除非故障，否则拓扑部件执行位置不变，并行度由每一个算子并行度决定，类似于 storm。而 spark Streaming 是每个批次都会根据数据本地性和资源情况进行调度，无固定的执行拓扑结构。 flink 是数据在拓扑结构里流动执行，而 Spark Streaming 则是对数据缓存批次并行处理。

1. 内置基础 Source

(1) 集合/元素 Source

// 从内存集合读取（测试用）
DataStream<String> stream = env.fromElements("A", "B", "C");
​
// 从序列生成
DataStream<Long> numbers = env.fromSequence(1, 100);

(2) 文件 Source

// 读取文本文件（批/流模式）
DataStream<String> fileStream = env.readTextFile("file:///path/to/data.txt");
​
// 使用FileSource（推荐）
FileSource<String> source = FileSource.forRecordStreamFormat(
    new TextLineInputFormat(), new Path("hdfs://path/to/files")
).build();
DataStream<String> stream = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "File Source");

---

2. 消息队列 Source

(1) Apache Kafka

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("input-topic")
    .setDeserializer(new SimpleStringSchema())
    .setStartingOffsets(OffsetsInitializer.earliest())
    .build();
​
DataStream<String> kafkaStream = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "Kafka Source");

(2) RabbitMQ/Pulsar

<!-- 添加依赖 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq_2.12</artifactId>
    <version>1.16.0</version>
</dependency>
RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
    .setHost("rabbitmq").setPort(5672).build();
​
DataStream<String> rmqStream = env.addSource(
    new RMQSource<>(connectionConfig, "queue-name", new SimpleStringSchema()));

---

3. 数据库 Source

(1) JDBC Source（批处理）

JdbcInputFormat inputFormat = JdbcInputFormat.buildJdbcInputFormat()
    .setDrivername("com.mysql.jdbc.Driver")
    .setDBUrl("jdbc:mysql://mysql:3306/db")
    .setUsername("user")
    .setPassword("pass")
    .setQuery("SELECT id, name FROM users")
    .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
    .finish();
​
DataStream<Row> dbStream = env.createInput(inputFormat);

(2) CDC (Change Data Capture)

-- Flink SQL 方式（MySQL CDC）
CREATE TABLE mysql_users (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'password',
    'database-name' = 'test',
    'table-name' = 'users'
);

支持的CDC连接器：

• MySQL CDC

• PostgreSQL CDC

• Oracle CDC (商业版)

• MongoDB CDC

---

4. 分布式文件系统 Source

(1) Hadoop HDFS

// 通过FileSource读取
FileSource<String> hdfsSource = FileSource.forRecordStreamFormat(
    new TextLineInputFormat(), 
    new Path("hdfs://namenode:8020/path/to/data")
).build();

(2) Amazon S3

// 需配置S3访问凭据
env.getCheckpointConfig().setCheckpointStorage("s3://my-bucket/checkpoints");
DataStream<String> s3Stream = env.readTextFile("s3://bucket/path/to/file");

---

5. 实时消息 Source

(1) Apache Pulsar

PulsarSource<String> source = PulsarSource.builder()
    .setServiceUrl("pulsar://pulsar:6650")
    .setTopics("persistent://tenant/ns/topic")
    .setDeserializationSchema(new SimpleStringSchema())
    .setSubscriptionName("flink-subscription")
    .build();

(2) AWS Kinesis

KinesisStreamsSource<String> source = KinesisStreamsSource.<String>builder()
    .setStreamName("my-stream")
    .setRegion("us-east-1")
    .setDeserializationSchema(new SimpleStringSchema())
    .build();