2. Flink编程模型与API：DataStream API基础、Source/Sink/Transformation算子详解、Flink程序骨架与执行流程

好，咱们进入第二章。这一章我打算把Flink的编程模型彻底讲透。你可能会觉得，不就是几个API嘛，文档翻翻就会了。但说实话，我在生产环境里踩过的坑，十有八九都是因为对这几个基础概念理解得不够深。咱们今天就把地基打牢。

2.1 Flink程序骨架：从零到一

先看一个最简的Flink程序长什么样。我习惯把它拆成四个固定步骤，你写任何Flink作业都逃不出这个框架。

Flink程序四步走：
  1. 获取执行环境 —— 创建StreamExecutionEnvironment
  2. 添加数据源 —— 调用addSource或从集合/文件读取
  3. 定义转换逻辑 —— 各种Transformation算子
  4. 指定输出 —— 调用addSink或print()

嗯，这里有个细节：execute() 是触发作业执行的开关。没有它，前面的所有定义都只是构建了一个DAG，不会真正跑起来。我曾经见过新手同事调试了半天，发现代码没报错但就是不输出数据，最后发现是忘了调 execute()。

// 最简Flink程序骨架
public class FlinkSkeleton {
    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(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);
        
        // 4. 指定输出
        counts.print();
        
        // 触发执行
        env.execute("Socket WordCount");
    }
}

2.2 DataStream API核心概念

DataStream API是Flink的"面子"，底层是StreamTask在干活。说白了，你写的每个算子最终都会被翻译成一个个的OperatorChain，在TaskManager上跑。

我个人习惯把DataStream理解成一个"无限流动的集合"。它跟Java的Stream API有点像，但区别大了去了——Flink的DataStream是有状态的，而且天然支持事件时间和水位线。

概念 说明 我踩过的坑
DataStream 未分区的数据流，并行度为1 默认并行度可能跟你想的不一样
KeyedStream 按key分区的数据流，相同key进入同一分区 keyBy后状态才能按key隔离
WindowedStream 在KeyedStream上开窗后的流 窗口类型选错会导致数据丢失
ConnectedStreams 连接两个流，共享状态 连接后要小心处理两个流的类型

2.3 Source算子详解：数据从哪来

Source是Flink作业的入口。Flink内置了丰富的Source实现，但生产环境里，我90%的场景都在用Kafka Source。

我的建议： 开发阶段用 fromElements 或 socketTextStream 快速验证逻辑；上线前一定要换成Kafka或Pulsar这类有回放能力的Source。为什么？因为一旦作业失败重启，没有数据回放能力，你可能会丢数据。

常用的Source类型：

  • 基于集合：fromCollection、fromElements —— 适合单元测试
  • 基于文件：readTextFile —— 适合批处理或回溯历史数据
  • 基于Socket：socketTextStream —— 快速原型验证
  • 基于消息队列：FlinkKafkaConsumer —— 生产首选
  • 自定义Source：实现 SourceFunction 接口

说到自定义Source，我记得有一次需要从自研的MQ里消费数据，当时图省事直接继承 RichSourceFunction，结果没处理好checkpoint的语义，导致重启后数据重复消费。后来老老实实实现了 CheckpointedFunction 接口，才把问题解决。

2.4 Transformation算子：数据的变形金刚

Transformation是Flink最核心的部分。说白了，就是把输入的数据流变成你想要的样子。我按使用频率给你排个序。

2.4.1 单流基础算子

算子 作用 并行度变化
map 一对一转换，输入一条输出一条 不变
flatMap 一对多转换，输入一条输出零到多条 不变
filter 过滤，保留满足条件的元素 可能减少
keyBy 按key分区，逻辑上将流分成多个分区 重新分区

你想想看，map 和 flatMap 的区别在哪？map 必须返回一个元素，而 flatMap 可以返回0个、1个或多个。我在做日志解析时经常用 flatMap，因为有些日志行可能格式不对，直接过滤掉；有些日志行又可能包含多条事件，需要拆开。

2.4.2 多流合并算子

