3. Flink DataStream API基础：Source、Transformation、Sink，以及一个简单的WordCount实战

好，咱们今天来啃Flink DataStream API这块硬骨头。说实话，很多初学者一上来就被各种概念搞懵了——Source是什么？Transformation又是什么？Sink又往哪写？

别急，我带你捋一遍。说白了，Flink的DataStream API就是一套流水线工具：你把数据从某个地方读进来（Source），然后对它做各种加工（Transformation），最后把结果写到某个地方去（Sink）。就这么简单。

核心三要素：Source（数据源）→ Transformation（转换算子）→ Sink（数据输出）

3.1 Source：数据从哪来？

Source就是数据的入口。我在做量化交易系统时，最常见的Source就是Kafka——实时行情数据、订单数据全从Kafka Topic里流进来。但Flink支持的Source远不止Kafka。

常用的Source类型：

  • 基于集合的Source：适合本地测试和调试，比如 fromCollection()、fromElements()
  • 基于文件的Source：readTextFile()，读取HDFS或本地文件系统
  • 基于Socket的Source：socketTextStream()，网络端口数据流
  • 自定义Source：实现 SourceFunction 接口，比如对接行情SDK

嗯，这里要注意：生产环境千万别用Socket Source。我曾经见过一个团队，把Socket Source直接部署到线上，结果网络抖动一次，整个任务就挂了。生产环境老老实实用Kafka或者Pulsar。

来个最简单的例子：

// 从本地集合创建DataStream
DataStream<String> textStream = env.fromElements(
    "flink hadoop spark",
    "flink flink spark",
    "hadoop spark flink"
);

你看，三行数据，就是一个最简单的数据源。当然，真实场景下，你可能会这样写：

// 从Kafka读取实时行情
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "quant-trade-group");

DataStream<String> stream = env.addSource(
    new FlinkKafkaConsumer<>("trade-topic", 
        new SimpleStringSchema(), props)
);

3.2 Transformation：数据怎么加工？

Transformation是整个Flink的灵魂。说白了，就是你对数据流做的各种操作——过滤、转换、分组、聚合等等。

我常用的Transformation算子：

算子 作用 我在量化项目中的用法
map 一对一转换 将行情JSON解析成Java对象
flatMap 一对多转换 将一条订单拆成多条成交记录
filter 过滤数据 只保留价格大于100元的股票
keyBy 按key分组 按股票代码分组计算均线
sum/reduce 聚合计算 计算每分钟成交量

举个例子，假设我们要过滤出价格大于100的股票，然后按股票代码分组：

DataStream<TradeRecord> filteredStream = tradeStream
    .filter(trade -> trade.getPrice() > 100.0)
    .keyBy(TradeRecord::getStockCode);

你想想看，如果不用Flink，你要自己写多线程、自己管理状态、自己处理数据倾斜……那得多痛苦？Flink把这些脏活累活全包了。

小技巧：在开发阶段，我习惯在Transformation链路上加一些 .print() 来调试，看看数据流中间结果对不对。等上线前再删掉。

3.3 Sink：结果写哪去？

Sink就是数据的出口。数据加工完了，总得有个去处吧？写到数据库、写到文件、写到消息队列，都行。

常见的Sink：

  • print() / printToErr()：控制台输出，调试用
  • writeAsText()：写入文件
  • addSink()：自定义Sink，比如写入MySQL、Redis、Elasticsearch
  • Flink自带的 JdbcSink、RedisSink 等

我曾经踩过一个坑：直接用 writeAsText() 往HDFS写数据，结果并行度一高，文件碎片多得吓人。后来改用 StreamingFileSink 配合滚动策略，才解决了这个问题。

来个写入MySQL的例子：

// 将结果写入MySQL
stream.addSink(JdbcSink.sink(
    "INSERT INTO word_count (word, cnt) VALUES (?, ?)",
    (statement, record) -> {
        statement.setString(1, record.getWord());
        statement.setLong(2, record.getCount());
    },
    JdbcExecutionOptions.builder()
        .withBatchSize(1000)
        .build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/quant")
        .withDriverName("com.mysql.cj.jdbc.Driver")
        .withUsername("root")
        .withPassword("123456")
        .build()
));

警告：生产环境一定要设置 withBatchSize 和 withBatchInterval，否则每条数据都写一次数据库，性能会非常差。

3.4 WordCount实战：从零到一跑起来

好了，理论说完了，咱们来写一个完整的WordCount。你别觉得WordCount太简单——它麻雀虽小五脏俱全，Source、Transformation、Sink全都有。

先画个流程图，让你对整个流程有个直观认识：

WordCount数据流处理流程 Source fromElements() flatMap 拆分单词 map 转为(word, 1) keyBy sum Sink (print) 数据从Source流入，经过flatMap→map→keyBy+sum，最终输出到Sink

好，现在上代码。这是完整的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 WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. Source：读取数据
        DataStream<String> textStream = env.fromElements(
            "flink hadoop spark",
            "flink flink spark",
            "hadoop spark flink"
        );
        
        // 3. Transformation：数据处理
        DataStream<Tuple2<String, Integer>> wordCountStream = textStream
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                    // 按空格拆分单词
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            .keyBy(value -> value.f0)  // 按单词分组
            .sum(1);                    // 对第二个字段求和
        
        // 4. Sink：输出结果
        wordCountStream.print();
        
        // 5. 执行任务
        env.execute("Flink WordCount");
    }
}

运行结果会是这样：

(flink,1)
(hadoop,1)
(spark,1)
(flink,2)
(spark,2)
(hadoop,2)
(flink,3)

你可能会问：为什么结果不是一次性输出？因为Flink是流式处理，来一条数据就处理一条。所以你会看到每个单词的计数在逐步累加。

个人经验：我在做量化交易系统的回测模块时，就是拿WordCount这个模板改的——把单词换成股票代码，把计数换成成交量。你看，一通百通。

3.5 避坑指南

最后，分享几个我踩过的坑：

  • 并行度设置不当：WordCount默认并行度等于CPU核数。如果你在本地跑，数据量小，并行度太高反而浪费资源。我习惯在开发环境设成1，生产环境再根据数据量调整。
  • 忘记调用 execute()：很多新手写完代码发现没输出，就是因为忘了这行。Flink是懒加载，不调用 execute() 啥都不会执行。
  • Tuple2的泛型问题：Java的泛型擦除会导致类型信息丢失。如果你用 keyBy(0) 这种写法，Flink可能不知道key的类型。我建议用Lambda表达式或者 TypeInformation 来明确指定。

好了，WordCount就讲到这里。你把这个跑通了，Flink DataStream API的基础就算入门了。下一节咱们会深入讲状态管理和Checkpoint，那才是Flink真正牛逼的地方。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →