Flink - DataStream API - 源算子

2. 源算子（Source）

Flink 执行过程中，数据来源称为数据源（data source），读取数据的算子称为源算子（source operator），1.12之前的版本添加 source 的方式为：

DataStream<String> stream = env.addSource(xxx);

入参称为源函数，需实现 SourceFunction 接口，1.12开始使用流批统一的 source 架构：

DataStreamSource<String> stream = env.fromSource(xxx);

2.1 准备环境

package com.sw.flinkwordcountdemo.pojo;

import java.io.Serializable;
import java.util.Objects;

/**
 * @author：suaxi
 * @date：2026/11/7 21:05
 */
public class User implements Serializable {

    public String id;
    public String username;
    public String classNo;

    public User() {
    }

    public User(String id, String username, String classNo) {
        this.id = id;
        this.username = username;
        this.classNo = classNo;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getClassNo() {
        return classNo;
    }

    public void setClassNo(String classNo) {
        this.classNo = classNo;
    }

    @Override
    public String toString() {
        return "User{" +
                "id='" + id + '\'' +
                ", username='" + username + '\'' +
                ", classNo='" + classNo + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (o == null || getClass() != o.getClass()) return false;
        User user = (User) o;
        return Objects.equals(id, user.id) && Objects.equals(username, user.username) && Objects.equals(classNo, user.classNo);
    }

    @Override
    public int hashCode() {
        return Objects.hash(id, username, classNo);
    }
}

Flink 会把这样的类作为一种特殊的 POJO 数据类型来对待，方便数据的解析和序列化

2.2 从集合中读取数据

package com.sw.flinkwordcountdemo.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * @author：suaxi
 * @date：2026/11/7 21:50
 */
public class FromCollection {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<Integer> data = List.of(1, 2, 3, 4, 5);
        DataStreamSource<Integer> ds = env.fromCollection(data);
        ds.print();
        env.execute();

        // 8> 3
        // 9> 4
        // 6> 1
        // 7> 2
        // 10> 5

    }
}

2.3 从文件读取数据

pom 依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>

package com.sw.flinkwordcountdemo.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author：suaxi
 * @date：2026/11/7 21:56
 */
public class FromFile {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FileSource<String> data = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/wordCount.txt")).build();
        env.fromSource(data, WatermarkStrategy.noWatermarks(), "file").print();
        env.execute();

        // 5> sun xiao chuan
        // 5> yao shui ge
        // 5> liu bo
        // 5> Giao ge
    }
}

2.4 从 Socket 读取数据

DataStream<String> stream = env.socketTextStream("localhost", 8088);

该方式吞吐量较小，稳定性较差，一般常用于测试场景

2.5 消息队列读取数据

以 RabbitMQ 为例

pom 依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq</artifactId>
    <version>${flink.version}</version>
</dependency>

package com.sw.flinkwordcountdemo.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * @author：suaxi
 * @date：2026/11/7 22:03
 */
public class FromRabbitMQ {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        RMQConnectionConfig rmqConnectionConfig = new RMQConnectionConfig.Builder()
                .setHost("localhost")
                .setPort(5672)
                .build();
        DataStreamSource<?> ds = env.addSource(
                new RMQSource<>(
                        rmqConnectionConfig,
                        "FROM-QMQP-DEMO",
                        true,
                        new SimpleStringSchema()
                )).setParallelism(1);
        ds.print();
        env.execute();
    }
}

2.6 从数据生成器读取数据

主要用于流任务、性能测试等场景

pom 依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-datagen</artifactId>
    <version>${flink.version}</version>
</dependency>

package com.sw.flinkwordcountdemo.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author：suaxi
 * @date：2026/11/7 22:54
 */
public class FromDataGenerator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataGeneratorSource<String> data = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) value -> "index-" + value,
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(2),
                Types.STRING
        );
        env.fromSource(data, WatermarkStrategy.noWatermarks(), "data-generator").print();
        env.execute();
    }
}

2.7 Flink 数据类型

Flink 使用“类型信息“（TypeInformation）来统一表示数据类型，TypeInformation 是 Flink 中所有类型描述符的基类

（1）支持的数据类型：

• 基本类型：所有 Java 基本类型和包装类，还有 Void、String、Date、BigDecimal、BigInteger
• 数组：基本类型数组、对象数组
• 复合数据
• TUPLE 元组：Flink 内置的元组类型，从 Tuple0 - Tuple25，最多25个，不支持空字段
• Scala 样例类及 Scala 元组：不支持空字段
• 行类型（ROW）：具有任意个字段的元组，可以为空
• POJO：Flink 自定义的类似于 Java Bean 的类
• 辅助类型：Option、Either、List、Map

• GENERIC 泛型：

  • 如果未遵循 Flink POJO 类型的规范，会被当作泛型处理（存在类型擦除问题），且由 Kryo 序列化

  • POJO 规范：

    • 公共类
    • 有无参构造
    • 所有属性声明为 public
    • 所有属性的类型都是可以序列化的

（2）Type Hints 类型提示

Flink 拥有类型提取系统，可以分析函数的输入、返回类型，自动获取类型信息，从而获取对应的序列化/反序列化器，当遇到 Java 类型擦除时，Flink 提供了以下方法显示声明返回类型：

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

.map(word -> Tuple2.of(word, 1L))
.returns(new TypeHint<Tuple2<Integer, SomeType>>(){})