【Flink】Flink数据源Source详解

Flink数据源Source详解：内置与自定义Connector

前言

上一篇我们学习了 Flink 的编程模型，知道了一个 Flink 程序由 Source → Transformation → Sink 三部分组成。这篇文章我们来深挖 Source——数据从哪来？

Source 是整个数据管道的起点，它决定了你能处理什么数据。Flink 内置了丰富的 Source，也支持自定义。这篇文章会带你全面了解 Flink 的数据源体系。

🏠个人主页：你的主页

目录

• 一、Source概述
• 二、新版Source架构
• 三、内置Source详解
• 四、Kafka Source实战
• 五、File Source实战
• 六、自定义Source
• 七、Source的并行度与分区
• 八、总结

一、Source概述

1.1 什么是Source？

Source（数据源） 是 Flink 程序的数据入口，负责从外部系统读取数据并转换为 Flink 内部的 DataStream。

┌─────────────────────────────────────────────────────────────────────┐
│ 数据流向 │
│ │
│ 外部系统 Flink 外部系统 │
│ ┌──────────┐ ┌──────────────┐ ┌──────────┐ │
│ │ Kafka │ ───────→ │ │ ───────→ │ MySQL │ │
│ │ File │ │ DataStream │ │ Redis │ │
│ │ MySQL │ Source │ 处理逻辑 │ Sink │ ES │ │
│ │ Socket │ ───────→ │ │ ───────→ │ Kafka │ │
│ └──────────┘ └──────────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘

1.2 Source的分类

按数据边界分：

按内置/自定义分：

1.3 新老API对比

Flink 1.12 开始推出了新版 Source API（Source 接口），老版（SourceFunction）逐步废弃：

// 老版API（已废弃）
env.addSource(new FlinkKafkaConsumer<>(...));

// 新版API（推荐）
env.fromSource(
KafkaSource.<String>builder().build(),
WatermarkStrategy.noWatermarks(),
"Kafka Source"
);

为什么换新API？ 新版 Source 支持：

• 流批一体
• 更好的 Watermark 集成
• 更清晰的架构

二、新版Source架构

2.1 核心组件

新版 Source 采用 SplitEnumerator + Reader 架构：

┌─────────────────────────────────────────────────────────────────────┐
│ 新版 Source 架构 │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ SplitEnumerator │ │
│ │ (运行在 JobManager) │ │
│ │ │ │
│ │ • 发现数据分片 (Split) │ │
│ │ • 将分片分配给 Reader │ │
│ │ • 处理 Reader 汇报的已完成分片 │ │
│ └───────────────────────────┬─────────────────────────────────┘ │
│ │ 分配 Split │
│ ↓ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ SourceReader │ │ SourceReader │ │ SourceReader │ │
│ │ (TM 1) │ │ (TM 2) │ │ (TM 3) │ │
│ │ │ │ │ │ │ │
│ │ 读取分片数据 │ │ 读取分片数据 │ │ 读取分片数据 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘

2.2 用大白话解释

想象你是一个快递公司老板：

• SplitEnumerator = 调度中心：扫描有哪些快递（Split），分配给快递员
• SourceReader = 快递员：负责实际取件并送到仓库（Flink）
• Split = 快递：每个快递就是一个数据分片

调度中心发现：今天有3个取件点（Split）
│
│ 分配任务
↓
快递员A ← 取件点1
快递员B ← 取件点2
快递员C ← 取件点3
│
│ 取回快递（数据）
↓
送到仓库（Flink DataStream）

三、内置Source详解

3.1 常用Source一览

┌────────────────────────────────────────────────────────────────────┐
│ Flink 内置 Source │
│ │
│ ┌───────────────────────┐ ┌───────────────────────┐ │
│ │ 消息队列 │ │ 文件系统 │ │
│ │ • KafkaSource │ │ • FileSource │ │
│ │ • PulsarSource │ │ • 支持 HDFS/S3/本地 │ │
│ │ • RabbitMQ │ │ │ │
│ └───────────────────────┘ └───────────────────────┘ │
│ │
│ ┌───────────────────────┐ ┌───────────────────────┐ │
│ │ 数据库 │ │ 开发测试 │ │
│ │ • JdbcSource │ │ • fromElements │ │
│ │ • MongoDBSource │ │ • fromCollection │ │
│ │ • CDC Source │ │ • socketTextStream │ │
│ └───────────────────────┘ └───────────────────────┘ │
└────────────────────────────────────────────────────────────────────┘

