当前位置：首页 > news >正文

从 0 到 1Flink DataStream API 入门与上手实战

news 2025/9/26 9:51:54

一、为什么选择 DataStream API？

DataStream API 是 Flink 处理连续数据流的核心编程接口，具备：

低延迟高吞吐：天然面向流处理的执行引擎。
强大的序列化与类型系统：原生支持基本类型、Tuples、POJOs；其他类型回退 Kryo；也可用 Avro 等外部序列化器。
灵活的数据接入与输出：从文件到 Kafka/Kinesis，再到数据库、对象存储、消息系统。

一句话：只要能被序列化，就能被 DataStream 流式处理。

二、可流式处理的数据类型与序列化

1）基础与复合类型

基本类型：String / Long / Integer / Boolean / Array
复合类型：Tuples、POJOs
其他类型默认回退 Kryo；也可替换为 Avro 等（推荐在跨语言/严格 schema 场景下使用）。

2）Java Tuples

Flink 在 Java 中定义了 Tuple0 ~ Tuple25：

Tuple2<String, Integer> person = Tuple2.of("Fred", 35);
// 注意：下标从 0 开始
String name = person.f0;
Integer age  = person.f1;

适用场景：原型/临时任务、字段不复杂时；大型项目建议用 POJO/Avro 以提升可读性与演进性。

3）POJOs（强烈推荐）

Flink 将满足以下条件的类识别为 POJO，并支持按字段名访问与schema 演进：

类是 public 且独立（不是非静态内部类）
具有 public 无参构造方法
所有非静态非 transient字段要么 public 非 final，要么有符合 JavaBeans 规范的 getter/setter
示例：

public class Person {public String name;public Integer age;public Person() {}public Person(String name, Integer age) {this.name = name;this.age = age;}
}
Person person = new Person("Fred Flintstone", 35);

最佳实践：POJO 字段改动、顺序调整等常见 schema 演进，Flink 原生序列化器可平滑处理。

三、首个 DataStream 示例：过滤成年人

下面的最小示例会在本地构建一个人员流，只保留年龄 ≥ 18 的元素，并打印输出。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.functions.FilterFunction;public class Example {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Person> flintstones = env.fromData(new Person("Fred", 35),new Person("Wilma", 35),new Person("Pebbles", 2));DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {@Overridepublic boolean filter(Person person) {return person.age >= 18;}});// 输出到 TaskManager 日志（IDE 控制台可见）adults.print();// 别忘了执行！env.execute("Adults Filter");}public static class Person {public String name;public Integer age;public Person() {}public Person(String name, Integer age) {this.name = name;this.age = age;}@Overridepublic String toString() {return name + ": age " + age;}}
}

输出示例：

1> Fred: age 35
2> Wilma: age 35

前缀 1> / 2> 表示由哪个并行子任务打印（线程/子任务编号）。

四、执行环境与作业提交流程（你在代码里做了什么）

每个 Flink 应用都需要一个 StreamExecutionEnvironment（流式程序必须使用此环境）。
你在代码里调用的 DataStream API（source → transform → sink）会构建作业图并绑定在 env 上。
调用 env.execute() 后，作业图被打包交给 JobManager，由其并行化并分发到多个 TaskManager 上运行，每个task slot承担作业的一段。

重要提醒：不调用 execute()，作业不会运行。

五、常用输入 Source（从原型到生产）

1）快速原型

fromData：快速构造小数据流（如上例）
Socket：配合 nc -lk 9999 快速进数据

DataStream<String> lines = env.socketTextStream("localhost", 9999);

2）文件/目录

import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.core.fs.Path;FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("file:///path")
).build();DataStream<String> lines = env.fromSource(fileSource,WatermarkStrategy.noWatermarks(),"file-input"
);