Flink DataStream API 常用操作总结
·
Flink DataStream API 提供了丰富且强大的操作符（Operators）来处理无界数据流。下面我将这些常用的 API 分为几大类，并附上说明和代码示例。
1. 数据源（Data Sources）
这是程序的起点，用于从外部系统获取数据。
| API | 说明 | 代码示例 |
|---|---|---|
env.fromElements(...) | 从给定的对象序列创建数据流。 | DataStream<String> stream = env.fromElements("A", "B", "C"); |
env.fromCollection(...) | 从 Java 的 Collection 创建数据流。 | DataStream<Integer> stream = env.fromCollection(Arrays.asList(1, 2, 3)); |
env.readTextFile(path) | 读取文本文件，按行读取。 | DataStream<String> stream = env.readTextFile("file:///path/to/file"); |
env.socketTextStream(...) | 从 Socket 连接读取数据。 | DataStream<String> stream = env.socketTextStream("localhost", 9999); |
env.addSource(new FlinkKafkaConsumer(...)) | 最常用。从 Apache Kafka 中消费数据。需要引入 Kafka 连接器依赖。 | Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), props)); |
env.addSource(new ...) | 自定义源函数。可以实现 SourceFunction 或 RichSourceFunction 来连接任何数据源。 | env.addSource(new MyCustomSource()); |
2. 转换操作（Transformations）
这是 DataStream API 的核心，用于对数据进行各种处理和变换。
2.1 基本转换（单流操作）
| API | 说明 | 代码示例 |
|---|---|---|
map | 一对一转换。输入一个元素，输出一个元素。 | stream.map(x -> x * 2); |
flatMap | 一对多或一对一转换。输入一个元素，可以输出零个、一个或多个元素。 | stream.flatMap((String sentence, Collector<String> out) -> { for (String word : sentence.split(" ")) { out.collect(word); } }); |
filter | 过滤。对每个元素进行判断，保留为 true 的元素。 | stream.filter(x -> x.startsWith("error")); |
keyBy | 关键操作。将流逻辑分区为不相交的分区（每个分区是一个KeyedStream）。具有相同 key 的记录会被分配到同一个分区。 | stream.keyBy(event -> event.getUserId()); 或 stream.keyBy("userId") (已不推荐) |
union | 合并两个或多个类型相同的数据流。 | DataStream<String> mergedStream = stream1.union(stream2, stream3); |
2.2 基于 KeyedStream 的转换
必须先执行 keyBy 操作得到 KeyedStream，才能使用这些操作。
| API | 说明 | 代码示例 |
|---|---|---|
sum, min, max | 滚动聚合。对当前 key 下的流数据进行聚合。 | keyedStream.sum("count"); keyedStream.max("temperature"); |
reduce | 自定义滚动聚合。接受一个 ReduceFunction，将两个值组合成一个相同类型的值。 | keyedStream.reduce((v1, v2) -> new Temperature(v1.id, v1.temp + v2.temp)); |
process | 最灵活的低阶操作。可以访问状态和定时器。 | keyedStream.process(new MyProcessFunction()); |
2.3 多流转换
| API | 说明 | 代码示例 |
|---|---|---|
connect | 连接两个数据流，允许共享状态。连接后的流是 ConnectedStreams。 | ConnectedStreams<String, Integer> connected = stream1.connect(stream2); |
coMap / coFlatMap | 对 ConnectedStreams 进行处理，可以分别处理两个流的元素。 | connected.map(new CoMapFunction<String, Integer, String>() { @Override public String map1(String value) { ... } @Override public String map2(Integer value) { ... } }); |
broadcast | 将一个流广播到所有并行任务，常用于动态配置规则。 | BroadcastStream<Rule> ruleStream = ruleStream.broadcast(ruleStateDescriptor); |
window | 将数据流切分成有限的“桶”（Windows），以便进行窗口计算。 | keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5))); |
3. 数据接收器（Data Sinks）
这是程序的终点，用于将处理后的数据输出到外部系统。
| API | 说明 | 代码示例 |
|---|---|---|
print() | 将每个元素打印到标准输出。非常用于调试。 | stream.print(); |
printToErr() | 将每个元素打印到标准错误输出。 | stream.printToErr(); |
writeAsText(...) | 将流写入文本文件。 | stream.writeAsText("file:///path/to/output"); |
addSink(new StreamFileSink(...)) | 更强大、更可靠的文件输出 Sink（支持精确一次语义）。 | stream.sinkTo(FileSink.forRowFormat( new Path("/path/to/output"), new SimpleStringEncoder<>("UTF-8") ).build()); |
addSink(new FlinkKafkaProducer(...)) | 最常用。将数据写入 Apache Kafka。 | stream.addSink(new FlinkKafkaProducer<>("localhost:9092", "topic", new SimpleStringSchema())); |
addSink(new ...) | 自定义 Sink 函数。可以实现 SinkFunction 或使用第三方连接器（如 JDBC、Elasticsearch、Cassandra 等）。 | stream.addSink(new MyCustomSink()); |
4. 其他重要 API 和概念
| API / 概念 | 说明 |
|---|---|
Window | 窗口是处理无界流的核心，包括滚动窗口、滑动窗口、会话窗口等。 |
WindowAll | 不分组，在整个流上开窗（非并行操作）。 |
Process Function | 低阶、最强大的函数。可以处理每条数据、访问状态、注册定时器（实现基于事件时间的操作）。是构建复杂逻辑的基石。 |
Async I/O | 用于异步访问外部数据库或服务，避免同步调用造成的阻塞，极大提高吞吐量。 |
Side Outputs | 旁路输出。除了主数据流，还可以将特定数据输出到多个额外的流中，用于处理延迟数据、异常数据等。 |
State | 状态。Flink 容错和实现有状态计算的基础，包括 ValueState, ListState, MapState 等。 |
Checkpointing | 检查点。Flink 容错机制，定期将状态快照持久化，用于故障恢复。 |
总结
掌握 Flink DataStream API 的关键在于理解其有状态和基于时间的计算模型。
- 入门三板斧：
map,filter,keyBy+sum/reduce。 - 进阶核心：
Window（窗口）、ProcessFunction（底层处理）、State（状态管理）。 - 生产必备：
FlinkKafkaConsumer/Producer（Kafka 连接）、Async I/O（异步访问）、Checkpointing（容错）。
建议从简单的转换操作开始，逐步尝试使用 keyBy 和聚合，然后深入探索窗口和状态，最后再研究连接器、异步 I/O 等高级特性。
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
更多推荐
7
0- 0
已为社区贡献28条内容
所有评论(0)