1.执行环境

一个 Flink 程序，其实就是对 DataStream 的各种转换。具体来说，代码基本上都由以下几部分构成：

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

• 触发程序执行（execute）

  其中，获取环境和触发执行，都可以认为是针对执行环境的操作。

1.1 创建执行环境

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

（1)getExecutionEnvironment

直接调用 getExecutionEnvironment 方法。它会根据当前运行的上下文 直接得到正确的结果：如果程序是独立运行的，就返回一个本地执行环境；如果是创建了 jar 包，然后从命令行调用它并提交到集群执行，那么就返回集群的执行环境。也就是说，这个方 法会根据当前运行的方式，自行决定该返回什么样的运行环境。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

(2) createLocalEnvironment

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

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();

(3) createRemoteEnvironment

这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定要在集群中运行的Jar包。

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
     .createRemoteEnvironment(
         "host", // JobManager 主机名
         1234, // JobManager 进程端口号
         "path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
    );

1.2 执行模式

（1）流执行模式（STREAMING）

这是 DataStream API 最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是 STREAMING 执行模式。

（2） 批执行模式（BATCH）

专门用于批处理的执行模式, 这种模式下，Flink 处理作业的方式类似于 MapReduce 框架。 对于不会持续计算的有界数据，用这种模式处理会更方便。

（3） 自动模式（AUTOMATIC）

将由程序根据输入数据源是否有界，来自动选择执行模式。

由于 Flink 程序默认是 STREAMING 模式。BATCH 模式的配置，主要有两种方式：

• 通过命令行配置

bin/flink run -Dexecution.runtime-mode=BATCH ...

• 通过代码配置

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

 用 BATCH 模式处理批量数据，用 STREAMING 模式处理流式数据。因为数据有界的时候，直接输出结果会更加高效；而当数据无界的时候, 我 们没得选择——只有 STREAMING 模式才能处理持续的数据流。  

1.3 触发执行

Flink 是由事件驱动的，只有等到数据到来，才会触发真正的计算， 这也被称为“延迟执行”或“懒执行”(lazy execution)。

所以我们需要显式地调用执行环境的 execute()方法，来触发程序执行。execute()方法将一 直等待作业完成，然后返回一个执行结果（JobExecutionResult）。

env.execute();

2.源算子

 Flink 可以从各种来源获取数据，然后构建 DataStream 进行转换处理。一般将数据的输入 来源称为数据源(data source)，而读取数据的算子就是源算子（source operator）。所以，source 就是我们整个处理程序的输入端。  

添加 source 的方式，是调用执行环境的 addSource()方法：

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

方法传入一个对象参数，需要实现 SourceFunction 接口；返回 DataStreamSource。这里的 DataStreamSource 类继承自 SingleOutputStreamOperator 类，又进一步继承自 DataStream。所以 很明显，读取数据的 source 操作是一个算子，得到的是一个数据流（DataStream）。

2.1 从集合中获取数据

在代码中直接创建一个 Java 集合，然后调用执行环境的 fromCollection 方法进行读取。这相当于将数据临时存储到内存中，形成特殊的数据结构后， 作为数据源使用，一般用于测试。

public static void main(String[] args) throws Exception {
    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);
    stream.print();
    env.execute();
}

 我们也可以不构建集合，直接将元素列举出来，调用 fromElements 方法进行读取数据：  

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>2.7.5</version>
   <scope>provided</scope>
</dependency>

2.3 从Socket中读取数据

读取 socket 文本流。这种方式由于吞吐量小、稳 定性较差，一般也是用于测试。

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

2.4 从Kafka读取数据

Kafka作为分布式消息传输队列，是一个高吞吐、易于扩展的消息系统。而消息队列的传 输方式，恰恰和流处理是完全一致的。Kafka 进行数据的收集和传输，Flink 进行分析计算。

Kafka 的连接比较复杂，Flink 内部并没有提供预实现的方法，只能采用通用的 addSource 方式、实现一个 SourceFunction 。

Flink官方提供连接工具flink-connector-kafka， 直接实现了一个消费者FlinkKafkaConsumer，它就是用来读取 Kafka 数据的 SourceFunction。

所以想要以 Kafka 作为数据源获取数据，我们只需要引入 Kafka 连接器的依赖。Flink 官 方提供的是一个通用的 Kafka 连接器，它会自动跟踪最新版本的 Kafka 客户端。目前最新版本 只支持 0.10.0 版本以上的 Kafka。

