Flink核心之DataStream API

Flink具有非常灵活的分层API设计，其中核心层就是DataStream/DataSet API。由于新版已经实现了流批一体，DataSet API即将被弃用，官方推荐统一使用DataStream API处理流数据与批数据。DataStream本身是Flink中一个用来表示数据集合的类，我们编写的Flink代码其实就是基于这种数据类型的处理，所以这套核心API就以DataStream命名。在用法上有点类似于常规的Java集合，但又有所不同。我们在代码中往往并不关心集合中具体的数据，而只是用API定义出一连串操作来处理他们：我们称之为数据流的“转换”。
一个Flink程序，其实就是对DataStream的各种转换。具体来说，代码基本上都是由以下几部分构成：

• 获取执行环境（execution environment）
• 读取数据源（source）
• 定义基于数据的转换操作（transformations）
• 定义计算结果的输出位置（sink）
• 触发程序执行（execute）

1. 执行环境（Execution Environment）

Flink 程序可以在各种上下文环境中运行：我们可以在本地 JVM 中执行程序，也可以提交到远程集群上运行。
不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前 Flink 的运行环境，从而建立起与 Flink 框架之间的联系。只有获取了环境 上下文信息，才能将具体的任务调度到不同的 TaskManager 执行。

1.1 创建执行环境

编写Flink程序的第一步，就是创建执行环境。我们要获取的执行环境，是<br />StreamExecutionEnvironment 类的对象，这是所有Flink程序的基础。在代码中创建执行环境的方式，就是调用这个类的静态方法，具体有以下三种。

• getExecutionEnvironment
• createLocalEnvironment
• createRemoteEnvironment

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//会根据当前运行环境直接得到正确的结果

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();
//这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的 CPU 核心数

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment.createRemoteEnvironment(
 "hadoop0", // JobManager 主机名
 7777, // JobManager 进程端口号
 "test/wordcount.jar" // 提交给 JobManager 的 JAR 包
);

1.2 执行模式（Execution Mode）

流批一体后，我们获取到的执行环境是StreamExecutionEnvironment，看名称它是做流处理的，那么我们如何去执行批处理呢？

// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

那么我们不难看出：基于ExecutionEnvironment读入数据创建的数据集合，就是DataSet；对应的调用的一整套转换方法，就是DataSet API。
由于Flink程序默认是STREAMING模式，我们这里重点介绍一下BATCH模式的配置，分别是命令行配置与代码配置

flink run -Dexecution.runtime-mode=BATCH /test/wordcount.jar

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

当然不建议在代码配置，使用命令行配置比较灵活，既可以使用流处理也可以使用批处理。

2. 读取数据源（Source）

创建好执行环境之后，我们先要读取数据才能对数据做处理，Flink读取数据，是调用执行环境的addSource()方法

DataStream<String> stream = env.addSource(...);

方法传入一个对象参数，需要实现 SourceFunction 接口；返回 DataStreamSource。这里的DataStreamSource 类继承自 SingleOutputStreamOperator 类，又进一步继承自 DataStream。所以很明显，读取数据的 source 操作是一个算子，得到的是一个数据流（DataStream）。
传入的参数是一个“源函数”（source function），需要实现 SourceFunction 接口。这是何方神圣，又该怎么实现呢？ 自己去实现它显然不会是一件容易的事。好在 Flink 直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的 source function，通常情况下足以应对我们的实际需求。
我们这里构建一个实际应用场景。比如网站的访问操作，我们把它抽象成一个三元组（用户名、用户访问的url，用户访问url的时间戳），我们这里创建一个类，将用户行为包装成一个对象

