Flink介绍

一、简介

Apache Flink 是一个开源的流处理框架，专为在无界和有界数据流上进行有状态的计算而设计。它具有高吞吐量、低延迟、容错性强等特点，同时支持流批统一处理，即可以用相同的代码处理批量数据和实时数据流。

1.1 主要特性

1、流批统一：Flink 以流处理为核心，将批处理看作是流处理的一种特殊情况，使用相同的编程模型和执行引擎来处理有界和无界数据流。
2、状态管理：支持在流处理过程中管理和维护状态，例如在处理实时数据时可以记住之前的计算结果。Flink 提供了高效的状态后端（如内存、文件系统、RocksDB 等）来存储状态。
3、时间语义：支持事件时间、处理时间和摄入时间三种时间语义，能够处理乱序事件，确保在复杂的实时场景下计算结果的准确性。
4、窗口操作：丰富的窗口操作，如时间窗口、计数窗口、会话窗口等，允许用户在流数据上进行分组和聚合操作。
5、容错机制：通过定期的 Checkpoint 和 Savepoint 机制实现容错，确保在发生故障时可以恢复计算状态，保证 Exactly-Once 的语义。
6、分布式执行：基于分布式架构，可以在集群环境中并行处理大规模数据，并且能够根据集群资源动态调整并行度。

1.2 核心概念

1、流（Stream）
定义：数据的无界序列，可分为有界流（Batch）和无界流（Stream）。有界流的数据量是有限的，处理方式类似于传统的批处理；无界流的数据是持续不断产生的，需要实时处理。
示例：网站的用户点击事件、传感器的实时数据等都属于无界流；而每天的销售数据统计属于有界流。

2、状态（State）
定义：Flink 中的状态指的是算子在处理数据时所维护的信息。在流处理中，很多操作需要记录历史数据，这些历史数据就存储在状态中。
示例：在计算一段时间内的用户点击次数时，算子需要记录每个用户的点击次数，这个点击次数就是状态。

3、时间（Time）
定义：Flink 支持三种时间语义，分别是事件时间（Event Time）、摄入时间（Ingestion Time）和处理时间（Processing Time）。
事件时间：事件实际发生的时间。
摄入时间：事件进入 Flink 系统的时间。
处理时间：事件被 Flink 算子处理的时间。
示例：一个传感器在 10:00 产生了一条数据，该数据在 10:01 进入 Flink 系统，在 10:02 被处理，那么事件时间是 10:00，摄入时间是 10:01，处理时间是 10:02。

4、窗口（Window）
定义：由于流数据是无界的，无法对整个流进行处理，因此需要将流数据划分成有限大小的 “桶”，这些 “桶” 就是窗口。
示例：计算每分钟的用户点击次数，就可以使用一分钟的时间窗口将流数据划分成多个窗口，然后对每个窗口内的数据进行统计。

5、水印（Watermark）
定义：水印是 Flink 中处理乱序事件的一种机制，它是一个时间戳，表示在这个时间戳之前的所有事件都已经到达。
示例：假设设置了一个 5 秒的水印延迟，当接收到一个时间戳为 10:00 的事件时，水印时间为 9:55，表示 9:55 之前的所有事件都已经到达。

1.3 应用场景

实时 ETL：对实时数据流进行清洗、转换和加载，将数据从源系统传输到目标系统。
实时监控：实时监测系统指标、业务数据等，及时发现异常情况并触发警报。
实时推荐：根据用户的实时行为数据进行个性化推荐。

二、Flink部署

2.1 部署在服务器上

1、环境准备
确保系统已经安装了 Java 8 或更高版本，并且配置了 JAVA_HOME 环境变量。可以使用以下命令检查 Java 版本：

java -version

2、下载 Flink
从 Flink 官方网站 下载适合的 Flink 版本
3、解压

tar -zxvf flink-1.16.2-bin-scala_2.12.tgz
cd flink-1.16.2

4、配置 Flink
编辑 conf/flink-conf.yaml 文件

# JobManager 的内存大小
jobmanager.memory.process.size: 1600m

# TaskManager 的内存大小
taskmanager.memory.process.size: 1728m

# TaskManager 的数量
taskmanager.numberOfTaskSlots: 2

# 集群的 Web UI 端口
rest.port: 8081

编辑 conf/master 文件，指定 JobManager 的地址

localhost:8081

编辑 conf/slaves 文件，指定 TaskManager 的地址

localhost

5、启动 Flink 集群
在 Flink 主目录下执行以下命令启动集群：

./bin/start-cluster.sh

6、验证
打开浏览器，访问 http://localhost:8081，如果能够看到 Flink 的 Web UI 界面，则说明部署成功。

2.2 部署在k8s上

1、环境准备
K8s 集群：确保你已经拥有一个可用的 K8s 集群，版本建议在 1.18 及以上。
kubectl 工具：安装并配置好 kubectl 命令行工具，用于与 K8s 集群进行交互。
Flink 镜像：可以使用官方提供的 Flink Docker 镜像，也可以根据需求自定义镜像。

2、创建命名空间（可选）
为 Flink 创建一个独立的命名空间，方便管理和隔离资源。

kubectl create namespace flink

3、配置 Flink 作业管理器（JobManager）
创建jobmanager-deployment.yaml 文件

apiVersion: apps/v1
# 资源的类型是 Deployment
kind: Deployment
# 用于存储资源的元数据信息
metadata:
  # Service 的名称为 flink-service
  name: flink-jobmanager
  # 所属的命名空间为 flink
  namespace: flink
# 定义 Service 的具体配置
spec:
  # 指定要创建的 Pod 副本数量为 1
  replicas: 1
  # 指定 Deployment 要管理的 Pod
  selector:
    matchLabels:
      app: flink
      component: jobmanager
  # 定义了用于创建 Pod 的模板
  template:
    metadata:
      # 为 Pod 添加标签，这些标签必须与 selector 中的标签匹配，以便 Deployment 能够管理这些 Pod。
      labels:
        app: flink
        component: jobmanager
    # 定义了 Pod 的具体配置    
    spec:
      containers:
      - name: jobmanager
        # 指定要使用的 Docker 镜像为 flink:1.16.2-scala_2.12-java11。
        image: flink:1.16.2-scala_2.12-java11
        # 启动容器时运行 jobmanager 服务
        args: ["jobmanager"]
        # 定义 Service 要暴露的端口
        ports:
        # 指定容器内部监听的端口号  
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob
        - containerPort: 8081
          name: ui
        # 定义容器的环境变量
        env:
          # 环境变量的名称为 JOB_MANAGER_RPC_ADDRESS。
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager

创建 JobManager 部署：

kubectl apply -f jobmanager-deployment.yaml

4、配置 Flink 任务管理器（TaskManager）
创建一个 taskmanager-deployment.yaml 文件

apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
  namespace: flink
spec:
  replicas: 2
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: flink:1.16.2-scala_2.12-java11
        args: ["taskmanager"]
        ports:
        - containerPort: 6122
          name: rpc
        - containerPort: 6125
          name: query
        env:
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        - name