<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", "hadoop102: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 客户端的一些属性。

2.5 自定义Source

创建一个自定义的数据源，实现 SourceFunction 接口。主要重写两个关键方法： run()和 cancel()。

• run()方法：使用运行时上下文对象（SourceContext）向下游发送数据；
• cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

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.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SourceCustom {
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //有了自定义的 source function，调用 addSource 方法
         DataStreamSource<Event> stream = env.addSource(new ClickSource());
         stream.print("SourceCustom");
         env.execute();
     }
}

SourceFunction 接口定义的数据源，并行度只能设置为 1，如果数据源设置为大于 1 的并行度，则会抛出异常

Exception in thread "main" java.lang.IllegalArgumentException: The parallelism    of non parallel operator must be 1.

如果我们想要自定义并行的数据源的话，需要使用 ParallelSourceFunction

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import java.util.Random;
public class ParallelSourceExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new CustomSource()).setParallelism(2).print();
        env.execute();
    }
     public static class CustomSource implements ParallelSourceFunction<Integer>{
         private boolean running = true;
         private Random random = new Random();
         @Override
         public void run(SourceContext<Integer> sourceContext) throws Exception {
             while (running) {
                 sourceContext.collect(random.nextInt());
             }
         }
        @Override
         public void cancel() {
             running = false;
        }
    }
}

2.6 Flink 支持的数据类型

2.6.1 Flink 的类型系统

为了方便地处理数据，Flink 有自己一整套类型系统。Flink 使用“类型信息” （TypeInformation）来统一表示数据类型。TypeInformation 类是 Flink 中所有类型描述符的基类。 它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。

2.6.2 Flink 支持的数据类型

对于常见的 Java 和 Scala 数据类型，Flink 都是支持的。Flink 在内部，Flink 对支持不同的类型进行了划分，这些类型可以在 Types 工具类中找到：

1. 基本类型

   所有 Java 基本类型及其包装类，再加上 Void、String、Date、BigDecimal 和 BigInteger。

2. 数组类型

   包括基本类型数组（PRIMITIVE_ARRAY）和对象数组(OBJECT_ARRAY)

3. 复合数据类型

• Java 元组类型（TUPLE）：这是 Flink 内置的元组类型，是 Java API 的一部分。最多 25 个字段，也就是从 Tuple0~Tuple25，不支持空字段
• Scala 样例类及 Scala 元组：不支持空字段
• 行类型（ROW）：可以认为是具有任意个字段的元组,并支持空字段
• POJO：Flink 自定义的类似于 Java bean 模式的类

1. 辅助类型

   Option、Either、List、Map 等

2. 泛型类型

   Flink 支持所有的 Java 类和 Scala 类。不过如果没有按照上面 POJO 类型的要求来定义， 就会被 Flink 当作泛型类来处理。Flink 会把泛型类型当作黑盒，无法获取它们内部的属性；它 们也不是由 Flink 本身序列化的，而是由 Kryo 序列化的。

   Flink 对 POJO 类型的要求如下：

• 类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；
• 类有一个公共的无参构造方法；
• 类中的所有字段是 public 且非 final 的；或者有一个公共的 getter 和 setter 方法，这些方法需要符合 Java bean 的命名规范。

2.6.3 类型提示（Type Hints）

Flink 还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息， 从而获得对应的序列化器和反序列化器。 但是，由于 Java 中泛型擦除的存在，在某些特殊情况下（比如 Lambda 表达式中），自动提取的信息是不够精细的，这时就需要显式地提供类型信 息，才能使应用程序正常工作或提高其性能。

Flink 专门提供了 TypeHint 类，它可以捕获泛型的类型信息，并且一直记录下来，为运行 时提供足够的信息。我们同样可以通过.returns()方法，明确地指定转换之后的 DataStream 里元 素的类型。

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})

3. 转换算子（Transformation）

数据源读入数据之后，可以使用各种转换算子，将一个或多个 DataStream 转换为 新的 DataStream。

一个 Flink 程序的核心，其实就是所有的转换操作，它们决定了处理的业务逻辑。

3.1 基本转换算子

3.1.1 映射（map）

map 主要用于将数据流中的数据进行转换，形成新的 数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素 。