3.2 开发测试用Source

开发时快速测试用的简单Source：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 1. 从集合创建（最简单）
DataStream<String> fromCollection = env.fromElements("hello", "flink", "world");

// 2. 从Java集合创建
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
DataStream<Integer> fromList = env.fromCollection(numbers);

// 3. 从Socket读取（常用于本地调试）
// 先启动: nc -lk 9999
DataStream<String> fromSocket = env.socketTextStream("localhost", 9999);

// 4. 生成序列（有界）
DataStream<Long> sequence = env.fromSequence(1, 100);

使用场景：

• fromElements/fromCollection：单元测试、快速验证逻辑
• socketTextStream：本地开发调试
• fromSequence：性能测试、生成测试数据

四、Kafka Source实战

Kafka 是最常用的数据源，重点讲解。

4.1 添加依赖

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>3.0.0-1.17</version>
</dependency>

版本对应关系：

4.2 基础用法

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

public class KafkaSourceExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 构建 Kafka Source
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
// Kafka 集群地址
.setBootstrapServers("localhost:9092")
// 订阅的 Topic
.setTopics("input-topic")
// 消费者组ID
.setGroupId("flink-consumer-group")
// 起始消费位置
.setStartingOffsets(OffsetsInitializer.earliest())
// 反序列化器
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();

// 使用 Source
DataStream<String> stream = env.fromSource(
kafkaSource,
WatermarkStrategy.noWatermarks(),
"Kafka Source"
);

stream.print();
env.execute("Kafka Source Demo");
}
}

4.3 消费起始位置配置

// 1. 从最早的数据开始消费（适合首次启动）
.setStartingOffsets(OffsetsInitializer.earliest())

// 2. 从最新的数据开始消费（跳过历史数据）
.setStartingOffsets(OffsetsInitializer.latest())

// 3. 从消费者组的已提交位置开始（默认行为）
.setStartingOffsets(OffsetsInitializer.committedOffsets())

// 4. 如果没有已提交的位置，则从最早开始
.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))

// 5. 从指定时间戳开始
.setStartingOffsets(OffsetsInitializer.timestamp(1609459200000L))

// 6. 从指定的 offset 开始
Map<TopicPartition, Long> offsets = new HashMap<>();
offsets.put(new TopicPartition("topic", 0), 100L);
.setStartingOffsets(OffsetsInitializer.offsets(offsets))

4.4 消费者组与Offset管理

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092")
.setTopics("input-topic")
.setGroupId("my-group")
// 消费语义
.setProperty("enable.auto.commit", "false") // Flink自己管理offset
.setProperty("auto.offset.reset", "earliest")
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();

Offset提交策略：

4.5 订阅多个Topic

// 方式1：指定Topic列表
.setTopics("topic1", "topic2", "topic3")

// 方式2：使用正则表达式
.setTopicPattern("order-.*") // 匹配所有 order- 开头的 Topic

// 方式3：指定Topic和分区
.setPartitions(
new HashSet<>(Arrays.asList(
new TopicPartition("topic1", 0),
new TopicPartition("topic1", 1),
new TopicPartition("topic2", 0)
))
)

4.6 自定义反序列化

处理复杂数据类型（如JSON）：

// 1. 使用 Jackson 反序列化 JSON
public class OrderDeserializer implements DeserializationSchema<Order> {

private final ObjectMapper mapper = new ObjectMapper();

@Override
public Order deserialize(byte[] bytes) throws IOException {
return mapper.readValue(bytes, Order.class);
}

@Override
public boolean isEndOfStream(Order order) {
return false; // 流永不结束
}

@Override
public TypeInformation<Order> getProducedType() {
return TypeInformation.of(Order.class);
}
}

// 使用
KafkaSource<Order> kafkaSource = KafkaSource.<Order>builder()
.setBootstrapServers("localhost:9092")
.setTopics("orders")
.setGroupId("order-consumer")
.setValueOnlyDeserializer(new OrderDeserializer())
.build();