import java.sql.Timestamp;
public class Event {
    public String user;
    public String url;
    public Long timestamp;
    public Event() {
    }
    public Event(String user, String url, Long timestamp) {
        this.user = user;
        this.url = url;
        this.timestamp = timestamp;
    }
    @Override
    public String toString() {
        return "Event{" +
                "user='" + user + '\'' +
                ", url='" + url + '\'' +
                ", timestamp=" + new Timestamp(timestamp) +
                '}';
    }

这里需要注意，我们定义的 Event，有这样几个特点：

• 类是公有（public）的
• 有一个无参的构造方法
• 所有属性都是公有（public）的
• 所有属性的类型都是可以序列化的
  Flink 会把这样的类作为一种特殊的 POJO 数据类型来对待，方便数据的解析和序列化。另外我们在类中还重写了 toString 方法，主要是为了测试输出显示更清晰。

2.1 从集合中读取数据

最简单的读取方式，就是在代码中创建一个Java集合，然后调用执行环境的fromCollection方法进行读取，一般用于测试。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//设置并行度 
env.setParallelism(1);
ArrayList<Event> clicks = new ArrayList<>();
clicks.add(new Event("Mary","./home",1000L));
clicks.add(new Event("Bob","./cart",2000L));
DataStream<Event> stream = env.fromCollection(clicks);

也可以不创建集合，直接列举

DataStreamSource<Event> stream2 = env.fromElements(
 new Event("Mary", "./home", 1000L),
 new Event("Bob", "./cart", 2000L)
);

2.2 从文件读取数据

在真正的实际应用中，不会直接将数据写在代码里。通常情况下我们会从存储介质中获取数据。比较常见的就是读取文件

DataStream<String> stream = env.readTextFile("clicks.csv");

说明:

• 参数可以是目录，也可以是文件；
• 路径可以是相对路径，也可以是绝对路径；
• 相对路径是从系统属性 user.dir 获取路径: idea 下是 project 的根目录, standalone 模式下是集群节点根目录；
• 也可以从 hdfs 目录下读取, 使用路径 hdfs://…, 由于 Flink 没有提供 hadoop 相关依赖, 需要 pom 中添加相关依赖:

<dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-client</artifactId>
 <version>${hadoop.version}</version>
 <scope>provided</scope>
</dependency>

2.3 从Socket读取数据

无论从集合还是文件，我们读取的数据其实都是有界数据。在流处理场景中，数据往往都是无界的，一种比较简单的方式，就是读取socket文本流，但这种一般也用于测试

DataStream<String> stream = env.socketTextStream("localhost", 7777);

2.4 从Kafka读取数据

Kafka 作为分布式消息传输队列，是一个高吞吐、易于扩展的消息系统。而消息队列的传输方式，恰恰和流处理是完全一致的。所以可以说 Kafka 和 Flink 天生一对，是当前处理流式 数据的双子星。在如今的实时流处理应用中，由 Kafka 进行数据的收集和传输，Flink 进行分析计算，这样的架构已经成为众多企业的首选。
略微遗憾的是，与 Kafka 的连接比较复杂，Flink 内部并没有提供预实现的方法。所以我 们只能采用通用的 addSource 方式、实现一个 SourceFunction 了。好在Kafka与Flink确实是非常契合，所以Flink官方提供了连接工具flink-connector-kafka， 直接帮我们实现了一个消费者 FlinkKafkaConsumer，它就是用来读取 Kafka 数据的 SourceFunction。
只需导入下面依赖

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

然后调用env.addSource()，传入FlinkKafkaConsumer的对象实例即可

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
public class SourceKafkaTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop0:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");
        DataStreamSource<String> stream = env.addSource(new
                FlinkKafkaConsumer<String>(
                "clicks",
                new SimpleStringSchema(),
                properties
        ));
        stream.print("Kafka");
        env.execute();
    }
}

创建 FlinkKafkaConsumer 时需要传入三个参数：

• 第一个参数 topic，定义了从哪些主题中读取数据。可以是一个 topic，也可以是 topic 列表，还可以是匹配所有想要读取的 topic 的正则表达式。当从多个 topic 中读取数据 时，Kafka 连接器将会处理所有 topic 的分区，将这些分区的数据放到一条流中去。
• 第二个参数是一个 DeserializationSchema 或者 KeyedDeserializationSchema。Kafka 消 息被存储为原始的字节数据，所以需要反序列化成 Java 或者 Scala 对象。上面代码中使用的 SimpleStringSchema，是一个内置的 DeserializationSchema，它只是将字节数组简单地反序列化成字符串。DeserializationSchema 和 KeyedDeserializationSchema 是公共接口，所以我们也可以自定义反序列化逻辑。

• 第三个参数是一个 Properties 对象，设置了 Kafka 客户端的一些属性。

  当然你也可以自定义Source，这里暂不做过多讨论。

