3.0、Flink Vs SparkStreaming

3.0.1、Concept

Spark围绕着速度、易用性、和复杂分析而生。
Flink围绕着高性能、稳定性和准确性，快速处理数据

3.0.2、Flink VS Spark

Flink
1、Flink计算模型是基于算子的流模型，实时处理流数据；在Flink中批处理被认为是特殊的一种流处理。
2、没有最小数据的延迟、带有一个独立于实际编程接口的优化器
3、基于流水线执行，处理速度比Spark要快；通过原生闭环算子，flink中的机器学习和图处理速度更快。
4、API比spark要少
5、编程语言提供了Java和Scala
6、Flink数据流运行的时候配置最少，可以实现低延迟和高吞吐；可以在批流两种处理模式下使用相同的算法
7、内部有自己的内存管理器，不同于GC;可以通过显式管理内存和消除内存峰值。
8、Flink基于Chandy-Lamport分布式快照的容错机制，而且是轻量级的、有助于保持高吞吐提供强大一致性保证
9、窗口标准基于记录或者用户自定义的。一般只处理一次记录来消除重复

Spark:
1、Spark计算模型基于微批处理模型，使用三方集群管理器操作；通过RDD抽象集来完成批数据处理。
2、和Flink相比，Spark有更高的延迟
3、在Spark中，作业需要手动优化，处理时间长
4、API层面很容易使用。高阶API提供了多种编程语言，如R,Scala,Python,Java
5、Spark中的迭代处理并不是基于原生的，在系统外实现为普通的for循环。
6、数据流在Spark中表示为DAG
7、Spark具有自动内存管理，提供了可配置的内存管理。
8、Spark中的窗口标准是基于时间的。

3.1、Architectural principles

3.1.1、特点

• 批流统一
• 支持高吞吐、低延迟‘高性能的流处理
• 支持有状态计算的Exactly-Once语义
• 支持高度灵活的窗口操作，如基于事件时间，基于会话时间，基于处理时间等。
• 支持Backpressure功能的持续流模型
• 支持基于轻量级Snapshot实现的容错
• 支持迭代计算
• 支持程序自动优化：即避免特定情况下的shuffle、排序等操作，中间结果有必要缓存

  3.1.2、组件栈

  

3.1.4、架构体系

3.1.5、编程模型

Flink提供了不同的抽象级别以开发流式或批处理应用：

1. 最底层提供了有状态流。它将通过过程函数(Process Function)嵌入到DataStream API中。它允许用户可以自由处理来自一个或多个流数据的事件，并使用一致、容错的状态。除此之外，用户可以注册事件时间和处理事件回调，从而程序可以实现复杂的计算。
2. DataStream/DataSet API 是Flink提供的核心API。DataSet处理有界的数据集，DataStream处理有界或无界的数据流。用户可以通过各种方法(map/flatmap/window/keyBy/sum/max/min/avg/join)等将数据进行转换/计算
3. Table API：是以表为中心的声明式SQL,其中表可能会动态变化(在表达流数据时)。Table API提供了例如select、project、join、group-by、aggregate等操作，使用起来更加简洁。
4. Flink提供了最高层级的抽象是SQL.这一层抽象在语法与表达能力上与Table API类似，但是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SQL查询可以直接在Table API定义的表上执行

   3.2、Deploy

   3.2.1、Flink原生安装

   ```shell

   1.下载flink安装包，下载地址：https://flink.apache.org/downloads.html

   2.上传flink安装包到Linux服务器上

   3.解压flink安装包

   tar -xvf flink-1.11.1-bin-scala_2.12.tgz -C /bigdata/

4.修改conf目录下的flink-conf.yaml配置文件

5.指定jobmanager的地址

jobmanager.rpc.address: localhost

6.指定taskmanager的可用槽位的数量

taskmanager.numberOfTaskSlots: 2

7.修改conf目录下的workers配置文件，指定taskmanager的所在节点

localhost

8.执行启动脚本

bin/start-cluster.sh

9.执行jps查看进程

jps

10.访问JobManager的web管理界面，端口8081

<a name="UY936"></a>
### 3.2.2 Ambari集成Flink
1、下载Flink二进制包<br />wget [http://www.us.apache.org/dist/flink/flink-1.7.2/flink-1.7.2-bin-hadoop27-scala_2.11.tgz](http://www.us.apache.org/dist/flink/flink-1.7.2/flink-1.7.2-bin-hadoop27-scala_2.11.tgz)<br />2、解压缩
```shell
tar -zxvf flink-1.7.2-bin-hadoop27-scala_2.11.tgz
mv flink-1.7.2-bin-hadoop27-scala_2.11 /opt/flink
chown -R hdfs:hdfs /opt/flink

3、环境变量配置

vi /etc/profile
 export FLINK_HOME=/opt/flink
 export PATH=$FLINK_HOME/bin:$PATH
:wq
source /etc/profile

4、下载安装Ambari-Flink软件包

git clone https://github.com/abajwa-hw/ambari-flink-service.git
mv ambari-flink-server  /var/lib/ambari-server/resources/stacks/HDP/$VERSION/services/FLINK

