📚 课程目录第3章 Flink基础

3、Flink基础：架构、环境搭建与API入门

这一章，我们正式进入Flink的世界。说实话，Flink这几年在实时计算领域几乎是统治级的存在。我最早接触Flink还是1.4版本，那时候社区还没这么火，但它的流处理理念已经让我眼前一亮。今天咱们把Flink的底裤扒一扒——架构、环境、DataStream API、Table API，一条龙讲清楚。

3.1 Flink架构：核心组件与运行原理

先看架构。Flink的架构其实不复杂，但很多人一开始会被各种术语搞晕。我习惯把它拆成两层：运行时层和API层。

核心思想：Flink是一个有状态的流处理引擎。它把一切数据都当作流来处理——批处理只是流处理的一个特例（有限流）。这个理念贯穿了整个架构设计。

咱们先画一张架构图，把各个组件的关系理清楚。

这张图我建议你多看两眼。API层是你写代码的地方，运行时层是Flink帮你干活的地方。我见过不少新手，代码写得飞起，但JobManager挂了就手足无措——说白了，就是没搞懂架构。

3.2 Flink运行环境搭建

环境搭建这块，我推荐两种方式：本地模式用于开发调试，Docker Compose用于模拟集群。咱们一个一个来。

3.2.1 本地模式（Standalone）

本地模式最简单，适合你刚上手时跑Demo。下载解压就行。

# 1. 下载 Flink（以 1.17 为例）
wget https://dlcdn.apache.org/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz

# 2. 解压
tar -xzf flink-1.17.0-bin-scala_2.12.tgz
cd flink-1.17.0

# 3. 启动集群（本地模式）
./bin/start-cluster.sh

# 4. 访问 Web UI
# 浏览器打开 http://localhost:8081

我的习惯：本地开发时，我会把 taskmanager.numberOfTaskSlots 设成 4，这样能同时跑多个任务，调试效率高。但生产环境一般设成 1，避免资源争抢。

3.2.2 Docker Compose 模式

如果你不想污染本地环境，用Docker最省心。我曾经在Mac上折腾了半小时环境变量，后来直接上Docker，5分钟搞定。

# docker-compose.yml
version: "3.8"
services:
  jobmanager:
    image: flink:1.17.0-scala_2.12
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

  taskmanager:
    image: flink:1.17.0-scala_2.12
    depends_on:
      - jobmanager
    command: taskmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
    scale: 2

启动命令：docker-compose up -d。然后访问 localhost:8081，你会看到两个TaskManager在线。

3.3 Flink DataStream API入门

DataStream API是Flink的核心API，也是我工作中用得最多的。说白了，它就是一套处理无界流的算子集合。

3.3.1 第一个流处理程序：WordCount

咱们写一个经典的WordCount，但这次是流式的，不是批处理。

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.util.Collector;

public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 数据源：从 socket 读取（模拟无界流）
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 3. 转换逻辑
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);

        // 4. 输出
        counts.print();

        // 5. 启动任务
        env.execute("Streaming WordCount");
    }

    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}

注意：这里用的是 socketTextStream，生产环境千万别这么干。socket源没有容错机制，一旦断连，数据就丢了。我一般用Kafka作为生产数据源。

3.3.2 常用算子一览

DataStream API的算子很多，但常用的就这几个。我整理了一张表，方便你对照。

算子	用途	示例
`map`	一对一转换	`stream.map(x -> x * 2)`
`flatMap`	一对多转换	`stream.flatMap(...)`
`filter`	过滤元素	`stream.filter(x -> x > 0)`
`keyBy`	按key分组	`stream.keyBy(event -> event.id)`
`window`	开窗聚合	`stream.window(TumblingEventTimeWindows.of(Time.minutes(5)))`
`process`	底层处理（可访问状态）	`stream.process(new MyProcessFunction())`

你想想看，这些算子组合起来，能解决90%的实时计算场景。我做过一个实时风控系统，核心逻辑就是 keyBy + window + process 三板斧。

3.4 Flink Table & SQL API入门

Table API和SQL API是Flink的高级API。它们把流数据抽象成动态表，你可以用SQL直接查询。这对数据分析师特别友好——不用写Java代码，一条SQL搞定。

3.4.1 核心概念：动态表

动态表是什么？说白了，就是一张会随时间变化的表。每来一条数据，表就更新一次。Flink把流转换成动态表，然后执行SQL，最后把结果再转回流。

3.4.2 实战：用SQL做实时聚合

假设我们有一个订单流，想实时统计每个品类的销售额。用SQL写，代码量少得惊人。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SQLExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 从Kafka读取数据，注册为动态表
        tableEnv.executeSql(
            "CREATE TABLE orders (" +
            "  order_id STRING," +
            "  category STRING," +
            "  amount DOUBLE," +
            "  ts TIMESTAMP(3) METADATA FROM 'timestamp'" +
            ") WITH (" +
            "  'connector' = 'kafka'," +
            "  'topic' = 'orders'," +
            "  'properties.bootstrap.servers' = 'localhost:9092'," +
            "  'format' = 'json'" +
            ")"
        );

        // 2. 执行SQL查询：实时聚合
        tableEnv.executeSql(
            "SELECT " +
            "  category," +
            "  SUM(amount) AS total_sales," +
            "  COUNT(*) AS order_count," +
            "  TUMBLE_END(ts, INTERVAL '1' MINUTE) AS window_end " +
            "FROM orders " +
            "GROUP BY category, TUMBLE(ts, INTERVAL '1' MINUTE)"
        ).print();
    }
}

我的经验：Table API适合快速验证和简单聚合。但如果你需要复杂的状态管理（比如自定义超时逻辑），还是得用DataStream API的ProcessFunction。我一般这样分工：SQL做报表，DataStream做风控。

3.4.3 DataStream与Table互转

实际项目中，经常需要在两种API之间切换。Flink提供了toDataStream和fromDataStream方法。

// DataStream → Table
DataStream<Order> orderStream = ...;
Table orderTable = tableEnv.fromDataStream(orderStream);

// Table → DataStream（追加模式）
DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);

// Table → DataStream（更新模式，带撤回）
DataStream<Row> changelogStream = tableEnv.toChangelogStream(resultTable);

避坑指南：我曾经在项目里直接用 toDataStream 转换一个带聚合的Table，结果发现数据对不上。后来才意识到——聚合结果有更新操作，必须用 toChangelogStream 才能正确处理撤回和更新。记住：有GROUP BY就用Changelog模式。

3.5 本章小结

这一章我们覆盖了Flink的三大块：

架构：JobManager + TaskManager + Slot，API层和运行时层分离
环境搭建：本地模式跑Demo，Docker模式模拟集群
API入门：DataStream API做流式处理，Table/SQL API做声明式查询

我个人建议你先把DataStream API练熟，再接触SQL。因为SQL虽然方便，但出了问题你很难排查——你不知道Flink在背后生成了什么执行计划。先理解底层，再用高层，这是我一直以来的学习路径。