3. 转换算子（Transform）

3.1 基本转换算子

3.1.1 映射（map）

map 是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素。
我们只需要基于 DataStrema 调用 map()方法就可以进行转换处理。方法需要传入的参数是 接口 MapFunction 的实现；返回值类型还是 DataStream，不过泛型（流中的元素类型）可能改变。
下面我们用不同的方式提取Event中的user字段功能。

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransMapTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 传入匿名类，实现 MapFunction
        stream.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event e) throws Exception {
                return e.user;
            }
        });
        // 传入 MapFunction 的实现类
        stream.map(new UserExtractor()).print();
        env.execute();
    }
    public static class UserExtractor implements MapFunction<Event, String> {
        @Override
        public String map(Event e) throws Exception {
            return e.user;
        }
    } }

3.1.2 过滤（filter）

filter 转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤 条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。
进行 filter 转换之后的新数据流的数据类型与原数据流是相同的。filter 转换需要传入的参 数需要实现 FilterFunction 接口，而 FilterFunction 内要实现 filter()方法，就相当于一个返回布尔类型的条件表达式。下面的代码会将数据流中用户 Mary 的浏览行为过滤出来

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransFilterTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 传入匿名类实现 FilterFunction
        stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event e) throws Exception {
                return e.user.equals("Mary");
            }
        });
        // 传入 FilterFunction 实现类
        stream.filter(new UserFilter()).print();
        env.execute();
    }
    public static class UserFilter implements FilterFunction<Event> {
        @Override
        public boolean filter(Event e) throws Exception {
            return e.user.equals("Mary");
        }
    }
}

3.1.3 扁平映射（flatMap）

flatMap 操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个 一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap 可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。
flatMap 操作会应用在每一个输入事件上面，FlatMapFunction 接口中定义了 flatMap 方法， 用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回 0 个、1 个或多个结果数据。因此 flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。所以 flatMap 方法也可以实现 map 方法和 filter 方法的功能，当返回结果 是 0 个的时候，就相当于对数据进行了过滤，当返回结果是 1 个的时候，相当于对数据进行了简单的转换操作。

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
public class TransFlatmapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        stream.flatMap(new MyFlatMap()).print();
        env.execute();
    }
    public static class MyFlatMap implements FlatMapFunction<Event, String> {
        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            if (value.user.equals("Mary")) {
                out.collect(value.user);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.url);
            }
        }
    }
}

3.2 聚合算子（Aggregation）

直观上看，基本转换算子确实是在“转换”——因为它们都是基于当前数据，去做了处理 和输出。而在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的 信息。比如之前 word count 程序中，要对每个词出现的频次进行叠加统计。这种操作，计算 的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并——这就是所谓的“聚合”（Aggregation），也对应着 MapReduce 中的 reduce 操作。

3.2.1 按键分区（keyBy）

对于 Flink 而言，DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区； 这个操作就是通过 keyBy 来完成的。keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑 上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
基于不同的 key，流中的数据将被分配到不同的分区中去；这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot中进行处理了。
在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。所以 这里 key 如果是 POJO 的话，必须要重写 hashCode()方法。keyBy()方法需要传入一个参数，这个参数指定了一个或一组 key。有很多不同的方法来指 定 key：比如对于 Tuple 数据类型，可以指定字段的位置或者多个位置的组合；对于 POJO 类型，可以指定字段的名称（String）；另外，还可以传入 Lambda 表达式或者实现一个键选择器 （KeySelector），用于说明从数据中提取 key 的逻辑。
我们可以以 id 作为 key 做一个分区操作，代码实现如下：

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransKeyByTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 使用 Lambda 表达式
        KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user);
        // 使用匿名类实现 KeySelector
        KeyedStream<Event, String> keyedStream1 = stream.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event e) throws Exception {
                return e.user;
            }
        });
        stream.print();
        env.execute();
    }
}

需要注意的是，keyBy 得到的结果将不再是 DataStream，而是会将 DataStream 转换为 KeyedStream。KeyedStream 可以认为是“分区流”或者“键控流”，它是对 DataStream 按照 key 的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定 key 的 类型。
KeyedStream 也继承自 DataStream，所以基于它的操作也都归属于 DataStream API。但它跟之前的转换操作得到的 SingleOutputStreamOperator 不同，只是一个流的分区操作，并不是 一个转换算子。KeyedStream 是一个非常重要的数据结构，只有基于它才可以做后续的聚合操 作（比如 sum，reduce）；而且它可以将当前算子任务的状态（state）也按照 key 进行划分、限定为仅对当前 key 有效。