基于 DataStream调用 map()方法就可以进行转换处理。方法需要传入的参数是 接口 MapFunction 的实现；返回值类型还是 DataStream，不过泛型（流中的元素类型）可能改变。

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;
        }
    }
}

 基于 **DataStream **调用 **map **方法，返回 的其实是一个 **SingleOutputStreamOperator**。  

public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper){}

3.1.2 过滤（filter）

filter 转换操作，对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。

进行 filter 转换之后的新数据流的数据类型与原数据流是相同的。filter 转换需要传入的参 数需要实现 FilterFunction 接口，而 FilterFunction 内要实现 filter()方法，就相当于一个返回布尔类型的条件表达式。

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）

将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生 0 到多个元素。

flatMap 可以认为是“扁平化”（flatten） 和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理 。

flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来 指定输出。输出结果时，调用收集器的.collect()方法；该方法可以多次调用，也可以不调用。

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）

3.2.1 按键分区（keyBy）

在 Flink 中，要做聚合，需要先进行分区； 这个操作就是通过 keyBy 来完成的。

keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑 上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应 着任务槽（task slot）。

在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。如果 key 是POJO 类型，必须要重写 hashCode()方法。

keyBy()方法需要传入一个参数，这个参数指定了一个或一组 key。有很多不同的方法来指 定key：

比如对于 Tuple 数据类型，可以指定字段的位置或者多个位置的组合；对于 POJO 类 型，可以指定字段的名称（String）；另外，还可以传入 Lambda 表达式或者实现一个键选择器 （KeySelector），用于说明从数据中提取 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;
            }
        });
        env.execute();
    }
}

keyBy 得到的结果是 DataStream，而是会将 DataStream 转换为 KeyedStream。KeyedStream 可以认为是“分区流”或者“键控流”，它是对 DataStream 按照 key 的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定 key 的 类型。

3.2.2 简单聚合

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 操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。

调用 KeyedStream 的 reduce 方法时，需要传入一个参数，实现 ReduceFunction 接口。

public interface ReduceFunction<T> extends Function, Serializable {
    T reduce(T value1, T value2) throws Exception;
}

ReduceFunction 接口里需要实现 reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件；所以，对于一组数据，我们可以先取两个进行合并，然后再 将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据， 这也就是 reduce“归约”的含义。

在流处理的底层实现过程中，实际上是将中间“合并的结果” 作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。

reduce 的语义是针对列表进行规约操作，运算规则由 ReduceFunction 中的 reduce 方法来定义，而在 ReduceFunction 内部会维护一个初始值为空的累加器，注意累加器的类型和输入元素的类型相同，当第一条元素到来时，累加器的值更新为第一条元素的值，当新的元 素到来时，新元素会和累加器进行累加操作，这里的累加操作就是 reduce 函数定义的运算规则。然后将更新以后的累加器的值向下游输出。

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();
     }
}

3.2.4 用户自定义函数（UDF）

3.2.4.1 函数类（Function Classes）

对于大部分操作而言，都需要传入一个用户自定义函数(UDF),实现相关操作的接口,来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类， 例如 MapFunction、FilterFunction、ReduceFunction 等。

