一、前言：为什么是 Flink？

1.1 流处理框架演进

1.2 为什么选择 Flink？

1.3 Flink vs Spark Streaming

二、Flink 核心概念

2.1 概念全景图

2.2 DataStream：流的抽象

代码示例：

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 从 Kafka 读取数据
DataStream<String> source = env.addSource(
    new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties)
);

// 数据处理
DataStream<Order> orders = source
    .map(json -> JSON.parseObject(json, Order.class))  // 解析 JSON
    .filter(order -> order.getAmount() > 0)            // 过滤
    .keyBy(Order::getUserId)                           // 按用户分组
    .window(TumblingEventTimeWindows.of(Time.hours(1))) // 1小时窗口
    .sum("amount");                                    // 求和

// 输出到 Kafka
orders.addSink(new FlinkKafkaProducer<>("output-topic", new OrderSchema(), properties));

// 执行
env.execute("Order Processing Job");

2.3 算子链与 Task

算子链优化：

• 减少序列化/反序列化
• 减少线程切换
• 提高吞吐量

// 禁用算子链
stream.map(x -> x).disableChaining();

// 开始新的算子链
stream.map(x -> x).startNewChain();

// 全局禁用算子链
env.disableOperatorChaining();

三、Flink 架构详解

3.1 整体架构

3.2 核心组件职责

3.3 Task Slot 与并行度

关键概念：

配置方式：

// 全局并行度
env.setParallelism(4);

// 算子级别并行度
stream.map(x -> x).setParallelism(2);

// flink-conf.yaml
parallelism.default: 4
taskmanager.numberOfTaskSlots: 4

3.4 作业提交流程

3.5 Graph 转换过程

四、部署模式

4.1 部署模式对比

4.2 资源管理器集成

4.3 Kubernetes 部署

# Flink Kubernetes Operator 示例
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-example
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.checkpoints.dir: s3://bucket/checkpoints
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
    replicas: 3
  job:
    jarURI: s3://bucket/jobs/my-job.jar
    parallelism: 6
    upgradeMode: savepoint

五、数据流转详解

5.1 数据交换模式

// Forward（默认，一对一）
stream.map(x -> x)  // 自动 Forward

// Rebalance（轮询）
stream.rebalance()

// Rescale（本地轮询）
stream.rescale()

// Shuffle（随机）
stream.shuffle()

// Broadcast（广播）
stream.broadcast()

// KeyBy（哈希分区）
stream.keyBy(x -> x.getKey())

5.2 背压机制

背压监控：

六、流批一体

6.1 流批一体理念

6.2 执行模式

// 流模式（默认）
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

// 批模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

// 自动模式（根据数据源判断）
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

6.3 流批一体示例

// 同一份代码，既可以流处理也可以批处理
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Order> orders = env
    .readSource(...)  // 可以是 Kafka（流）或文件（批）
    .map(Order::parse)
    .keyBy(Order::getUserId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .sum("amount");

orders.sinkTo(...);

env.execute();

七、Flink SQL 简介

7.1 SQL 架构

7.2 SQL 快速上手

-- 创建 Kafka 源表
CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
);

-- 创建结果表
CREATE TABLE order_stats (
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    user_id BIGINT,
    total_amount DECIMAL(10,2),
    order_count BIGINT,
    PRIMARY KEY (window_start, window_end, user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/analytics',
    'table-name' = 'order_stats'
);

-- 窗口聚合
INSERT INTO order_stats
SELECT 
    TUMBLE_START(order_time, INTERVAL '1' HOUR) as window_start,
    TUMBLE_END(order_time, INTERVAL '1' HOUR) as window_end,
    user_id,
    SUM(amount) as total_amount,
    COUNT(*) as order_count
FROM orders
GROUP BY 
    TUMBLE(order_time, INTERVAL '1' HOUR),
    user_id;

八、快速上手指南

8.1 开发环境搭建

<!-- pom.xml -->
<properties>
    <flink.version>1.17.0</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <!-- Flink 核心 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    
    <!-- Flink Clients -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
    
    <!-- Kafka Connector -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>
    
    <!-- Table API & SQL -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

8.2 Hello World

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 读取数据
        DataStream<String> text = env.socketTextStream("localhost", 9999);
        
        // 3. 处理数据
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(Tuple2.of(word, 1));
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(t -> t.f0)
            .sum(1);
        
        // 4. 输出结果
        counts.print();
        
        // 5. 执行
        env.execute("Socket Word Count");
    }
}

8.3 本地运行与调试

// 本地模式（开发调试）
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

// 本地 WebUI 模式（可以查看执行图）
Configuration config = new Configuration();
config.setInteger(RestOptions.PORT, 8081);
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);

// 设置并行度为 1，方便调试
env.setParallelism(1);

九、常用配置速查

9.1 核心配置

# flink-conf.yaml

# JobManager 配置
jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1600m

# TaskManager 配置
taskmanager.memory.process.size: 4096m
taskmanager.numberOfTaskSlots: 4

# 并行度
parallelism.default: 4

# Checkpoint 配置
execution.checkpointing.interval: 60000
execution.checkpointing.mode: EXACTLY_ONCE
state.backend: rocksdb
state.checkpoints.dir: hdfs:///flink/checkpoints

# 高可用配置
high-availability: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.storageDir: hdfs:///flink/ha

9.2 内存配置详解

# 内存配置示例
taskmanager.memory.process.size: 8g
taskmanager.memory.flink.size: 6g
taskmanager.memory.task.heap.size: 2g
taskmanager.memory.managed.size: 2g
taskmanager.memory.network.min: 256m
taskmanager.memory.network.max: 1g
taskmanager.memory.jvm-overhead.min: 256m
taskmanager.memory.jvm-overhead.max: 1g

十、总结

10.1 Flink 核心优势

10.2 核心概念回顾

10.3 一句话总结

Flink = 真正的流处理 + 流批一体 + 有状态计算 + 精确一次语义

如果你需要低延迟、高吞吐、有状态的流处理，Flink 是当前最佳选择。