4.7 完整生产配置示例

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
// 基础配置
.setBootstrapServers("kafka1:9092,kafka2:9092,kafka3:9092")
.setTopics("input-topic")
.setGroupId("flink-app-001")

// 起始位置
.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))

// 有界模式（批处理时使用）
// .setBounded(OffsetsInitializer.latest())

// 反序列化
.setValueOnlyDeserializer(new SimpleStringSchema())

// Kafka Consumer 高级配置
.setProperty("session.timeout.ms", "30000")
.setProperty("max.poll.records", "500")
.setProperty("fetch.min.bytes", "1")
.setProperty("fetch.max.wait.ms", "500")

// 安全认证（如需要）
// .setProperty("security.protocol", "SASL_PLAINTEXT")
// .setProperty("sasl.mechanism", "PLAIN")

.build();

// 配置 Watermark（处理事件时间）
WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
.<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withIdleness(Duration.ofMinutes(1)); // 处理空闲分区

DataStream<String> stream = env.fromSource(
kafkaSource,
watermarkStrategy,
"Kafka Source"
);

五、File Source实战

5.1 基础用法

import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;

// 读取单个文件
FileSource<String> fileSource = FileSource
.forRecordStreamFormat(
new TextLineInputFormat(),
new Path("/data/input/data.txt")
)
.build();

// 读取目录下所有文件
FileSource<String> dirSource = FileSource
.forRecordStreamFormat(
new TextLineInputFormat(),
new Path("/data/input/")
)
.build();

DataStream<String> stream = env.fromSource(
fileSource,
WatermarkStrategy.noWatermarks(),
"File Source"
);

5.2 持续监控目录（流模式）

FileSource<String> fileSource = FileSource
.forRecordStreamFormat(
new TextLineInputFormat(),
new Path("/data/input/")
)
// 持续监控新文件
.monitorContinuously(Duration.ofSeconds(10))
.build();

使用场景：日志文件持续追加，需要实时处理新增内容。

5.3 读取CSV文件

// 使用 CsvReaderFormat 读取 CSV
CsvReaderFormat<Order> csvFormat = CsvReaderFormat.forPojo(Order.class);

FileSource<Order> csvSource = FileSource
.forRecordStreamFormat(csvFormat, new Path("/data/orders.csv"))
.build();

5.4 读取不同存储系统

// 本地文件系统
new Path("file:///data/input/")

// HDFS
new Path("hdfs://namenode:8020/data/input/")

// S3
new Path("s3://bucket-name/data/input/")

// OSS（阿里云）
new Path("oss://bucket-name/data/input/")

六、自定义Source

当内置Source无法满足需求时，可以自定义。

6.1 简单方式：实现SourceFunction（已废弃但简单）

/**
* 自定义 Source：生成随机订单
* 注意：SourceFunction 已废弃，仅用于学习理解
*/
public class RandomOrderSource implements SourceFunction<Order> {

private volatile boolean isRunning = true;
private final Random random = new Random();

@Override
public void run(SourceContext<Order> ctx) throws Exception {
while (isRunning) {
// 生成随机订单
Order order = new Order(
"ORDER_" + System.currentTimeMillis(),
"USER_" + random.nextInt(100),
random.nextDouble() * 1000,
System.currentTimeMillis()
);

// 发送数据
ctx.collect(order);

// 控制发送频率
Thread.sleep(100);
}
}

@Override
public void cancel() {
isRunning = false;
}
}

// 使用
env.addSource(new RandomOrderSource());

6.2 新版方式：实现Source接口

新版 Source 需要实现三个组件：

/**
* 自定义数据库 Source（简化版）
*/
public class DatabaseSource implements Source<Record, DatabaseSplit, DatabaseEnumeratorState> {

private final String jdbcUrl;
private final String query;

public DatabaseSource(String jdbcUrl, String query) {
this.jdbcUrl = jdbcUrl;
this.query = query;
}

@Override
public Boundedness getBoundedness() {
return Boundedness.BOUNDED; // 有界数据源
}

@Override
public SplitEnumerator<DatabaseSplit, DatabaseEnumeratorState> createEnumerator(
SplitEnumeratorContext<DatabaseSplit> context) {
return new DatabaseSplitEnumerator(context, jdbcUrl, query);
}

@Override
public SourceReader<Record, DatabaseSplit> createReader(SourceReaderContext context) {
return new DatabaseSourceReader(context, jdbcUrl);
}

// … 其他方法省略（序列化器等）
}