实际项目中，很少只有一个数据源。我做过一个风控系统，同时消费用户行为流、规则配置流和模型特征流，这时候就需要合并算子。

  • union：合并多个同类型流，保留所有数据
  • connect：连接两个不同类型流，可以共享状态
  • coMap/coFlatMap：对ConnectedStreams进行操作
注意： union 要求所有流的数据类型必须一致；connect 则允许不同类型，但处理起来更复杂。我曾经在项目里用 connect 把订单流和支付流关联起来，两个流类型不同，但通过共享状态实现了实时对账。

2.5 Sink算子：数据去哪了

Sink是Flink作业的出口。数据经过一系列Transformation后，最终要落到某个地方。我见过不少同学在Source和Transformation上花了很多功夫，结果Sink随便写写，最后数据写不进去或者写错了都不知道。

常用的Sink：

  • print()：调试用，输出到标准输出
  • writeAsText：写入文件，生产环境慎用
  • FlinkKafkaProducer：写入Kafka，生产环境常用
  • JDBCSink：写入关系型数据库
  • 自定义Sink：实现SinkFunction接口
避坑指南： 我曾经用JDBCSink往MySQL写数据，没做批量提交，结果每秒几万条数据直接把数据库连接池打满了。后来改成了 JdbcExecutionOptions.builder().withBatchSize(1000).build()，问题才解决。记住，Sink一定要考虑背压和事务语义。

2.6 执行流程：从代码到运行

咱们把整个流程串起来看看。你写的Flink程序，到底是怎么变成分布式作业的？

Flink程序执行流程 用户代码（DataStream API） StreamGraph（逻辑DAG） JobGraph（优化后DAG） ExecutionGraph（并行化） TaskManager物理执行 关键说明 • StreamGraph：你写的代码 • JobGraph：优化合并算子链 • ExecutionGraph：分配并行度 我的经验 调试时看JobGraph就够了 性能问题看ExecutionGraph 物理执行看Flink UI

这个流程你得多看几遍。我刚开始学Flink时，总觉得代码写完就完事了，结果作业跑起来跟预期完全不一样。后来才明白，从StreamGraph到ExecutionGraph这一步，Flink做了很多优化——比如算子链合并、分区策略调整。你写的每个算子，最终可能被合并到同一个Task里执行。

2.7 实战：写一个完整的Flink作业

咱们把今天学的串起来，写一个完整的实时特征计算作业。这个作业会从Kafka读取用户行为数据，计算每个用户最近5分钟的点击次数，然后写入MySQL。

public class RealTimeFeatureJob {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        // 2. 添加Kafka Source
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "feature-group");
        
        DataStream<String> source = env.addSource(
            new FlinkKafkaConsumer<>("user_click", 
                new SimpleStringSchema(), props));
        
        // 3. 转换逻辑
        DataStream<UserClick> clicks = source
            .map(line -> parseClick(line))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserClick>forBoundedOutOfOrderness(
                    Duration.ofSeconds(5))
                    .withTimestampAssigner((event, ts) -> event.getTimestamp()));
        
        DataStream<FeatureResult> features = clicks
            .keyBy(UserClick::getUserId)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new ClickCountAggregate())
            .map(count -> new FeatureResult(count.userId, 
                "click_count_5min", count.count));
        
        // 4. 写入MySQL Sink
        features.addSink(JdbcSink.sink(
            "INSERT INTO user_features(user_id, feature_name, feature_value) VALUES(?, ?, ?)",
            (ps, value) -> {
                ps.setString(1, value.userId);
                ps.setString(2, value.featureName);
                ps.setLong(3, value.featureValue);
            },
            JdbcExecutionOptions.builder().withBatchSize(1000).build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:mysql://localhost:3306/features")
                .withDriverName("com.mysql.cj.jdbc.Driver")
                .withUsername("root")
                .withPassword("password")
                .build()
        ));
        
        // 5. 触发执行
        env.execute("RealTimeFeatureJob");
    }
}
注意： 这个例子里的 assignTimestampsAndWatermarks 是事件时间处理的关键。如果不设置，Flink默认使用处理时间，那窗口计算的结果就不准确了。我见过有人把事件时间窗口开好了，结果忘了设置水位线，窗口永远不触发——因为水位线默认是 Long.MIN_VALUE。
← 上一章 📖 返回目录 下一章 →