5、重启ambari-server
ambari-server restart
6、安装flink


7、修改yarn参数

<property>
      <name>yarn.client.failover-proxy-provider</name>              
      <value>org.apache.hadoop.yarn.client.RequestHedgingRMFailoverProxyProvider</value>
</property>

8、重启yarn
9、测试

flink run --jobmanager yarn-cluster \
-yn 1 \
-ytm 768 \
-yjm 768 \
/opt/flink/examples/batch/WordCount.jar \
--input hdfs://hdp1:8020/user/hdfs/demo/input/word  \
--output hdfs://hdp3:8020/user/hdfs/demo/output/wc/

10、flink sql client

sql-client.sh embedded

11、错误解决
java.lang.NoClassDefFoundError: com/sun/jersey/core/util/FeaturesAndProperties
解决方案:在flink的lib包里面添加如下的两个jar包
12、wordcount

/**
*
*从一个Socket端口中实时的读取数据，然后实时统计相同单词出现的次数，该程序会一直运行
*启动程序前先使用nc -l 8888启动一个socket用来发送数据
*/
public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        //创建流式计算的ExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //调用Source，指定Socket地址和端口
        DataStream<String> lines = env.socketTextStream("localhost", 10088);
        DataStream<Tuple2<String, Integer>> words = lines.
                flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String line, Collector<Tuple2<String, Integer>> collector)
                            throws Exception {
                        String[] words = line.split(" ");
                        for (String word : words) {
                            collector.collect(Tuple2.of(word, 1));
                        }
                    }
                });
        //按照key分组并聚合
        DataStream<Tuple2<String, Integer>> result = words.keyBy(0).sum(1);
        //将结果打印到控制台
        result.print();
        //执行
        env.execute("StreamingWordCount");
    }
}

3.3、Submit mode

3.3.1、Session mode

流程：
1、该模式首先会预分配资源，根据指定的资源参数提前初始化一个Flink集群，常驻于Yarn中（本文中都是基于Yarn的资源管理器），并启动一个JobManager和固定数量的TaskManager。
2、这个时候可以直接提交作业运行，节省了申请资源分配的开销。
总结：
1、这种模式下所有的作业共享这些固定的资源，而且作业之间不能隔离，会出现资源竞争的情况。
2、当一个TM发生故障宕机后，那么所有在这个节点上的作业都会失败。
3、当提交运行越来越多的Job后，JM的负载压力也会越高。
场景：
因此该模式通常用于部署一些运行时间短，对延迟要求性不高的任务。

3.3.2、Per-Job mode

该模式见名知意，即每个作业Job提交到Yarn上形成自己独立的Flink集群，即拥有作业自己的JobManager和TaskManager。
1、因此在提交作业的时候，由于需要单独的申请资源，因此启动延迟会相对比较高；
2、但是作业之间的资源是隔离的，也就是说作业之间是相互独立的，不会相互造成影响，一个Job的TaskManager失败不会影响到其他的作业运行；
3、另外JobManager也是相互独立的，负载不会相当那么高；
4、当job完成后，分配的资源就会得到释放回收。不像Session模式资源属于常占的。
场景：
基于以上几点特性，Per-Job模式一般用于提交长时间运行的作业。通常生产环境使用该种模式

3.3.3、Per-Job Vs Session Mode

通常来讲，无论是Per-job模式还是Session模式，都需要一个客户端入口向Yarn发起提交请求，用来提交作业。那么在main方法执行之前，直到env.execute()方法，客户端还需要做一些工作：
1、需要获取作业相关依赖
2、分析执行环境并获取逻辑计划，通常来讲就是从StreamGraph到JobGraph的生成。
3、将依赖和JobGraph上传到集群
当以上工作做完之后，Flink Runtime将会被触发，并通过env.execute方法执行job。
那么这个时候思考一下，如果在初始化分析并获取依赖阶段，如果依赖东西非常多且很大，那么就会占用更多的带宽；如果业务执行逻辑非常复杂，那么在解析转换逻辑计划生成JobGraph的时候就需要更多的内存和CPU。因此基于这种场景下，客户端资源将会成为一个瓶颈。
当然上述描述的场景问题在两种提交模式中都会存在。因此出现了第三种提交模式，即Application mode

3.3.4、Application mode（Flink1.11+）

1、针对客户端做大量的前期工作的问题：client所做的一些事情都被转移到JM上来做了，也就是说程序主方法在集群中进行执行，入口点在ApplicationClusterEntryPoint。那么开发者只需要负责发起提交请求。
另外如果在主方法内在多个env.execute()/executeAsync()被调用，那么在Application 模式下，这些作业都被看作是同一个应用程序(如果是以per-job模式提交，那么会启动多个应用程序)。这么看来的话，application模式其实是per-job模式和session模式的折中方案。
2、针对依赖传输造成的带宽占用问题：flink作业需要依赖的包有flink-dist.jar(发布用的)、扩展包(位于{FLINK_HOME}/lib)、插件库(${FLINK_HOME/plugin})、用户jar包。那么针对这些依赖包，我们可以提前上传到HDFS上，然后通过yarn.provided.lib.dirs参数来指定存储路径。
通过这种方式，所有的flink作业不用重复上传依赖包，这样就可以直接从hdfs上拉取，而且NodeManager也会缓存这些依赖包进一步加速作业提交。
使用application模式提交demo

bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dtaskmanager.numberOfTaskSlots=2 \
-Dparallelism.default=10 \
-Dyarn.application.name="MyFlinkApp" \
-Dyarn.provided.lib.dirs="hdfs://myhdfs/flink-common-deps/lib;hdfs://myhdfs/flink-common-deps/plugins"

3.3.5、 Session mode Vs Per-Job mode Vs Application mode

3.4、Source

3.4.1、Flink预定义Source操作

在flink中，source主要负责数据的读取。
flink预定义的source中又分为并行source(主要实现ParallelSourceFunction接口)和非并行source(主要实现了SourceFunction接口)
附上官网相关的说明：
you can always write your own custom sources by implementing the SourceFunction for non-parallel sources, or by implementing the ParallelSourceFunction interface or extending the RichParallelSourceFunction for parallel sources

3.4.1.1、基于File的数据源

1.1底层原理

   Flink在进行文件读取的时候会启动两个subTask,一个subTask用来监听，另外一个subTask用来数据读取。且监听的subTask是属于非并行的，即并行度为1，而数据读取的subTask是属于并行的 ，通常并行度和job的并行度是一致的

readTextFile(path):逐行读取文本文件，将符合TextInputFormat规范的文件，作为字符串返回
readFile(fileInputFormat, path)：根据指定的文件输入格式读取
readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)：上面两个方法底层其实调用的就是该方法

1.2 代码

public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> textSource = env.readTextFile("/data/GitCode/flink/src/main/resources/word.txt");
        //如果不设置并行度的话，这里会跟机器的核数一致
        int parallelism = textSource.getParallelism();
        System.out.println("textFile并行度:" + parallelism);
        textSource.print();
        env.execute("TextFileSourceDemo");
    }

3.4.1.2、基于Socket的数据源(一般用于学习,测试)

2.1、底层原理

通常调用socketTextStream方法来接收socket数据，元素可以用分隔符分开，该方法底层实现了SourceFunction接口，是属于非并行source

2.2、代码

 public static void main(String[] args) throws Exception {
        //并行度和系统核数一致
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //非并行的source,只有1个并行度
        DataStreamSource<String> lines = env.socketTextStream(args[0], Integer.parseInt(args[1]));
        int parallelism = lines.getParallelism();
        System.out.println("socketTextStream并行度："+parallelism);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap((String line, Collector<Tuple2<String, Integer>> out) -> Arrays.stream(line.split(" ")).forEach(word -> out.collect(Tuple2.of(word, 1)))).returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = wordAndOne.keyBy(t -> t.f0).sum(1);
        summed.print();
        env.execute("WordCount");
    }

3.4.1.3、基于集合的数据源(学习,测试)

从集合中创建一个数据流,集合中所有的元素类型都是一致的.其底层调用了SourceFunction接口，属于非并行source。注意:fromParallelCollection方法是属于并行的,底层调用了RichParallelSourceFunction接口

1. fromCollection(Collection)

   val list=List(1,2,3,4,5,6,7,8,9)
   val inputStream=env.fromCollection(list)

2. fromCollection(Iterator, Class)

   val iterator=Iterator(1,2,3,4)
   val inputStream=env.fromCollection(iterator)

3. fromElements(T …)
   从一个给定的对象序列中创建一个数据流，所有的对象必须是相同类型的。

   DataStreamSource<Integer> fromSource = env.fromElements(1, 2, 3, 4);
   int parallelism = fromSource.getParallelism();
   System.out.println("fromElements并行度:" + parallelism);

4. generateSequence(from, to)
   从给定的间隔中并行地产生一个数字序列。

val inputStream = env.generateSequence(1,10)

3.4.1.4、基于Kafka的数据源(重点)

这里给出简单的入门示例，实际工作场景中需要考虑到容错、数据一致性、分区发现、结合水印实现一些窗口操作等功能

public class kafkaSource {
    private static Properties properties = new Properties();
    private static final String BOOTSTRAPLIST = "hdp1:6667,hdp2:6667";
    private static final String GROUPID = "metric-group";
    private static final String TOPIC = "metric";
    static {
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAPLIST);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUPID);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    }
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> kafkaSource = env.addSource(new FlinkKafkaConsumer<String>(TOPIC, new SimpleStringSchema(), properties));
        kafkaSource.print();
        env.execute();
    }
}

3.4.2、Flink自定义Source操作

3.4.2.1、非并行Source

即实现SourceFunction接口

public class CustomNonParallelSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        List<String> list = new ArrayList<>();
        list.add("Spark");
        list.add("Hadoop");
        list.add("Flink");
        list.add("HBase");
        list.add("Es");
        SingleOutputStreamOperator<String> singleOutputStreamOperator = env.addSource(new MyWordSource(list)).returns(String.class);
        System.out.println("自定义非并行source对应的并行度:" + singleOutputStreamOperator.getParallelism());
        singleOutputStreamOperator.print();
        env.execute();
    }
    static class MyWordSource implements SourceFunction<String> {
        private List<String> words;
        private boolean isStop = false;
        public MyWordSource() {
        }
        public MyWordSource(List<String> words) {
            this.words = words;
        }
        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            if (words != null && words.size() > 0) {
                while (!isStop) {
                    words.forEach(word -> ctx.collect(word));
                }
            }
        }
        @Override
        public void cancel() {
            isStop = true;
        }
    }
}

