Flink-DataStream API
一、什么样的数据可以用于流式传输
Flink的DataStream API 允许流式传输他们可以序列化的任何内容。Flink自己的序列化程序用于
- 基本类型：即字符串、长、整数、布尔值、数组
- 复合类型：元组、POJO和Scala样例类
基本类型我们已经很熟悉了，下面我们看下复合类型。
1、元组
对于java，Flink定义了Tuple0到Tuple25类型，例如：
Tuple2<String, Integer> person = Tuple2.apply("Fred",35);
String name = person._1;
Integer age = person._2;2、POJO
如果满足以下条件，Flink将数据类型识别为POJO类型（并允许“按名称”字段引用）
- 该类是公共且独立的（没有非静态内部类）
- 该类有一个公共的无参数构造函数
- 类（以及所有超类）中的所有非静态、非瞬态字段要么是公共的（和非最终的），要么具有遵循getter和setterJavabean命名约定的公共getter和setter方法。
示例：
public class Person {
public String name;
public Integer age;
public Person() {}
public Person(String name, Integer age) {
. . .
}
}
Person person = new Person("Fred Flintstone", 35);3、样例类
样例类（Case classes）和普通类差不多，只有几点关键差别。样例类非常适合用于不可变的数据，多用于模式匹配。
case class Book(isbn: String)
val frankenstein = Book("978-0486282114")注意在实例化样例类Book时，并没有使用关键字new，这是因为样例类有一个默认的apply方法来负责对象的创建。
二、完整示例
该示例来自官方网站，是将有关人员的记录流作为输入，并对其进行过滤以仅包含成年人
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.fromElements(
new Person("Fred", 35),
new Person("Wilma", 35),
new Person("Pebbles", 2));
DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
@Override
public boolean filter(Person person) throws Exception {
return person.age >= 18;
}
});
adults.print();
env.execute();
}
public static class Person {
public String name;
public Integer age;
public Person() {}
public Person(String name, Integer age) {
this.name = name;
this.age = age;
}
public String toString() {
return this.name.toString() + ": age " + this.age.toString();
}
}
}1、执行环境
每个Flink应用程序都需要一个执行环境，在该例中为env。流应用程序需要使用StreamExecutionEnvironment。
在应用程序中进行的DataStream API调用会构建一个附加到StreamExecutionEnvironment的作业图。当调用env.execute()时，此图会打包并发送到JobManager，JobManager会并行化作业并将其切片分发给TaskManager执行。作业的每个并行切片都将在一个任务槽中执行。
如果不调用execute(), 应用程序则不会执行。
此分布式运行时取决于您的应用程序是否可序列化。它还要求集群中的每个节点都可以使用所有依赖项。
2、source
上面的示例使用env.fromElements(...)构造DataStream<Person>。这是一种将简单流组合在一起以用于原型或测试的便捷方法。StreamExecutionEnvironment上还有一个fromCollection(Collection)方法。因此，也可以这样做：
List<Person> people = new ArrayList<Person>();
people.add(new Person("Fred", 35));
people.add(new Person("Wilma", 35));
people.add(new Person("Pebbles", 2));
DataStream<Person> flintstones = env.fromCollection(people);在原型设计时将一些数据导入流的另一种方便方法是使用socket或文件
DataStream<String> lines = env.socketTextStream("localhost", 9999);DataStream<String> lines = env.readTextFile("file:///path");在实际应用中，最常用的数据源是那些支持低延迟、高吞吐量并行读取以及倒带和重放的数据源——这是高性能和容错的先决条件——例如Apache Kafka、Kinesis和各种文件系统。REST API和数据库也经常使用。
3、sink
上面的示例使用adults.print()将其结果打印到任务管理器日志（在IDE中运行时将显示在IDE的控制台中）。这将在流的每个元素上调用toString()。
例如输出如下：
1> Fred: age 35
2> Wilma: age 35
其中1>和2>表示哪个子任务（即线程）产生了输出。
在生产中，常用的接收器包括FileSink、各种数据库和几个发布子系统。
---------------------------------------------------------------------------------------------------------------------------------
大多数高校硕博生毕业要求需要参加学术会议，发表EI或者SCI检索的学术论文会议论文：
可访问艾思科蓝官网，浏览即将召开的学术会议列表。会议如下：
第八届大数据与应用统计国际学术研讨会（ISBDAS 2025）
https://ais.cn/u/fEzmy2
第二届生成式人工智能与信息安全国际学术会议（GAIIS 2025）
https://ais.cn/u/uAbENn
第四届电子技术与人工智能国际学术会议（ETAI 2025）
https://ais.cn/u/vqM7Nj
第四届网络安全、人工智能与数字经济国际学术会议（CSAIDE 2025）
https://ais.cn/u/ZrERn2