3.2.2 简单聚合

有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink 为我们 内置实现了一些最基本、最简单的聚合 API，主要有以下几种：

• sum()：在输入流上，对指定的字段做叠加求和的操作。
• min()：在输入流上，对指定的字段求最小值。
• max()：在输入流上，对指定的字段求最大值。
• minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。

• maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与 min()/minBy()完全一致。

  简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数； 但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。

  对于元组类型的数据，同样也可以使用这两种方式来指定字段。需要注意的是，元组中字段的名称，是以 f0、f1、f2、…来命名的。

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransTupleAggreationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple2<String, Integer>> stream = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("a", 3),
                Tuple2.of("b", 3),
                Tuple2.of("b", 4)
        );
        stream.keyBy(r -> r.f0).sum(1).print();
        stream.keyBy(r -> r.f0).sum("f1").print();
        stream.keyBy(r -> r.f0).max(1).print();
        stream.keyBy(r -> r.f0).max("f1").print();
        stream.keyBy(r -> r.f0).min(1).print();
        stream.keyBy(r -> r.f0).min("f1").print();
        stream.keyBy(r -> r.f0).maxBy(1).print();
        stream.keyBy(r -> r.f0).maxBy("f1").print();
        stream.keyBy(r -> r.f0).minBy(1).print();
        stream.keyBy(r -> r.f0).minBy("f1").print();
        env.execute();
    }
}

而如果数据流的类型是 POJO 类，那么就只能通过字段名称来指定，不能通过位置来指定

了

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransPojoAggregationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        stream.keyBy(e -> e.user).max("timestamp").print(); // 指定字段名称
        env.execute();
    }
}

简单聚合算子返回的，同样是一个 SingleOutputStreamOperator，也就是从 KeyedStream 又 转换成了常规的 DataStream。所以可以这样理解：keyBy 和聚合是成对出现的，先分区、后聚合，得到的依然是一个 DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不 变。
一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。 所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值 的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个 key 的数据流上。

3.2.3 归约聚合（reduce）

如果说简单聚合是对一些特定统计需求的实现，那么 reduce 算子就是一个一般化的聚合 统计操作了。从大名鼎鼎的 MapReduce 开始，我们对 reduce 操作就不陌生：它可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。
与简单聚合类似，reduce 操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。
ReduceFunction 接口里需要实现 reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件；所以，对于一组数据，我们可以先取两个进行合并，然后再 将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据， 这也就是 reduce“归约”的含义。在流处理的底层实现过程中，实际上是将中间“合并的结果” 作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。
其实，reduce 的语义是针对列表进行规约操作，运算规则由 ReduceFunction 中的 reduce方法来定义，而在 ReduceFunction 内部会维护一个初始值为空的累加器，注意累加器的类型和输入元素的类型相同，当第一条元素到来时，累加器的值更新为第一条元素的值，当新的元素到来时，新元素会和累加器进行累加操作，这里的累加操作就是 reduce 函数定义的运算规 则。然后将更新以后的累加器的值向下游输出。
我们可以单独定义一个函数类实现 ReduceFunction 接口，也可以直接传入一个匿名类。 当然，同样也可以通过传入 Lambda 表达式实现类似的功能。
与简单聚合类似，reduce 操作也会将 KeyedStream 转换为 DataStrema。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。
看一下稍复杂的例子，我们将数据流按照用户 id 进行分区，然后用一个 reduce 算子实现 sum 的功能，统计每个 用户访问的频次；进而将所有统计结果分到一组，用另一个 reduce 算子实现 maxBy 的功能，记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Calendar;
import java.util.Random;
public class ClickSource implements SourceFunction<Event> {
    // 声明一个布尔变量，作为控制数据生成的标识位
    private Boolean running = true;
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        Random random = new Random(); // 在指定的数据集中随机选取数据
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1",
                "./prod?id=2"};
        while (running) {
            ctx.collect(new Event(
                    users[random.nextInt(users.length)],
                    urls[random.nextInt(urls.length)],
                    Calendar.getInstance().getTimeInMillis()
            ));
            // 隔 1 秒生成一个点击事件，方便观测
            Thread.sleep(1000);
        }
    }
    @Override
    public void cancel() {
        running = false;
    }
}

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 这里的 ClickSource()使用了之前自定义数据源小节中的 ClickSource()
        env.addSource(new ClickSource())
                // 将 Event 数据类型转换成元组类型
                .map(new MapFunction<Event, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Event e) throws Exception {
                        return Tuple2.of(e.user, 1L);
                    }
                })
                .keyBy(r -> r.f0) // 使用用户名来进行分流
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1,
                                                       Tuple2<String, Long> value2) throws Exception {
                        // 每到一条数据，用户 pv 的统计值加 1
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .keyBy(r -> true) // 为每一条数据分配同一个 key，将聚合结果发送到一条流中去
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1,
                                                       Tuple2<String, Long> value2) throws Exception {
                        // 将累加器更新为当前最大的 pv 统计值，然后向下游发送累加器的值
                        return value1.f1 > value2.f1 ? value1 : value2;
                    }
                })
                .print();
        env.execute();
    }
}

