第4章 Flink DataStream API：DataStream基础、Source与Sink、Transformation算子详解

好，咱们进入Flink DataStream API的核心地带。说实话，很多初学者一上来就被各种算子搞晕了。我当年也是，看着map、flatMap、filter这些名字觉得简单，真到写业务逻辑时才发现坑不少。这一章，咱们就把DataStream的底裤扒清楚。

4.1 DataStream基础概念

DataStream是什么？说白了，它就是Flink里表示无限数据流的核心抽象。你从Kafka读进来的每一条消息，Socket传来的每一行日志，都会被封装成DataStream里的一个元素。

我习惯把DataStream想象成一条传送带。数据像零件一样源源不断流过来，我们在传送带旁边装各种机械臂（算子），对零件做加工、筛选、分流。这个比喻虽然糙，但理不糙。

DataStream有三种基本类型：

• DataStream[T]：最通用的流，元素类型为T
• KeyedStream[T, K]：按key分过组的流，后面才能用window、aggregate等操作
• ConnectedStreams[T1, T2]：两个流连接后的结果，用于合并不同结构的数据

嗯，这里要注意：KeyedStream不是凭空产生的，它必须由DataStream调用keyBy()得到。我见过有人直接在DataStream上调用window，结果编译报错——因为window只认KeyedStream。

4.2 Source：数据从哪来

Source就是数据源。Flink内置了丰富的Source实现，但实际生产中，90%的场景只用两种：Kafka Source和自定义Source。

Source类型	适用场景	并行度	注意事项
Socket Source	测试、调试	1	生产环境别用
Kafka Source	生产主流	可配置	注意offset提交策略
File Source	批处理、回溯	可配置	监控目录变化需配置
自定义Source	特殊协议	自己实现	记得实现checkpoint

来个Kafka Source的代码示例，这是最常用的写法：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置Kafka参数
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "flink-group");

// 创建Kafka Source
DataStream<String> stream = env
    .addSource(new FlinkKafkaConsumer<>(
        "input-topic",
        new SimpleStringSchema(),
        kafkaProps
    ));

自定义Source呢？我建议你实现SourceFunction接口，但要注意：必须实现cancel()方法，否则Flink无法正常停止你的Source。我之前有个项目，自定义Source没写cancel逻辑，每次重启都要kill进程，惨痛教训。

4.3 Sink：数据往哪去

Sink是数据流的终点。数据经过一系列变换后，最终要落到某个存储系统里。Flink的Sink设计很有意思——它不保证精确一次语义，除非你用了两阶段提交协议。

常用的Sink包括：

• Kafka Sink：写回Kafka，常用于流式ETL
• JDBC Sink：写入关系型数据库
• File Sink：写入HDFS/S3，常用于批处理
• 自定义Sink：实现SinkFunction

写个Kafka Sink的例子：

stream
    .addSink(new FlinkKafkaProducer<>(
        "localhost:9092",
        "output-topic",
        new SimpleStringSchema()
    ));

4.4 Transformation算子详解

算子才是DataStream的灵魂。Flink提供了几十种算子，但核心的其实就几个。咱们一个一个过。

4.4.1 map：一对一转换

map是最简单的算子。输入一个元素，输出一个元素。类型可以变，但数量不变。

DataStream<String> input = ...;
DataStream<Integer> output = input
    .map(new MapFunction<String, Integer>() {
        @Override
        public Integer map(String value) {
            return value.length();
        }
    });

你想想看，map适合什么场景？数据清洗、格式转换、字段提取。我有个项目用map把JSON字符串转成Java对象，一行代码搞定。

4.4.2 flatMap：一对多转换

flatMap和map的区别在于：一个输入可以产生0个、1个或多个输出。它用Collector来收集输出结果。

DataStream<String> lines = ...;
DataStream<String> words = lines
    .flatMap(new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String line, Collector<String> out) {
            for (String word : line.split(" ")) {
                out.collect(word);
            }
        }
    });

这个例子很经典——单词拆分。flatMap把一行文本拆成多个单词。注意看，out.collect()可以调用多次，也可以不调用。如果一行是空字符串，那就什么都不输出。

4.4.3 filter：过滤算子

filter就是做条件判断。返回true保留，false丢弃。逻辑简单，但用好了能省不少资源。

DataStream<Integer> numbers = ...;
DataStream<Integer> evenNumbers = numbers
    .filter(new FilterFunction<Integer>() {
        @Override
        public boolean filter(Integer value) {
            return value % 2 == 0;
        }
    });

filter有个容易被忽略的点：它不会改变数据类型。输入是Integer，输出还是Integer。只是数量变少了。

4.4.4 keyBy：分组算子

keyBy是DataStream API里最重要的算子之一。它把数据按key分组，相同key的数据会被分到同一个分区。

DataStream<Event> events = ...;
KeyedStream<Event, String> keyed = events
    .keyBy(new KeySelector<Event, String>() {
        @Override
        public String getKey(Event event) {
            return event.getUserId();
        }
    });

keyBy之后，你才能做窗口聚合、状态操作。为什么？因为Flink需要知道哪些数据属于同一个分组，才能做增量计算。

4.5 知识体系总览

为了让你更直观地理解DataStream API的整体结构，我画了一张图。这张图把Source、Transformation、Sink的串联关系展示得很清楚。

4.6 算子链与性能

最后聊个实战话题：算子链。Flink默认会把多个算子合并到一个Task里执行，减少网络开销。但有时候，你希望强制断开算子链，比如某个算子特别耗CPU，你想单独给它分配资源。

断开算子链的方法：

stream
    .map(...)
    .disableChaining()  // 断开算子链
    .filter(...)
    .startNewChain()    // 开启新链
    .keyBy(...)
    .window(...);