3.4.2.2、并行Source

通过实现ParallelSourceFunction接口或者继承RichParallelSourceFunction来定义并行source。
其中RichParallelSourceFunction功能更加丰富些，可以获取上下文信息，提供open和close方法等更加强大的功能
简单的实时流

public class CustomSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<String> words = new ArrayList<>();
        words.add("spark");
        words.add("hadoop");
        words.add("flink");
//        DataStreamSource<String> wordsSource = env.addSource(new MyWordSource(words));
        SingleOutputStreamOperator singleOutputStreamOperator = env.addSource(new MyWordParallelSource(words)).returns(String.class);
//      如果不设置并行度，则和系统核数一致
        System.out.println("自定义并行source对应parallelism:" + singleOutputStreamOperator.getParallelism());
        singleOutputStreamOperator.print();
        env.execute("CustomSource");
    }
    static class MyWordParallelSource extends RichParallelSourceFunction {
        private boolean isStop = false;
        private List<String> words;
        public MyWordParallelSource() {
        }
        /**
         * 通常在该方法做一些初始化的工作，如创建数据库连接等
         * @param parameters
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
        }
        /**
         * 释放资源
         * @throws Exception
         */
        @Override
        public void close() throws Exception {
            super.close();
        }
        public MyWordParallelSource(List<String> word) {
            this.words = word;
        }
        @Override
        public void run(SourceContext ctx) throws Exception {
            int numberOfParallelSubtasks = getRuntimeContext().getNumberOfParallelSubtasks();
            while (!isStop) {
                if (words != null && words.size() > 0) {
                    words.forEach(word -> ctx.collect(numberOfParallelSubtasks + "--->" + word));
                }
                Thread.sleep(2000);
            }
        }
        /**
         * 该方法对无界数据流有作用
         */
        @Override
        public void cancel() {
            isStop = true;
        }
    }
    static class MyWordSource extends RichSourceFunction<String> {
        private List<String> words;
        private boolean flag = true;
        public MyWordSource(List<String> words) {
            this.words = words;
        }
        /**
         * 产生数据，用sourceContext将数据发生出去
         *
         * @param sourceContext
         * @throws Exception
         */
        @Override
        public void run(SourceContext sourceContext) throws Exception {
            //获取当前subTask的运行上下文
            RuntimeContext runtimeContext = getRuntimeContext();
            int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
            while (flag) {
                if (words != null && words.size() > 0) {
                    words.forEach(word -> sourceContext.collect(indexOfThisSubtask + "--->" + word));
                }
                Thread.sleep(2000);
            }
        }
        /**
         * 停止source,对于无界数据流有用
         */
        @Override
        public void cancel() {
            flag = false;
        }
    }
}

3.4.2.3、自定义source读取mysql

public class MysqlSource extends RichSourceFunction<Student> {
    private static Logger logger = LoggerFactory.getLogger(MysqlSource.class);
    private static final String URL ="jdbc:mysql://localhost:3306/hibernate?useUnicode=true&characterEncoding=UTF-8";
    private static final String USERNAME="root";
    private static final String PASSWORD="123456";
    PreparedStatement preparedStatement;
    private Connection connection;
    private static Connection getConnection() {
        Connection connection = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            connection = DriverManager.getConnection(URL, USERNAME, PASSWORD);
        } catch (Exception e) {
            logger.error("创建Mysql连接异常,信息:[{}]",e);
            e.printStackTrace();
        }
        return  connection;
    }
    /**
     * open()方法中建立连接，这样不用每次invoke的时候建立和释放连接
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String sql = "select * from student";
        preparedStatement = this.connection.prepareStatement(sql);
    }
    /**
     * DataStream调用一次run方法用来获取数据
     * @param ctx
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        ResultSet resultSet =
                preparedStatement.executeQuery();
        while (resultSet.next()){
            Student student = new Student(
                    resultSet.getInt("id"),
                    resultSet.getString("name").trim(),
                    resultSet.getString("password").trim(),
                    resultSet.getInt("age"));
            ctx.collect(student);
        }
    }
    @Override
    public void cancel() {
    }
    /**
     * 关闭时释放连接
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        super.close();
        if(connection!=null){
            connection.close();
        }
        if(preparedStatement!=null){
            preparedStatement.close();
        }
    }
}

3.4.3、总结

3.5、Transformation

一个流的转换操作将会应用在一个或者多个流上面，这些转换操作将流转换成一个或者多个输出流，将这些转换算子组合在一起来构建一个数据流图。大部分的数据流转换操作都是基于用户自定义函数udf。udf函数打包了一些业务逻辑并定义了输入流的元素如何转换成输出流的元素。像MapFunction这样的函数，将会被定义为类，这个类实现了Flink针对特定的转换操作暴露出来的接口。
DataStream API针对大多数数据转换操作提供了转换算子，这里将转换算子分为四类：

• 基本转换算子：将会作用在数据流中的每一条单独的数据上
• KeyedStream转换算子：在数据有key的情况下，对数据应用转换算子。
• 多流转换算子：合并多条流为一条流或者将一条流分割为多条流
• 分布式转换算子：将重新组织流里面的事件。

下面将分别讲解这四种算子类型。

3.5.1、基本转换算子

基本转换算子会针对流中的每一个单独的事件做处理，也就是说每一个输入数据会产生一个输出数据。单值转换，数据分割，数据过滤，都是基本转换操作的典型例子.

3.5.1.1、map

类型转换：DataStream → DataStream
作用：map算子将每一个输入的事件传送到用户自定义的一个mapper，这个mapper只返回一个输出事件
实现原理：底层调用的是transform方法，真正实现逻辑封装在StreamMap类中，该类底层实现了OneInputStreamOperator接口，在processElement方法中实现具体的功能

public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper, TypeInformation<R> outputType) {
        return transform("Map", outputType, new StreamMap<>(clean(mapper)));
    }

操作案例:
功能：将字符串转为大写
方式一：直接调用原生的map算子

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
//方式一：调用原生自带的map算子
SingleOutputStreamOperator<String> wordSource = socketTextStream.map(word -> word.toUpperCase());
wordSource.print();
env.execute();

方式二：调用底层的transform算子重定义实现

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
//方式二：
SingleOutputStreamOperator<String> wordSource = socketTextStream.transform("MyMap", TypeInformation.of(String.class), new StreamMap<>(String::toUpperCase));
wordSource.print();
env.execute();

方式三：继承实现类自定义实现

/**
     * 类似于StreamMap操作
     */