实现复杂度较高，一般情况下建议：

6.3 使用RichSourceFunction（可并行）

/**
* 可并行的自定义 Source
*/
public class ParallelRandomSource extends RichParallelSourceFunction<Integer> {

private volatile boolean isRunning = true;

@Override
public void run(SourceContext<Integer> ctx) throws Exception {
int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
int numSubtasks = getRuntimeContext().getNumberOfParallelSubtasks();

// 每个并行实例生成不同范围的数据
int start = subtaskIndex * 1000;
int end = start + 999;

while (isRunning) {
int value = start + new Random().nextInt(end – start);
ctx.collect(value);
Thread.sleep(100);
}
}

@Override
public void cancel() {
isRunning = false;
}
}

// 使用时可以设置并行度
env.addSource(new ParallelRandomSource()).setParallelism(4);

七、Source的并行度与分区

7.1 并行读取

Source 的并行度决定了同时有多少个 Reader 在读取数据：

┌─────────────────────────────────────────────────────────────────────┐
│ Source 并行读取 │
│ │
│ Kafka Topic: 6个分区 │
│ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │ P0 │ │ P1 │ │ P2 │ │ P3 │ │ P4 │ │ P5 │ │
│ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │
│ │ │ │ │ │ │ │
│ └───────┴───────┼───────┴───────┼───────┘ │
│ │ │ │
│ Source 并行度=3 ↓ ↓ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Reader 0 │ │ Reader 1 │ │ Reader 2 │ │
│ │ 负责 P0, P1 │ │ 负责 P2, P3 │ │ 负责 P4, P5 │ │
│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘

7.2 并行度设置建议

// 设置 Source 并行度
env.fromSource(kafkaSource, watermarkStrategy, "Kafka Source")
.setParallelism(6); // 等于 Kafka 分区数

注意：Source 并行度大于分区数时，多余的实例会空转浪费资源。

7.3 数据分发策略

Source 读取的数据如何分发给下游？

// 1. Forward（默认，Source和下游并行度相同时）
source.map(...); // 数据直接传递，不走网络

// 2. Rebalance（轮询分发）
source.rebalance().map(...); // 均匀分发到所有并行实例

// 3. Shuffle（随机分发）
source.shuffle().map(...); // 随机选择下游实例

// 4. Rescale（局部轮询）
source.rescale().map(...); // 在相邻的并行实例间轮询

八、总结

这篇文章我们深入学习了 Flink 的数据源体系：

核心要点

Kafka Source核心配置

KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092") // 必须
.setTopics("topic") // 必须
.setGroupId("group") // 推荐
.setStartingOffsets(OffsetsInitializer.earliest()) // 起始位置
.setValueOnlyDeserializer(new SimpleStringSchema()) // 反序列化
.build();

Source选择指南

需要从 Kafka 读数据？ → KafkaSource
需要从文件读数据？ → FileSource
开发测试？ → fromElements / socketTextStream
对接私有系统？ → 自定义 Source

下一篇文章，我们将学习 Flink 算子大全，深入了解 map、filter、keyBy 等各种 Transformation 操作。

热门专栏推荐

• Agent小册
• Java基础合集
• Python基础合集
• Go基础合集
• 大数据合集
• 前端小册
• 数据库合集
• Redis 合集
• Spring 全家桶
• 微服务全家桶
• 数据结构与算法合集
• 设计模式小册
• Ai工具小册

等等等还有许多优秀的合集在主页等着大家的光顾，感谢大家的支持

文章到这里就结束了，如果有什么疑问的地方请指出，诸佬们一起来评论区一起讨论😊 希望能和诸佬们一起努力，今后我们一起观看感谢您的阅读🙏 如果帮助到您不妨3连支持一下，创造不易您们的支持是我的动力🌟