reduce 同简单聚合算子一样，也要针对每一个 key 保存状态。因为状态不会清空，所以我们需要将 reduce 算子作用在一个有限 key 的流上。

3.3 用户自定义函数（UDF）

3.3.1 函数类

对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口， 来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类， 例如 MapFunction、FilterFunction、ReduceFunction 等。所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。
下面例子实现了FilterFunction接口，用来筛选url中包含“home”的事件。

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransFunctionUDFTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> clicks = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        DataStream<Event> stream = clicks.filter(new FlinkFilter());
        stream.print();
        env.execute();
    }
    public static class FlinkFilter implements FilterFunction<Event> {
        @Override
        public boolean filter(Event value) throws Exception {
            return value.url.contains("home");
        }
    }
}

也可以通过匿名函数类实现

DataStream<String> stream = clicks.filter(new FilterFunction<Event>() {
 @Override
 public boolean filter(Event value) throws Exception {
 return value.url.contains("home");
 }
});

为了方法更加通用，我们还可以将用于过滤关键字“home”抽象出来作为类的属性，调用构造方法时传进去。

    DataStream<Event> stream = clicks.filter(new KeyWordFilter("home"));
    public static class KeyWordFilter implements FilterFunction<Event> {
        private String keyWord;
        KeyWordFilter(String keyWord) {
            this.keyWord = keyWord;
        }
        @Override
        public boolean filter(Event value) throws Exception {
            return value.url.contains(this.keyWord);
        }
    }

3.3.2 匿名函数（Lambda）

匿名函数（Lambda 表达式）是 Java 8 引入的新特性，方便我们更加快速清晰地写代码。Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。
Flink 的所有算子都可以使用 Lambda 表达式的方式来进行编码，但是，当 Lambda 表 达式使用 Java 的泛型时，我们需要显式的声明类型信息。
下例演示了如何使用 Lambda 表达式来实现一个简单的 map() 函数，我们使用 Lambda  表达式来计算输入的平方。在这里，我们不需要声明 map() 函数的输入 i 和输出参数的数据类型，因为 Java 编译器会对它们做出类型推断。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TransFunctionLambdaTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> clicks = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        //map 函数使用 Lambda 表达式，返回简单类型，不需要进行类型声明
        DataStream<String> stream1 = clicks.map(event -> event.url);
        stream1.print();
        env.execute();
    }
}

由于 OUT 是 String 类型而不是泛型，所以 Flink 可以从函数签名 OUT map(IN value)  的实现中自动提取出结果的类型信息。
但是对于像 flatMap() 这样的函数，它的函数签名 void flatMap(IN value, Collector  out) 被 Java 编译器编译成了 void flatMap(IN value, Collector out)，也就是说将 Collector 的泛型信息擦除掉了。这样 Flink 就无法自动推断输出的类型信息了。

// flatMap 使用 Lambda 表达式，抛出异常
DataStream<String> stream2 = clicks.flatMap((event, out) -> {
out.collect(event.url);
});
stream2.print();