static class MyStreamMap extends AbstractStreamOperator<String> implements OneInputStreamOperator<String, String> {
    @Override
    public void processElement(StreamRecord<String> element) throws Exception {
      String elementValue = element.getValue();
      output.collect(element.replace(elementValue.toUpperCase()));
    }
}
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
SingleOutputStreamOperator<String> wordSource = socketTextStream.transform("MyStreamMap", TypeInformation.of(String.class), new MyStreamMap());
wordSource.print();
env.execute();

方式四：实现RichMapFunction类

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
SingleOutputStreamOperator<String> wordSource = socketTextStream.map(new RichMapFunction<String, String>() {
            /**
             * 在构造对象之后，执行map方法之前执行一次
             * 通常用于初始化工作，例如连接创建等
             * @param parameters
             * @throws Exception
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }
            /**
             * 在关闭subtask之前执行一次，例如做一些释放资源的工作
             * @throws Exception
             */
            @Override
            public void close() throws Exception {
                super.close();
            }
            @Override
            public String map(String s) throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                return indexOfThisSubtask + ":" + s.toUpperCase();
            }
        });
wordSource.print();
env.execute();

3.5.1.2、flatmap

类型转换：DataStream → DataStream
作用：和map算子比较类似，不同之处在于针对每一个输入事件flatMap会生成0个、1个或多个输出元素。
实现原理：同map实现原理一样，底层调用的是transform方法，真正实现逻辑封装在StreamMap类中，该类底层实现了OneInputStreamOperator接口，在processElement方法中实现具体的功能
操作案例：
功能：将一行记录按照空格分割，并转换为大写
方式一：直接调用原生算子，其他方式同map算子

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
SingleOutputStreamOperator<String> flatMap = socketTextStream.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String s, Collector<String> collector) throws Exception {
        Arrays.stream(s.split(" ")).forEach(word -> collector.collect(word.toUpperCase()));
    }
});
flatMap.print();
env.execute("FlatMapDemo");
//如输入 a b c 
//最后输出
A
B
C

3.5.1.3、filter

类型转换：DataStream → DataStream
作用：通过在每个输入事件上对一个布尔条件进行求值来过滤掉一些元素，然后将剩下的元素继续发送。一个true的求值结果会把输入事件保留下来并发送到输出，而如果求值结果为false,则输入事件会被抛弃掉
实现原理：底层调用的是transform方法，真正实现逻辑封装在StreamFilter类中，该类底层实现了OneInputStreamOperator接口，在processElement方法中实现具体的功能
操作案例：
功能：仅保留以A开头的字符串

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.filter(word->word.startsWith("A")).print();
env.execute("FilterDemo");
//如输入
bca
ad
Ab
Ac
//最后得到的结果为
Ab
Ac

3.5.2、KeyedStream转换算子(通常和window配合使用)

3.5.2.1、KeyBy

类型转换：DataStream → KeyedStream
作用：其实就是按照key进行分组，根据不同的key分配到不同的分区中，所有具有相同key的事件会在同一个slot中进行处理
实现原理：

