📚 课程目录第4章 Flink 基础概念

4. Flink 基础概念：DataStream、算子、时间语义、状态与容错

好，咱们进入 Flink 的核心地带。说实话，Flink 能成为实时计算的主流选择，靠的就是这几个概念：DataStream、算子、时间语义、状态和容错。你把这几个东西吃透了，后面搭特征平台就会顺手很多。

我个人习惯把 Flink 想象成一条「数据流水线」。数据从源头进来，经过各种加工环节，最后产出结果。每个环节都有它的讲究。

4.1 DataStream：一切皆流

在 Flink 里，DataStream 就是你的数据载体。它代表一个持续不断、可能有界也可能无界的数据流。说白了，你从 Kafka 读进来的每一条消息，都会被包装成一个 DataStream 里的元素。

我记得刚开始用 Flink 时，总把它跟 Spark 的 RDD 对比。区别在哪？RDD 是静态数据集，而 DataStream 是动态的——数据在「流」动，你没法一次性拿到全量数据。这个思维转变很重要。

看个最简单的例子：

// 从 Socket 读取数据流
DataStream<String> textStream = env.socketTextStream("localhost", 9999);

// 从 Kafka 读取（生产环境常用）
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");
DataStream<String> kafkaStream = env.addSource(
    new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), props)
);

我的习惯： 开发阶段用 socketTextStream 调试很方便，但上线前一定要切到 Kafka 或 Pulsar 这类可靠的消息队列。Socket 断连了不会自动恢复，踩过这个坑。

4.2 算子：数据的加工厂

算子就是你对 DataStream 做的各种操作。Flink 提供了丰富的算子，我按用途把它们分成三类：

类别	常用算子	说明
基础转换	`map`, `flatMap`, `filter`	一对一、一对多、过滤，最常用
分组聚合	`keyBy`, `sum`, `reduce`, `aggregate`	按 key 分组后做聚合计算
多流操作	`union`, `connect`, `coMap`	合并或关联多个流

这里有个关键点：keyBy 之后，数据会被分到不同的并行子任务里。你想想看，如果 key 分布不均匀，就会出现「数据倾斜」——某个子任务忙死，其他闲死。我在做用户行为特征提取时，就遇到过某个热门用户 ID 把单节点打爆的情况。

避坑指南： 我曾经在 keyBy 之后直接用了 process 算子，结果状态越积越多，内存爆了。后来加了 TTL（Time-To-Live）配置才解决。记住：有状态算子一定要考虑状态过期策略。

4.3 时间语义：Event Time 才是王道

时间语义是 Flink 最核心也最容易搞混的概念。Flink 定义了三种时间：

Event Time：事件发生的时间，通常由业务数据携带。比如用户点击按钮的 timestamp。
Ingestion Time：数据进入 Flink 的时间。
Processing Time：算子处理数据的当前系统时间。

做实时特征平台，我强烈建议你用 Event Time。为什么？因为 Processing Time 不稳定——网络延迟、背压、重启都会导致结果不一致。举个例子，你统计「过去 5 分钟的用户点击量」，如果用 Processing Time，某次 Flink 重启后数据延迟到达，统计结果就全乱了。

启用 Event Time 很简单：

// 设置时间特性为 Event Time
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// 提取事件时间戳并设置水位线
DataStream<MyEvent> withTimestamps = stream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
    );

重点： 水位线（Watermark）是 Flink 处理乱序数据的核心机制。它告诉系统「在这个时间戳之前的数据应该都到了」。我一般设置 5-10 秒的乱序容忍度，具体看业务容忍度。

4.4 状态：Flink 的记忆力

状态是 Flink 区别于普通流处理引擎的关键。没有状态，你只能做无状态转换（比如 map、filter）。有了状态，你才能做聚合、窗口、模式匹配这些高级操作。

Flink 的状态分为两种：

Operator State：算子级别的状态，每个并行子任务共享。适合做 Source/Sink 的偏移量记录。
Keyed State：按 key 划分的状态，每个 key 有自己的状态副本。最常用。

Keyed State 又有几种类型：

状态类型	用途	示例
`ValueState`	保存单个值	用户最近一次登录时间
`ListState`	保存列表	用户最近 10 次点击记录
`MapState`	保存键值对	用户各商品浏览次数
`ReducingState`	增量聚合	累计求和

写个实际例子，统计每个用户的点击次数：

DataStream<Tuple2<String, Long>> clickCounts = clicks
    .keyBy(event -> event.getUserId())
    .process(new KeyedProcessFunction<String, ClickEvent, Tuple2<String, Long>>() {
        private ValueState<Long> countState;

        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Long> descriptor =
                new ValueStateDescriptor<>("clickCount", Long.class);
            // 设置状态 TTL，避免无限增长
            StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(24))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .build();
            descriptor.enableTimeToLive(ttlConfig);
            countState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void processElement(ClickEvent event, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
            Long current = countState.value();
            if (current == null) current = 0L;
            countState.update(current + 1);
            out.collect(Tuple2.of(event.getUserId(), current + 1));
        }
    });

个人经验： 状态后端我推荐用 RocksDB，特别是状态量大的场景。Heap 状态虽然快，但容易 OOM。RocksDB 把状态存到磁盘，内存只做缓存，稳得多。不过要注意，RocksDB 的读写性能比 Heap 慢一个数量级，需要调优。

4.5 容错：Checkpoint 是救命稻草

实时作业最怕什么？挂了之后数据对不上。Flink 的 Checkpoint 机制就是为了解决这个问题。它定期保存作业的状态快照，一旦失败，就从最近的 Checkpoint 恢复。

Checkpoint 的核心原理是 Chandy-Lamport 分布式快照算法。听起来高大上，其实思路很简单：

JobManager 向所有 Source 算子发送「开始快照」的 barrier。
barrier 随着数据流往下游传播，每个算子收到 barrier 后保存自己的状态。
所有算子都保存成功后，这个 Checkpoint 才算完成。

配置 Checkpoint 也很直接：

// 开启 Checkpoint，每 10 秒一次
env.enableCheckpointing(10000);

// 高级配置
CheckpointConfig config = env.getCheckpointConfig();
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
config.setMinPauseBetweenCheckpoints(5000);  // 两个 Checkpoint 之间最少间隔
config.setCheckpointTimeout(60000);           // 超时时间
config.setMaxConcurrentCheckpoints(1);        // 同时只能有一个 Checkpoint 在进行
config.enableExternalizedCheckpoints(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

我曾经踩过的坑： 有次把 Checkpoint 间隔设成了 1 秒，结果频繁的快照把磁盘 IO 打满了，作业反而变慢。后来调到 30 秒一次，配合增量 Checkpoint，才稳定下来。记住：Checkpoint 不是越频繁越好，要权衡恢复时间和性能开销。

4.6 小结：这些概念怎么串起来？

咱们捋一下：DataStream 是数据载体，算子是对数据的操作，时间语义决定了你怎么看待数据发生的时刻，状态让你记住中间结果，Checkpoint 保证挂了能恢复。这五个概念，就是 Flink 的骨架。

在后面的章节里，我们会用这些概念去搭建实时特征平台。比如用 keyBy 按用户分组，用 ValueState 保存用户画像，用 Event Time 保证特征的时间准确性，用 Checkpoint 保证 7x24 小时稳定运行。每一步都离不开今天讲的这些基础。