3.2.4.2 匿名函数（Lambda）

Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的(匿名）类。

对于像 flatMap() 这样的函数，它的函数签名 void flatMap(IN value, Collector) 被 Java 编译器编译成了 void flatMap(IN value, Collector out)，也就是说将 Collector 的泛型信息擦除掉了。这样 Flink 就无法自动推断输出的类型信息了。

在这种情况下，我们需要显式地指定类型信息，否则输出将被视为 Object 类型，这会导致低效的序列化。

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

3.2.4.3 富函数类（Rich Function Classes）

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其 Rich 版本。富函数类一般是以抽象类的形式出现。例如：RichMapFunction、RichFilterFunction、 RichReduceFunction 等。

与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

Rich Function 有生命周期的概念。典型的生命周期方法有：

• open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当 一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调 用。所以像文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的 工作，都适合在 open()方法中完成。

• close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一 些清理工作。

  这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的， 实际工作方法，例如 RichMapFunction 中的 map()，在每条数据到来后都会触发一次调用。

3.2.5 物理分区（Physical Partitioning）

分区（partitioning）：将数据进行重新分布，传递到不同的流分区去进行下一步处理 。

 **Flink **对于经过转换操作之后的 **DataStream**，提供了一系列的底层操作接口，能够帮 我们实现数据流的手动重分区。为了同 **keyBy **相区别，我们把这些操作统称为“物理分区” 操作。

物理分区与 keyBy 另一大区别在于，keyBy 之后得到的是一个 KeyedStream，而物理分区之后结果仍是 DataStream，且流中元素数据类型保持不变。从这一点也可以看出，分区算子并不对数据进行转换处理，只是定义了数据的传输方式。

3.2.5.1 随机分区（shuffle）

DataStream 的.shuffle()方法，将数据随机地分配到下游算子的并行任务中去。

随机分区服从均匀分布（uniform distribution），可以把流中的数据随机打乱，均匀地传递到下游任务分区 ，因为是完全随机的，所以对于同样的输入数据, 每次执行得到的结果也不会相同。

3.2.5.2 轮询分区（Round-Robin）

轮询分区按照先后顺序将数据做依次分发 。

DataStream 的.rebalance()方法，就可以实现轮询重分区。rebalance 使用的是 Round-Robin 负载均衡算法，可以将输入流数据平均分配到下游的并行任务中去。

3.2.5.3 重缩放分区（rescale）

调用rescale()方法时，其实底层也是使用 Round-Robin 算法进行轮询，但是只会将数据轮询发送到下游并行任务的一部分中。

“发牌人”如果有多个，那么 rebalance 的方式是每个发牌人都面向所有人发牌；而 rescale 的做法是分成小团体，发牌人只给自己团体内的所有人轮流发牌。

当下游任务（数据接收方）的数量是上游任务（数据发送方）数量的整数倍时，rescale 的效率明显会更高。比如当上游任务数量是 2，下游任务数量是 6 时，上游任务其中一个分区 的数据就将会平均分配到下游任务的 3 个分区中。

从底层实现上看，rebalance 和 rescale 的根本区别在于任务之间的连接机制不同。rebalance 将会针对所有上游任务（发送数据方）和所有下游任务（接收数据方）之间建立通信通道，这是一个笛卡尔积的关系；而 rescale 仅仅针对每一个任务和下游对应的部分任务之间建立通信 通道，节省了很多资源。

3.2.5.4 广播（broadcast）

经过广播操作之后，数据会在不同的分区都保留一份，可能进行重复处理。 可以通过调用 DataStream 的 broadcast()方法，将输入数据复制并发送到下游算子的所有并行任务中去。

3.2.5.5 全局分区（global）

调用.global()方法，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了 1

3.2.5.6 自定义分区（Custom）

当 Flink 提供的所有分区策略都不能满足用户的需求时，可以通过partitionCustom()方法来自定义分区策略。

在调用时，方法需要传入两个参数，第一个是自定义分区器（Partitioner）对象，第二个是应用分区器的字段，它的指定方式与 keyBy 指定 key 基本一样：可以通过字段名称指定，也可以通过字段位置索引来指定，还可以实现一个 KeySelector。

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class CustomPartitionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 将自然数按照奇偶分区
        env.fromElements(1, 2, 3, 4, 5, 6, 7, 8)
            .partitionCustom(new Partitioner<Integer>() {
                @Override
                public int partition(Integer key, int numPartitions) {
                    return key % 2;
                }
            }, new KeySelector<Integer, Integer>() {
                @Override
                public Integer getKey(Integer value) throws Exception {
                    return value;
                }
            })
            .print().setParallelism(2);
        env.execute();
    }
}

4. 输出算子（Sink）

4.1 连接到外部系统

Flink 内部提供了一致性检查点（checkpoint） 来保障我们可以回滚到正确的状态；但如果我们在处理过程中任意读写外部系统，发生故障后就很难回退到从前了。

为了避免这样的问题，Flink 的 DataStream API 专门提供了向外部写入数据的方法： addSink。与 addSource 类似，addSink 方法对应着一个“Sink”算子，主要就是用来实现与外部系统连接、并将数据提交写入的；Flink 程序中所有对外的输出操作，一般都是利用 Sink 算子完成的。

除去一些 Flink 预实现的 Sink，一般情况下Sink算子的创建是通过调用 DataStream 的.addSink()方法实现的。

stream.addSink(new SinkFunction(…));

addSink 方法需要传入 一个参数，实现的是 SinkFunction 接口。在这个接口中只需要重写一个方法 invoke(),用来将指定的值写入到外部系统中。这个方法在每条数据记录到来时都会调用：

default void invoke(IN value, Context context) throws Exception

 **Flink **官方目前支持的第三方系统连接器： 