1. 根据key进行hash运算得到hash值，即变量keyHash
2. 使用murmurHash公式进行计算，即调用murmurHash(keyHash)
3. 对murmurHash和最大并行度进行求余运算得到变量keyGroupId = murmurHash%maxParallelism
4. 计算并行度索引Index = (keyGroupId*parallelism)/maxParallelism

   public static int assignKeyToParallelOperator(Object key, int maxParallelism, int parallelism) {
        Preconditions.checkNotNull(key, "Assigned key must not be null!");
        return computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
    }
   //1.计算key的hashCode值
   public static int assignToKeyGroup(Object key, int maxParallelism) {
   Preconditions.checkNotNull(key, "Assigned key must not be null!");
   return computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
   }
   //2.使用murmurHash进行计算，并和最大并行度进行求余运算
   public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {
   return MathUtils.murmurHash(keyHash) % maxParallelism;
   }
   //3.计算获取索引
   public static int computeOperatorIndexForKeyGroup(int maxParallelism, int parallelism, int keyGroupId) {
        return keyGroupId * parallelism / maxParallelism;
    }

   操作案例：
   功能：基于key进行求和统计
   方式一：针对Tuple类型，使用下标进行分组 ```java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource socketTextStream = env.socketTextStream(“localhost”, 8888);

socketTextStream.map(word-> Tuple3.of(word.split(“ “)[0],word.split(“ “)[1],Double.valueOf(word.split(“ “)[2]))) .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class,String.class,Double.class)) .keyBy(t->t.f0+t.f1 ) //这里不支持 In many cases lambda methods don’t provide enough information for automatic type extraction .sum(2) .print();

env.execute(“KeyByDemo”);

方式二：针对Bean，使用字段名称进行分<br />注意：该种方式下，Bean中的字段不能私有化，且必须要有无参构造器<br />WordCount Bean
```java
public class WordCount {
  public String word;
  public Integer count;
  public WordCount() {
  }
  public WordCount(String word, Integer count) {
    this.word = word;
    this.count = count;
  }
  public static WordCount of(String word, Integer count) {
    return new WordCount(word, count);
  }
  public String getWord() {
    return word;
  }
  public void setWord(String word) {
    this.word = word;
  }
  public Integer getCount() {
    return count;
  }
  public void setCount(Integer count) {
    this.count = count;
  }
  @Override
  public String toString() {
    return "WordCount{" +
      "word='" + word + '\'' +
      ", count=" + count +
      '}';
  }
}
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(words->WordCount.of(words,1))
  .returns(WordCount.class)
  .keyBy(WordCount::getWord)
  .sum("count")
  .print();

方式三：实现KeySelector类

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(word -> Tuple3.of(word.split(" ")[0], word.split(" ")[1], Double.valueOf(word.split(" ")[2])))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, Double.class))
  .keyBy(new KeySelector<Tuple3<String, String, Double>, String>() {
    @Override
    public String getKey(Tuple3<String, String, Double> value) throws Exception {
      return value.f0 + value.f1;
    }
  })
  .sum(2)
  .print();
env.execute("KeyByDemo");

3.5.2.2、fold(已过时)

类型转换：KeyedStream → DataStream
作用：合并当前元素和前一次折叠操作的结果，并产生一个新的值，返回的流中包含每一次折叠的结果，而不是只返回最后一次折叠的结果
实现原理：还是调用transform方法，逻辑封装在StreamGroupedFold类中，该类实现了OneInputStreamOperator和OutputTypeConfigurable接口，其中有一个value的判断逻辑

OUT value = values.value();
//如果前一次折叠的结果不为空，则和当前结果进行合并，并更新，否则取当前结果(即初始化)
if (value != null) {
  OUT folded = userFunction.fold(outTypeSerializer.copy(value), element.getValue());
  values.update(folded);
  output.collect(element.replace(folded));
} else {
  OUT first = userFunction.fold(outTypeSerializer.copy(initialValue), element.getValue());
  values.update(first);
  output.collect(element.replace(first));
}

操作案例：
功能：按照key进行sum求和

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.flatMap((String words, Collector<Tuple2<String,Integer>> out)-> Arrays.stream(words.split(" ")).forEach(word-> out.collect(Tuple2.of(word,1)) ))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class,Integer.class))
  .keyBy(t->t.f0)
  .fold(Tuple2.of("", 0), new FoldFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
    @Override
    public Tuple2<String, Integer> fold(Tuple2<String, Integer> tuple2, Tuple2<String, Integer> o) throws Exception {
      //tuple2为中间存储的结果或初始化值
      o.f1 = o.f1 + tuple2.f1;
      return o;
    }
  }).print();
env.execute("FoldDemo");
//如输入
a 10
a 20
a 30
//最后得到
a 60

3.5.2.3、sum/min/minBy/max/maxBy

类型转换：KeyedStream → DataStream
作用：分组数据流上的滚动聚合操作，即按照key进行求最大值/最小值/求和等操作。这里只给出min和minBy的区别，其他都类似。
其中min和minBy的区别是min返回的是一个最小值，而minBy返回的是其字段中包含最小值的元素，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
通俗来讲当有多个字段时，min的作用是对其中指定字段进行分组取最小值，而不关心其他字段，结果中只会取第一次出现的字段；而minBy按照指定字段返回最小值所在的记录，其中有一个布尔值参数，当为false时，则取最小值的最新记录
实现原理：底层调用的还是transform方法，具体的实现逻辑封装在StreamGroupedReduce类中，该类也是实现了OneInputStreamOperator接口