在这种情况下，我们需要显式地指定类型信息，否则输出将被视为 Object 类型

// flatMap 使用 Lambda 表达式，必须通过 returns 明确声明返回类型
DataStream<String> stream2 = clicks.flatMap((Event event, Collector<String> out) -> {
out.collect(event.url);
}).returns(Types.STRING);
stream2.print();

当使用 map() 函数返回 Flink 自定义的元组类型时也会发生类似的问题。下例中的函数签 名 Tuple2 map(Event value) 被类型擦除为 Tuple2 map(Event value)。

//使用 map 函数也会出现类似问题，以下代码会报错
DataStream<Tuple2<String, Long>> stream3 = clicks
.map( event -> Tuple2.of(event.user, 1L) );
stream3.print();

一般来说，这个问题可以通过多种方式解决

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class ReturnTypeResolve {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> clicks = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 想要转换成二元组类型，需要进行以下处理
        // 1) 使用显式的 ".returns(...)"
        DataStream<Tuple2<String, Long>> stream3 = clicks
                .map(event -> Tuple2.of(event.user, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));
        stream3.print();
        // 2) 使用类来替代 Lambda 表达式
        clicks.map(new MyTuple2Mapper())
                .print();
        // 3) 使用匿名类来代替 Lambda 表达式
        clicks.map(new MapFunction<Event, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Event value) throws Exception {
                return Tuple2.of(value.user, 1L);
            }
        }).print();
        env.execute();
    }
    // 自定义 MapFunction 的实现类
    public static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String,
            Long>> {
        @Override
        public Tuple2<String, Long> map(Event value) throws Exception {
            return Tuple2.of(value.user, 1L);
        }
    }
}

3.3.3 富函数类

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其 Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、 RichReduceFunction 等。
既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不 同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
一个常见的应用场景就是，如果我们希望连接到一个外部数据库进行读写操作，那么将连接操作放在 map()中显然不是个好选择——因为每来一条数据就会重新连接一次数据库；所以 我们可以在 open()中建立连接，在 map()中读写数据，而在 close()中关闭连接。所以我们推荐的最佳实践如下：

public class MyFlatMap extends RichFlatMapFunction<IN, OUT>> {
 @Override
 public void open(Configuration configuration) {
 // 做一些初始化工作
 // 例如建立一个和 MySQL 的连接
 }
 @Override
 public void flatMap(IN in, Collector<OUT out) {
 // 对数据库进行读写
 }
 @Override
 public void close() {
 // 清理工作，关闭和 MySQL 数据库的连接。
 } }

4. 输出算子（Sink）

在 Flink 中，如果我们希望将数据写入外部系统，其实并不是一件难事。我们知道所有算 子都可以通过实现函数类来自定义处理逻辑，所以只要有读写客户端，与外部系统的交互在任 何一个处理算子中都可以实现。例如在 MapFunction 中，我们完全可以构建一个到 Redis 的连接，然后将当前处理的结果保存到 Redis 中。如果考虑到只需建立一次连接，我们也可以利用 RichMapFunction，在 open() 生命周期中做连接操作。
这样看起来很方便，却会带来很多问题。Flink 作为一个快速的分布式实时流处理系统， 对稳定性和容错性要求极高。一旦出现故障，我们应该有能力恢复之前的状态，保障处理结果 的正确性。这种性质一般被称作“状态一致性”。Flink 内部提供了一致性检查点（checkpoint） 来保障我们可以回滚到正确的状态；但如果我们在处理过程中任意读写外部系统，发生故障后就很难回退到从前了。
为了避免这样的问题，Flink 的 DataStream API 专门提供了向外部写入数据的方法：addSink。与 addSource 类似，addSink 方法对应着一个“Sink”算子，主要就是用来实现与外 部系统连接、并将数据提交写入的；Flink 程序中所有对外的输出操作，一般都是利用 Sink 算 子完成的。 Sink 一词有“下沉”的意思，有些资料会相对于“数据源”把它翻译为“数据汇”。不论 怎样理解，Sink 在 Flink 中代表了将结果数据收集起来、输出到外部的意思，所以我们这里统一把它直观地叫作“输出算子”。
Flink 官方提供了一部分的框架的 Sink 连接器。如图所示，列出了 Flink 官方目前支持的第三方系统连接器

