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>() {@Overridepublic 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