IN value = element.getValue();
IN currentValue = values.value();
//可以看到最底层调用的其实是用户自定义函数
if (currentValue != null) {
  IN reduced = userFunction.reduce(currentValue, value);
  values.update(reduced);
  output.collect(element.replace(reduced));
} else {
  values.update(value);
  output.collect(element.replace(value));
}

操作案例：
功能：一条记录中包含3个字符，用逗号分隔，最后一个字符为整数类型，按照切分后的第一个字段为key取最小值
方式一：调用minBy算子，first参数为false

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(word -> Tuple3.of(word.split(",")[0], word.split(",")[1], Integer.valueOf(word.split(",")[2])))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, Integer.class))
  .keyBy(t -> t.f0)
  .minBy(2,false)
  .print();
env.execute("MinMinByDemo");
//如输入
hadoop,spark,1
hadoop,hadoop,2
hadoop,spark,10  -->根据hadoop为key，取最小值
spark,hadoop,50
spark,test,40
spark,teste,100
spark,hadoop,200 -->根据spark为key,取最小值
//结果输出
4> (hadoop,spark,1)
4> (hadoop,spark,1)
4> (hadoop,spark,1)
1> (spark,hadoop,50)
1> (spark,test,40)
1> (spark,test,40)
1> (spark,test,40)

方式二：调用minBy算子，first参数为true

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(word -> Tuple3.of(word.split(",")[0], word.split(",")[1], Integer.valueOf(word.split(",")[2])))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, Integer.class))
  .keyBy(t -> t.f0)
  .minBy(2,false)
  .print();
//如输入
spark,hadoop,50
spark,test,40
spark,teste,100
spark,hadoop,200
flink,hadoop,400
spark,Hadoop,400
flink,spark,300
flink,java,500
flink,spark,400
flink,es,100
spark,java,300
spark,flink,300
flink,java,50
//最终输入
1> (spark,hadoop,50)
1> (spark,test,40)
1> (spark,test,40)
1> (spark,test,40)
4> (flink,hadoop,400)
1> (spark,test,40)
4> (flink,spark,300)
4> (flink,spark,300)
4> (flink,spark,300)
4> (flink,es,100)
1> (spark,test,40)
1> (spark,test,40)
4> (flink,java,50) --即获取最小值所在的记录

方式三：调用min算子

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(t -> Tuple2.of(t.split(",")[0], Integer.valueOf(t.split(",")[1])))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class))
  .keyBy(t -> t.f0)
  .min(1)
  .print();
env.execute();

3.5.2.4、reduce

类型转换：KeyedStream → DataStream
作用：合并当前元素和上次聚合的结果，产生一个新值，和fold算子比较类似
实现原理：和fold算子一致
操作案例：
功能：简单的单词统计

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.flatMap((String words, Collector<Tuple2<String,Integer>> out)-> Arrays.stream(words.split(" ")).forEach(word->out.collect(Tuple2.of(word,1))))
  .returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class,Integer.class))
  .keyBy(t->t.f0)
  .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
    @Override
    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> stringIntegerTuple2, Tuple2<String, Integer> t1) throws Exception {
      stringIntegerTuple2.f1 = stringIntegerTuple2.f1 + t1.f1;
      return stringIntegerTuple2;
    }
  })
  .print();
env.execute("ReduceByDemo");
//如输入
a
b
c
d
e
a
f
d
a
d
c
//最终结果为
3> (a,1)
1> (b,1)
2> (c,1)
3> (d,1)
1> (e,1)
3> (a,2)
1> (f,1)
3> (d,2)
3> (a,3)
3> (d,3)
2> (c,2)

3.5.3、多流转换算子

3.5.3.1、union

类型：多个DataStream→ DataStream
作用：对两个或两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream
实现原理：底层会重新创建一个DataStream对象，并把原来老的多个DataStream对象放到一个list表中，进行校验类型是否一致，其中DataStream构造器参数有一个UnionTransformation类，底层继承了Transformation类

List<Transformation<T>> unionedTransforms = new ArrayList<>();
unionedTransforms.add(this.transformation);
for (DataStream<T> newStream : streams) {
  if (!getType().equals(newStream.getType())) {
    throw new IllegalArgumentException("Cannot union streams of different types: "
                                       + getType() + " and " + newStream.getType());
  }
  unionedTransforms.add(newStream.getTransformation());
}
return new DataStream<>(this.environment, new UnionTransformation<>(unionedTransforms));

操作案例：
功能：合并元素

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<Integer> source2 = env.fromElements(11, 22, 33, 44, 55);
DataStream<Integer> unionSource = source1.union(source2);
unionSource.print();
env.execute();

3.5.3.2、connect

类型：DataStream,DataStream → ConnectedStreams
作用：连接两个保持各自类型的数据流，两个流被connect之后，只是被放在了同一个流中，但是内部仍然是保持着各自的数据类型和形式，并且相互独立