我们可以看到，像 Kafka 之类流式系统，Flink 提供了完美对接，source/sink 两端都能连 接，可读可写；而对于 Elasticsearch、文件系统（FileSystem）、JDBC 等数据存储系统，则只提供了输出写入的 sink 连接器。
除 Flink 官方之外，Apache Bahir 作为给 Spark 和 Flink 提供扩展支持的项目，也实现了一些其他第三方系统与 Flink 的连接器。
除此之外，就需要用户自定义实现sink连接器了。

4.1 输出到文件

最简单的输出方式，当然就是写入文件了。对应着读取文件作为输入数据源，Flink 本来 也有一些非常简单粗暴的输出到文件的预实现方法：如 writeAsText()、writeAsCsv()，可以直接将输出结果保存到文本文件或 Csv 文件。但我们知道，这种方式是不支持同时写入一份文 件的；所以我们往往会将最后的 Sink 操作并行度设为 1，这就大大拖慢了系统效率；而且对于故障恢复后的状态一致性，也没有任何保证。所以目前这些简单的方法已经要被弃用。
Flink 为此专门提供了一个流式文件系统的连接器：StreamingFileSink，它继承自抽象类 RichSinkFunction，而且集成了 Flink 的检查点（checkpoint）机制，用来保证精确一次（exactly once）的一致性语义。StreamingFileSink 为批处理和流处理提供了一个统一的 Sink，它可以将分区文件写入 Flink 支持的文件系统。它可以保证精确一次的状态一致性，大大改进了之前流式文件 Sink 的方式。它的主要操作是将数据写入桶（buckets），每个桶中的数据都可以分割成一个个大小有限的分 区文件，这样一来就实现真正意义上的分布式文件存储。我们可以通过各种配置来控制“分桶” 的操作；默认的分桶方式是基于时间的，我们每小时写入一个新的桶。换句话说，每个桶内保 存的文件，记录的都是 1 小时的输出数据。
StreamingFileSink 支持行编码（Row-encoded）和批量编码（Bulk-encoded，比如 Parquet）格式。这两种不同的方式都有各自的构建器（builder），调用方法也非常简单，可以直接调用 StreamingFileSink 的静态方法：

• 行编码：StreamingFileSink.forRowFormat（basePath，rowEncoder）

• 批量编码：StreamingFileSink.forBulkFormat（basePath，bulkWriterFactory）

  在创建行或批量编码 Sink 时，我们需要传入两个参数，用来指定存储桶的基本路径 （basePath）和数据的编码逻辑（rowEncoder 或 bulkWriterFactory）。 下面我们就以行编码为例，将一些测试数据直接写入文件：

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import java.util.concurrent.TimeUnit;
public class SinkToFileTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStreamSource<Event> stream = env.fromElements(new Event("Mary",
                        "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=100", 3000L),
                new Event("Alice", "./prod?id=200", 3500L),
                new Event("Bob", "./prod?id=2", 2500L),
                new Event("Alice", "./prod?id=300", 3600L),
                new Event("Bob", "./home", 3000L),
                new Event("Bob", "./prod?id=1", 2300L),
                new Event("Bob", "./prod?id=3", 3300L));
        StreamingFileSink<String> fileSink = StreamingFileSink
                .<String>forRowFormat(new Path("./output"),
                        new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)
                                )
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5
                                ))
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .build())
                .build();
        // 将 Event 转换成 String 写入文件
        stream.map(Event::toString).addSink(fileSink);
        env.execute();
    }
}

这里我们创建了一个简单的文件 Sink，通过.withRollingPolicy()方法指定了一个“滚动策 略”。“滚动”的概念在日志文件的写入中经常遇到：因为文件会有内容持续不断地写入，所以 我们应该给一个标准，到什么时候就开启新的文件，将之前的内容归档保存。也就是说，上面 的代码设置了在以下 3 种情况下，我们就会滚动分区文件：

• 至少包含 15 分钟的数据
• 最近 5 分钟没有收到新的数据
• 文件大小已达到 1 GB

4.2 输出到Kafka

