flink 学习笔记 — 编程模型

flink 抽象分层结构

    flink 作为流式处理框架，不仅具有高效的流数据和批数据处理性能，具有针对开发人员使用的高效的底层API，同时有方便分析人员使用的 table/sql API。下图中可见，越上层的 API 使用越简单，但是本身的表达能力也越弱，越下层的 API 开发和学习成本越高，表达能力也越强。因此，flink API 可以适合不同层次的使用人员快速上手。

flink分层架构图.png

• Process Function 是最底层的 API，对于开发人员来讲，它也是使用最为灵活的一层。它提供了丰富的接口给开发人员使用，在功能实现上更符合开发的要求。它可以嵌入到 DataStream API 中。这里可以实现复杂的计算。
• DataSet / DataStream API 是封装好的，其中 DataSet API 主要是为处理批量数据（bounded data）准备的，而DataStream API 是专为处理流式数据（unbounded data）而存在。事实上，这个阶段的 API 处理能够满足大部分的开发需求。因此，这里将主要对这层的 API 做详细的介绍。
• Flink提供的最高级抽象是SQL，可以使用类似关系型数据库的运算操作，主要为开发了解较少的分析人员使用。

DataStream API 介绍

数据流执行概况

    flink 中的流式数据是无边界（unbounded）的，在执行过程中，flink 程序本身主要是由数据流和算子共同组成。在整个过程中包括 source、transform、sink 三个部分。

• source：连接数据源，flink 中数据源主要包括：textFile、socket、kafka、flink 本身的 sink 等等。
• transform：转化算子，flink 中算子主要用于数据的逻辑处理，例如：FlatMap、Map、sum 等等。

• sink：连接数据落地，flink sink 决定将数据写入到哪里，主要包括：file、socket、kafka、mysql 等等。

  
  flink-stream执行图.jpg

    由上图可以发现，一个完整的 flink 程序，必须包括 source、transform、sink 三部分，并且在最后开启 execute 才能保证程序的正常执行。通常，transform 的一个算子和数据是一一对应关系，但是有时会把一些算子链在一起(chain在一起)，从而提高任务执行的性能。

数据流并行计算

    我们知道，flink 程序是一个分布式的并行计算框架，因此，在 flink 程序执行的过程中，是并行计算的。这里引用 flink 官网上描述并行计算的流程图：

flink-stream并行计算流程图.jpg

• 事实上，在此执行图中，除了 sink 算子的并行度（一个算子子任务的数量）为 1，其他算子的并行度均为 2。
• 一对一流（例如，在上图中的Source和map（） 算子之间）保存数据元的分区和排序。这意味着map（） 算子的subtask [1] 将以与Source 算子的subtask [1]生成的顺序相同的顺序看到相同的数据元。
• 重新分配流（在上面的map（）和keyBy / window之间，以及 keyBy / window和Sink之间）重新分配流。每个 算子子任务将数据发送到不同的目标子任务，具体取决于所选的转换。实例是 keyBy（） （其通过散列Keys重新分区），广播（） ，或Rebalance （） （其重新分区随机地）。在重新分配交换中，数据元之间的排序仅保存在每对发送和接收子任务中（例如，map（）的子任务[1] 和子任务[2]keyBy / window）。因此，在此示例中，保存了每个Keys内的排序，但并行性确实引入了关于不同Keys的聚合结果到达接收器的顺序的非确定性。

    下面我们采用一个简单的单词统计的例子进行说明：

package test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class NcWordCount {
    public static void main(String []args) throws Exception {

        // 初始化执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置计算并行度
        env.setParallelism(1);

        // 连接 source
        DataStreamSource<String> lines = env.socketTextStream("localhost", 9000);

        // 转换操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> r = lines.flatMap(new TokenZer())
                .keyBy(0)
                .timeWindow(Time.seconds(10))
                .sum(1);

        // sink
        // r.writeAsText("out");
        r.print();
        env.execute("test");
    }

    static class TokenZer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] words = value.split(" ");

            for (String word:words) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

• 这里使用 socket 数据源，因此我们在执行时需要在本地启动一个 socket 连接。

nc -l 9000

• 这里采用 sink 到控制台的方式进行测试，当然，我们可以调用 writeAsText() 函数，将我们的结果 sink 到文件中。
• 实际上，flink 程序执行中可以通过 env.setParallelism() 来设置计算并行度，如果没有指定，那在程序执行过程中会根据所在的机器的核心数来确定执行并行度。而设置合理的计算并行度可以有效提升程序的计算性能，当然，计算并行度也要和本地资源做一个 balance。下图是默认并行度的执行结果，这里本地机器是 8 核心，因此在默认并行度下，计算结果 sink 到文件时，会生成 8 个文件。 当把并行度设置为 1 时(如代码)，最终会有一个结果文件。

flink-stream计算并行度.jpg

DataSet API 介绍

     在了解了 DataStream API 的编程模式后，对 DataSet 能更快的上手。事实上，与 spark 相反，flink 流式计算中，以流式数据为主，批量数据处理是建立在流式处理的基础上。在编程过程中，DataSet API 的使用跟 DataStream API 的使用有着及其相似的流程。

• 在初始化执行环境时，既可以使用 StreamExecutionEnvironment， 又可以使用 ExecutionEnvironment。
• 执行过程中产生的数据集都是以 DataSet 的形式存在。

package test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount {
    public static void main(String []args) throws Exception {
        // 初始化环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 数据源
        DataSet<String> dataSet = env.fromElements("hello word spark java scala");

        // 做一次聚合操作
        AggregateOperator<Tuple2<String, Integer>> sum = dataSet.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String values, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] value = values.split(" ");
                for (String val : value) {
                    out.collect(new Tuple2<>(val, 1));
                }
            }
        }).groupBy(0)
                .sum(1);

//        count.print().setParallelism(3);
        sum.print();

//        env.execute("flink-test");

    }
}

总结

     本文主要对 flink 的编程模型(DataStream/DataSet API)进行了简单的介绍，当然，主要还是处于 API 的使用阶段，在后续的 flink 使用中会通过源码分析进行更详细的介绍。同时，在 Stream API 中，我们看到了 window 窗口的影子，窗口的使用是 flink 引擎中重要的一环，接下来也会有更详尽的描述。