实现原理：内部重新创建了一个ConnectedStreams对象，且该对象内部重写了大部分常用算子，包含上面总结的算子，但是算子名称和基本算子不太一致，如map算子称为Co-Map,flatMap算子成为Co-Flat Map
操作案例：
功能：字符串和int类型DataStream进行关联，并增加不同的前缀

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<String> streamStr = env.fromElements("hello", "str", "good");
ConnectedStreams<Integer, String> connect = stream1.connect(streamStr);
connect.map(new CoMapFunction<Integer, String, String>() {
  @Override
  public String map1(Integer value) throws Exception {
    return "hhhh1"+value;
  }
  @Override
  public String map2(String value) throws Exception {
    return "11111"+value;
  }
}).print();
env.execute("ConnectDemo");
}
//最后结果为
2> hhhh13
3> 11111hello
4> hhhh11
1> hhhh12
1> 11111good
4> 11111str
4> hhhh15
3> hhhh14

3.5.4、分布式转换算子

针对于flink的分区，有五种类型：shuffle,轮询，global，广播和自定义；8种分区器(这里只给出6种，剩下的两种比较简单，有兴趣的读者可以自行阅读)

3.5.4.1、8种分区器

3.5.4.2、五种分区类型

1、shuffle

Random

类型转换：DataStream → DataStream
作用：将数据随机的分配到下游算子的并行任务中去。
实现原理：底层创建ShufflePartitioner对象，该对象继承了StreamPartitioner。具体逻辑如下：

//根据并行度个数随机分配
public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
        return random.nextInt(numberOfChannels);
    }

操作案例:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//非并行source
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
//获取读取source时的并行度为1
int sourceParallelism = socketTextStream.getParallelism();
System.out.println("读取socket source并行度:" + sourceParallelism);
SingleOutputStreamOperator<String> mapped = socketTextStream.map(new RichMapFunction<String, String>() {
  @Override
  public String map(String value) throws Exception {
    int index = getRuntimeContext().getIndexOfThisSubtask();
    return index + "-->" + value;
  }
}).setParallelism(2);
//获取map之后的并行度，这里得到手动设置的2
int mappedParallelism = mapped.getParallelism();
System.out.println("经过map之后的并行度：" + mappedParallelism);
DataStream<String> shuffle = mapped.shuffle();
//获取shuffle之后的并行度,这里的值和系统核数保持一致
int shuffleParallelism = shuffle.getParallelism();
System.out.println("经过shuffle之后的并行度:" + shuffleParallelism);
shuffle.addSink(new RichSinkFunction<String>() {
  @Override
  public void invoke(String value, Context context) throws Exception {
    System.out.println(getRuntimeContext().getIndexOfThisSubtask() + "--->" + value);
  }
});
env.execute();

2、轮询

Round-Robin

类型转换：DataStream → DataStream
作用：使用Round-Robin负载均衡算法将输入流平均分配到随后的并行运行的任务中去
实现原理：即调用rebalance算子，该算子底层由RebalancePartitioner类实现具体的分配逻辑，该类同样继承自StreamPartitioner，具体逻辑如下：

//这里真正实现轮询
@Override
    public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
        nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;
        return nextChannelToSendTo;
    }

操作案例:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(new RichMapFunction<String, String>() {
  @Override
  public String map(String value) throws Exception {
    int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
    return indexOfThisSubtask + "-->map后-->" + value;
  }
})
  .rebalance()
  .addSink(new RichSinkFunction<String>() {
    @Override
    public void invoke(String value, Context context) throws Exception {
      int index = getRuntimeContext().getIndexOfThisSubtask();
      System.out.println(value + "--->rebalance--->" + index);
    }
  });
env.execute();
//最后结果
3-->map后-->spark--->rebalance--->1
0-->map后-->hadoop--->rebalance--->2
1-->map后-->flink--->rebalance--->3
2-->map后-->java--->rebalance--->0
3-->map后-->es--->rebalance--->2
0-->map后-->good--->rebalance--->3
1-->map后-->yes--->rebalance--->0
2-->map后-->are--->rebalance--->1

3、Rescale

类型转换：DataStream → DataStream
作用：使用的也是round-robin算法，但只会将数据发送到接下来的并行运行的任务中的一部分任务中。本质上，当发送者任务数量和接收者任务数量不一样时，rescale分区策略提供了一种轻量级的负载均衡策略。如果接收者任务的数量是发送者任务的数量的倍数时，rescale操作将会效率更高
实现原理：调用rescale算子，底层由RescalePartitioner分区器实现，该分区器继承自StreamPartitioner，具体处理逻辑如下：

@Override
    public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
        if (++nextChannelToSendTo >= numberOfChannels) {
            nextChannelToSendTo = 0;
        }
        return nextChannelToSendTo;
    }

操作案例:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
socketTextStream.map(new RichMapFunction<String, String>() {
  @Override
  public String map(String value) throws Exception {
    int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
    return indexOfThisSubtask + "-->map后-->" + value;
  }
})
  .rescale()
  .addSink(new RichSinkFunction<String>() {
    @Override
    public void invoke(String value, Context context) throws Exception {
      int index = getRuntimeContext().getIndexOfThisSubtask();
      System.out.println(value + "--->rescale--->" + index);
    }
  });
env.execute();