除 Flink 官方之外，Apache Bahir 作为给 Spark 和 Flink 提供扩展支持的项目，也实现了一些其他第三方系统与 Flink 的连接器。

4.2 输出到文件

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.3 输出到 Kafka

Flink 与 Kafka 的连接器提供了端到端的精确一次（exactly once）语义保证，这在实际项目中是最高级别的一致性保证。

将数据输出到Kafka：

1. 添加 Kafka 连接器依赖
2. 启动 Kafka 集群
3. 编写输出到 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", "hadoop102:9092");
        DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");
        stream
             .addSink(new FlinkKafkaProducer<String>(
                "clicks",
                 new SimpleStringSchema(),
                 properties
            ));
        env.execute();
    }
}

4.4 输出到Redis

Flink 没有直接提供官方的 Redis 连接器，不过 Bahir 项目还是担任了合格的辅助角色，提供了 Flink-Redis 的连接工具。但版本升级略显滞后，目前连接器版本为 1.0，支持的 Scala 版本最新到 2.11。

测试步骤：

1. 导入的 Redis 连接器依赖

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

1. 启动 Redis 集群

2. 编写输出到 Redis 的示例代码

   连接器为提供了一个 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("hadoop102").build();
         env.addSource(new ClickSource())
         .addSink(new RedisSink<Event>(conf, new MyRedisMapper()));
         env.execute();
    }
}

RedisSink 的构造方法需要传入两个参数：

• JFlinkJedisConfigBase：Jedis 的连接配置

• RedisMapper：Redis 映射类接口，说明怎样将数据转换成可以写入 Redis 的类型

  定义一个 Redis 的映射类，实现 RedisMapper 接口。

public static 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**，每来一条数据就会做一次转换。  

4.5 输出到 Elasticsearch

ElasticSearch 是一个分布式的开源搜索和分析引擎，适用于所有类型的数据。ElasticSearch 有着简洁的 REST 风格的 API，以良好的分布式特性、速度和可扩展性而闻名。

写入数据的 ElasticSearch 的测试步骤如下：

1. 添加 Elasticsearch 连接器依赖

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

1. 启动 Elasticsearch 集群

2. 编写输出到 Elasticsearch 的示例代码

   与 RedisSink 类 似 ， 连 接 器 也 为 我 们 实 现 了 写 入 到 Elasticsearch 的 SinkFunction——ElasticsearchSink。区别在于，这个类的构造方法是私有（private）的，我们 需要使用 ElasticsearchSink 的 Builder 内部静态类，调用它的 build()方法才能创建出真正的 SinkFunction。 而 Builder 的构造方法中又有两个参数：

• httpHosts：连接到的 Elasticsearch 集群主机列表
• elasticsearchSinkFunction：这并不是我们所说的 SinkFunction，而是用来说明具体处 理逻辑、准备数据向 Elasticsearch 发送请求的函数

4.6 输出到MySQL（JDBC）

写入数据的 MySQL 的测试步骤如下：

1. 添加依赖

<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>

1. 启动 MySQL，在 database 库下建表 clicks

mysql> create table clicks(
-> user varchar(20) not null,
-> url varchar(100) not null);

1. 编写输出到 MySQL 的示例代码

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SinkToMySQL {
     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));
         stream.addSink(
             JdbcSink.sink(
                 "INSERT INTO clicks (user, url) VALUES (?, ?)",
                (statement, r) -> {
                 statement.setString(1, r.user);
                statement.setString(2, r.url);
                 },
                JdbcExecutionOptions.builder()
                     .withBatchSize(1000)
                    .withBatchIntervalMs(200)
                     .withMaxRetries(5)
                     .build(),
                 new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                     .withUrl("jdbc:mysql://localhost:3306/userbehavior")
                    // 对于 MySQL 5.7，用"com.mysql.jdbc.Driver"
                    .withDriverName("com.mysql.cj.jdbc.Driver")
                    .withUsername("username")
                    .withPassword("password")
                    .build()
                )
            );
        env.execute();
    }
}

1. 运行代码，用客户端连接 MySQL，查看是否成功写入数据

4.7 自定义Sink输出

Flink 为我们提供了通用的 SinkFunction 接口和对应的 RichSinkDunction 抽象类，只要实现它，通过简单地调用 DataStream 的.addSink()方法就可以自定义写入任何外部存储。

在实现 SinkFunction 的时候，需要重写的一个关键方法 invoke()，在这个方法中可以实现将流里的数据发送出去的逻辑。