和输入源类似，直接看代码吧

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.Properties;
public class SinkToKafkaTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "hadoop0:9092");
        DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");
        stream
                .addSink(new FlinkKafkaProducer<String>(
                        "clicks",
                        new SimpleStringSchema(),
                        properties
                ));
        env.execute();
    }
}

FlinkKafkaProducer 继承了抽象类 TwoPhaseCommitSinkFunction，这是一个实现了“两阶段提交”的 RichSinkFunction。两阶段提 交提供了 Flink 向 Kafka 写入数据的事务性保证，能够真正做到精确一次（exactly once）的状态一致性。关于这部分内容，我们后续再谈。
可以启动一个消费者查看是否收到了数据

/kafka-console-consumer.sh --bootstrap-server hadoop0:9092 --topic clicks

4.3 输出到Redis

Flink 没有直接提供官方的 Redis 连接器，不过 Bahir 项目还是担任了合格的辅助角色，为我们提供了 Flink-Redis 的连接工具。但版本升级略显滞后，目前连接器版本为 1.0，支持的 Scala 版本最新到 2.11。由于我们的测试不涉及到 Scala 的相关版本变化，所以并不影响使用。

<dependency>
 <groupId>org.apache.bahir</groupId>
 <artifactId>flink-connector-redis_2.11</artifactId>
 <version>1.0</version>
</dependency>

连接器为我们提供了一个 RedisSink，它继承了抽象类 RichSinkFunction，这就是已经实现好的向 Redis 写入数据的 SinkFunction。我们可以直接将 Event 数据输出到 Redis。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
public class SinkToRedisTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建一个到 redis 连接的配置
        FlinkJedisPoolConfig conf = new
                FlinkJedisPoolConfig.Builder().setHost("hadoop0").build();
        env.addSource(new ClickSource())
                .addSink(new RedisSink<Event>(conf, new MyRedisMapper()));
        env.execute();
    }
}
class MyRedisMapper implements RedisMapper<Event> {
    @Override
    public String getKeyFromData(Event e) {
        return e.user;
    }
    @Override
    public String getValueFromData(Event e) {
        return e.url;
    }
    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.HSET, "clicks");
    }
}

在这里我们可以看到，保存到 Redis 时调用的命令是 HSET，所以是保存为哈希表（hash），表名为“clicks”；保存的数据以 user 为 key，以 url 为 value，每来一条数据就会做一次转换。

查看Redis是否收到数据

4.4 输出到Elasticsearch

Flink 为 ElasticSearch 专门提供了官方的 Sink 连接器，Flink 1.13 支持当前最新版本的 ElasticSearch

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import java.util.ArrayList;
import java.util.HashMap;
public class SinkToEsTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=100", 3000L),
                new Event("Alice", "./prod?id=200", 3500L),
                new Event("Bob", "./prod?id=2", 2500L),
                new Event("Alice", "./prod?id=300", 3600L),
                new Event("Bob", "./home", 3000L),
                new Event("Bob", "./prod?id=1", 2300L),
                new Event("Bob", "./prod?id=3", 3300L));
        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("hadoop0", 9200, "http"));
// 创建一个 ElasticsearchSinkFunction
        ElasticsearchSinkFunction<Event> elasticsearchSinkFunction = new
                ElasticsearchSinkFunction<Event>() {
                    @Override
                    public void process(Event element, RuntimeContext ctx, RequestIndexer
                            indexer) {
                        HashMap<String, String> data = new HashMap<>();
                        data.put(element.user, element.url);
                        IndexRequest request = Requests.indexRequest()
                                .index("clicks")
                                .type("type") // Es 6 必须定义 type
                                .source(data);
                        indexer.add(request);
                    }
                };
        stream.addSink(new ElasticsearchSink.Builder<Event>(httpHosts,
                elasticsearchSinkFunction).build());
        env.execute();
    }
}

4.5 输出到MySQL（JDBC）

关系型数据库有着非常好的结构化数据设计、方便的 SQL 查询，是很多企业中业务数据 存储的主要形式。MySQL 就是其中的典型代表。尽管在大数据处理中直接与 MySQL 交互的 场景不多，但最终处理的计算结果是要给外部应用消费使用的，而外部应用读取的数据存储往 往就是 MySQL。所以我们也需要知道如何将数据输出到 MySQL 这样的传统数据库。

添加依赖:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>