Flink流处理核心编程

╰半夏微凉° 2022-10-23 15:26 246阅读 0赞

# Flink流处理核心编程方法 #

### 目录 ###

*  Flink流处理核心编程方法
 *   *  一、Environment
     *  二、Source
     *   *  2.1 准备工作
         *  2.2 从Java的集合中读取数据
         *  2.3 从文件读取数据
         *  2.4 从Socket读取数据
         *  2.5 从Kafka读取数据
         *  2.6 自定义Source
     *  三、Transform
     *   *  3.1 map
         *  3.2 flatMap
         *  3.3 filter
         *  3.4 keyBy
         *  3.5 shuffle
         *  3.6 split和select
         *  3.7 connect
         *  3.8 union
         *  3.9 简单滚动聚合算子
         *  3.10 reduce
         *  3.11 process
         *  3.12 对流重新分区的几个算子
     *  四、Sink
     *   *  4.1 KafkaSink
         *  4.2 RedisSink
         *  4.3 ElasticsearchSink
         *  4.4 自定义Sink
     *  五、执行模式(Execution Mode)
     *   *  5.1 选择执行模式
         *  5.2 配置BATH执行模式
         *  5.3 有界数据用STREAMING和BATCH的区别

和其他所有的计算框架一样，Llink也有一些基础的开发步骤以及基础，核心的API，从开发步骤的角度来讲，主要分为四大部分  
![在这里插入图片描述][20210302204212557.png]

## 一、Environment ##

![在这里插入图片描述][20210302204223909.png]

Flink Job在提交执行计算时，需要首先建立和Flink框架之间的联系，也就指的是当前的flink运行环境，只有获取了环境信息，才能将task调度到不同的taskManager执行。而这个环境对象的获取方式相对比较简单

// 批处理环境
    ExecutionEnvironment benv = ExecutionEnvironment.getExecutionEnvironment();
    // 流式数据处理环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

## 二、Source ##

![在这里插入图片描述][20210302204254901.png]

Flink框架可以从不同的来源获取数据，将数据提交给框架进行处理, 我们将获取数据的来源称之为数据源(**Source**)。

### 2.1 准备工作 ###

1.  导入注解工具依赖, 方便生产POJO类

<dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.16</version>
        <scope>provided</scope>
    </dependency>

1.  准备一个WaterSensor类方便演示

package com.atguigu.flink.java.chapter_5;
    
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    
    /** * @Author lzc * @Date 2020/12/8 22:22 * 水位传感器：用于接收水位数据 * * id:传感器编号 * ts:时间戳 * vc:水位 */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class WaterSensor { 
        private String id;
        private Long ts;
        private Integer vc;
    }

### 2.2 从Java的集合中读取数据 ###

一般情况下，可以将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用。这里的数据结构采用集合类型是比较普遍的。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.util.Arrays;
    import java.util.List;
    
    public class Flink01_Source_Collection { 
        public static void main(String[] args) throws Exception { 
            List<WaterSensor> waterSensors = Arrays.asList(
              new WaterSensor("ws_001", 1577844001L, 45),
              new WaterSensor("ws_002", 1577844015L, 43),
              new WaterSensor("ws_003", 1577844020L, 42));
    
            // 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env
              .fromCollection(waterSensors)
             .print();
            env.execute();
        }
    }

### 2.3 从文件读取数据 ###

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class Flink02_Source_File { 
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env
              .readTextFile("input")
              .print();
    
            env.execute();
        }
    }

说明:

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

<dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.3</version>
        <scope>provided</scope>
    </dependency>

### 2.4 从Socket读取数据 ###

// 1. 创建流式执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2. 读取文件
    DataStreamSource<String> lineDSS = env.socketTextStream("hadoop162", 9999);
    // 3. 转换数据格式
    SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
            .flatMap((String line, Collector<String> words) -> { 
                Arrays.stream(line.split(" ")).forEach(words::collect);
            })
            .returns(Types.STRING)
            .map(word -> Tuple2.of(word, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG));
    // 4. 分组
    KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
            .keyBy(t -> t.f0);
    // 5. 求和
    SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
            .sum(1);
    // 6. 打印
    result.print();
    // 7. 执行
    env.execute();

### 2.5 从Kafka读取数据 ###

添加相应的依赖

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>1.11.2</version>
    </dependency>

参考代码

import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
    
    import java.util.Properties;
    
    public class Flink03_Source_Kafka { 
        public static void main(String[] args) throws Exception { 
    
            // 0.Kafka相关配置
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092");
            properties.setProperty("group.id", "Flink01_Source_Kafka");
            properties.setProperty("auto.offset.reset", "latest");
    
            // 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env
              .addSource(new FlinkKafkaConsumer<>("sensor", new SimpleStringSchema(), properties))
              .print("kafka source");
    
            env.execute();
        }
    }

### 2.6 自定义Source ###

大多数情况下，前面的数据源已经能够满足需要，但是难免会存在特殊情况的场合，所以flink也提供了能**自定义数据源**的方式.

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.Socket;
    import java.nio.charset.StandardCharsets;
    
    public class Flink04_Source_Custom { 
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env
              .addSource(new MySource("hadoop102", 9999))
              .print();
    
            env.execute();
        }
    
        public static class MySource implements SourceFunction<WaterSensor> { 
            private String host;
            private int port;
            private volatile boolean isRunning = true;
            private Socket socket;
    
            public MySource(String host, int port) { 
                this.host = host;
                this.port = port;
            }
    
    
            @Override
            public void run(SourceContext<WaterSensor> ctx) throws Exception { 
                // 实现一个从socket读取数据的source
                socket = new Socket(host, port);
                BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
                String line = null;
                while (isRunning && (line = reader.readLine()) != null) { 
                    String[] split = line.split(",");
                    ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
                }
            }
    
            /** * 大多数的source在run方法内部都会有一个while循环, * 当调用这个方法的时候, 应该可以让run方法中的while循环结束 */
    
            @Override
            public void cancel() { 
                isRunning = false;
                try { 
                    socket.close();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    /* sensor_1,1607527992000,20 sensor_1,1607527993000,40 sensor_1,1607527994000,50 */

## 三、Transform ##

转换算子可以把一个或多个DataStream转成一个新的DataStream.程序可以把多个复杂的转换组合成复杂的数据流拓扑.

### 3.1 map ###

作用：将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素  
![在这里插入图片描述][20210302204802783.png]

参数： lambda表达式或**MapFunction**实现类  
返回： DataStream → DataStream  
示例：得到一个新的数据流: 新的流的元素是原来流的元素的平方

*  匿名内部类对象

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class Flink01_TransForm_Map_Anonymous { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            env
              .fromElements(1, 2, 3, 4, 5)
              .map(new MapFunction<Integer, Integer>() { 
                  @Override
                  public Integer map(Integer value) throws Exception { 
                      return value * value;
                  }
              })
              .print();
    
            env.execute();
        }
    }

*  Lambda表达式表达式

env
      .fromElements(1, 2, 3, 4, 5)
      .map(ele -> ele * ele)
      .print();

*  静态内部类

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class Flink01_TransForm_Map_StaticClass { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            env
              .fromElements(1, 2, 3, 4, 5)
              .map(new MyMapFunction())
              .print();
    
            env.execute();
        }
    
        public static class MyMapFunction implements MapFunction<Integer, Integer> { 
    
            @Override
            public Integer map(Integer value) throws Exception { 
                return value * value;
            }
        }
        
    }

*  Rich…Function类  
    **所有Flink函数类都有其Rich版本**。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。也有意味着提供了更多的，更丰富的功能。例如：**RichMapFunction**

import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class Flink01_TransForm_Map_RichMapFunction { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(5);
    
            env
              .fromElements(1, 2, 3, 4, 5)
              .map(new MyRichMapFunction()).setParallelism(2)
              .print();
    
            env.execute();
        }
    
        public static class MyRichMapFunction extends RichMapFunction<Integer, Integer> { 
            // 默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次
            @Override
            public void open(Configuration parameters) throws Exception { 
                System.out.println("open ... 执行一次");
            }
    
            // 默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次
            @Override
            public void close() throws Exception { 
                System.out.println("close ... 执行一次");
            }
    
            @Override
            public Integer map(Integer value) throws Exception { 
                System.out.println("map ... 一个元素执行一次");
                return value * value;
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

1.  默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次, 而且先被调用
2.  默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次, 而且是最后被调用
3.  getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态. 开发人员在需要的时候**自行调用获取运行时上下文对象**.

### 3.2 flatMap ###

作用： 消费一个元素并产生零个或多个元素  
![在这里插入图片描述][20210302205141986.png]

参数： FlatMapFunction实现类  
返回： DataStream → DataStream  
示例：

*  匿名内部类写法

// 新的流存储每个元素的平方和3次方
    env
      .fromElements(1, 2, 3, 4, 5)
      .flatMap(new FlatMapFunction<Integer, Integer>() { 
          @Override
          public void flatMap(Integer value, Collector<Integer> out) throws Exception { 
              out.collect(value * value);
              out.collect(value * value * value);
          }
      })
      .print();

*  Lambda表达式写法

env
      .fromElements(1, 2, 3, 4, 5)
      .flatMap((Integer value, Collector<Integer> out) -> { 
          out.collect(value * value);
          out.collect(value * value * value);
      }).returns(Types.INT)
      .print();

说明: 在使用Lambda表达式表达式的时候, 由于泛型擦除的存在, 在运行的时候无法获取泛型的具体类型, 全部当做Object来处理, 及其低效, 所以Flink要求当参数中有泛型的时候, 必须明确指定泛型的类型.

### 3.3 filter ###

作用： 根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃  
![在这里插入图片描述][20210302205134187.png]

参数： FlatMapFunction实现类  
返回： DataStream → DataStream  
示例：

*  匿名内部类写法

// 保留偶数, 舍弃奇数
    env
      .fromElements(10, 3, 5, 9, 20, 8)
      .filter(new FilterFunction<Integer>() { 
          @Override
          public boolean filter(Integer value) throws Exception { 
              return value % 2 == 0;
          }
      })
      .print();

*  Lambda表达式写法

env
      .fromElements(10, 3, 5, 9, 20, 8)
      .filter(value -> value % 2 == 0)
      .print();

### 3.4 keyBy ###

作用：  
把流中的数据分到不同的分区中.具有**相同key的元素**会分到同一个分区中.一个分区中可以有多重不同的key.  
在内部是使用的**hash分区**来实现的.  
![在这里插入图片描述][20210302205242790.png]

参数： Key选择器函数: interface KeySelector<IN, KEY>

注意: 什么值不可以作为KeySelector的Key:

*  没有覆写hashCode方法的POJO, 而是依赖Object的hashCode. 因为这样分组没有任何的意义: 每个元素都会得到一个独立无二的组. 实际情况是:可以运行, 但是分的组没有意义.
 *  任何类型的数组  
    返回  
    DataStream → KeyedStream  
    示例
 *  匿名内部类写法

// 奇数分一组, 偶数分一组
    env
      .fromElements(10, 3, 5, 9, 20, 8)
      .keyBy(new KeySelector<Integer, String>() { 
          @Override
          public String getKey(Integer value) throws Exception { 
              return value % 2 == 0 ? "偶数" : "奇数";
          }
      })
      .print();
    env.execute();

*  Lambda表达式写法

env
      .fromElements(10, 3, 5, 9, 20, 8)
      .keyBy(value -> value % 2 == 0 ? "偶数" : "奇数")
      .print();

### 3.5 shuffle ###

作用： 把流中的元素随机打乱. 对同一个组数据, 每次执行得到的结果都不同.

参数： 无  
返回： DataStream → DataStream  
示例：

env
      .fromElements(10, 3, 5, 9, 20, 8)
      .shuffle()
      .print();
    env.execute();

### 3.6 split和select ###

已经过时, 在1.12中已经被移除  
作用： 在某些情况下，我们需要将数据流根据某些特征拆分成两个或者多个数据流，给不同数据流增加标记以便于从流中取出。 split用于给流中的每个元素添加标记. select用于根据标记取出对应的元素, 组成新的流.  
![在这里插入图片描述][20210302205413828.png]  
![在这里插入图片描述][20210302205418865.png]

参数：  
split参数: interface OutputSelector  
select参数: 字符串

返回  
split: SingleOutputStreamOperator -> SplitStream  
slect: SplitStream -> DataStream  
示例

*  匿名内部类写法

// 奇数一个流, 偶数一个流
    SplitStream<Integer> splitStream = env
      .fromElements(10, 3, 5, 9, 20, 8)
      .split(new OutputSelector<Integer>() { 
          @Override
          public Iterable<String> select(Integer value) { 
              return value % 2 == 0
                ? Collections.singletonList("偶数")
                : Collections.singletonList("奇数");
          }
      });
    splitStream
      .select("偶数")
      .print("偶数");
    
    splitStream
      .select("奇数")
      .print("奇数");
    env.execute();

*  Lambda表达式写法

// 奇数一个流, 偶数一个流
    SplitStream<Integer> splitStream = env
      .fromElements(10, 3, 5, 9, 20, 8)
      .split(value -> value % 2 == 0
        ? Collections.singletonList("偶数")
        : Collections.singletonList("奇数"));
    splitStream
      .select("偶数")
      .print("偶数");
    
    splitStream
      .select("奇数")
      .print("奇数");
    env.execute();

### 3.7 connect ###

作用：

在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。

Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。  
![在这里插入图片描述][20210302205513292.png]

参数： 另外一个流  
返回： DataStream\[A\], DataStream\[B\] -> ConnectedStreams\[A,B\]  
示例

DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
    DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");
    // 把两个流连接在一起: 貌合神离
    ConnectedStreams<Integer, String> cs = intStream.connect(stringStream);
    cs.getFirstInput().print("first");
    cs.getSecondInput().print("second");
    env.execute();

注意:

1.  两个流中存储的数据类型可以不同
2.  只是机械的合并在一起, 内部仍然是分离的2个流
3.  只能2个流进行connect, 不能有第3个参与

### 3.8 union ###

作用：对**两个或者两个以上**的DataStream进行union操作，产生一个包含所有DataStream元素的**新DataStream**  
![在这里插入图片描述][20210302205604153.png]

示例

DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5);
    DataStreamSource<Integer> stream2 = env.fromElements(10, 20, 30, 40, 50);
    DataStreamSource<Integer> stream3 = env.fromElements(100, 200, 300, 400, 500);
    
    // 把多个流union在一起成为一个流, 这些流中存储的数据类型必须一样: 水乳交融
    stream1
      .union(stream2)
      .union(stream3)
      .print();

connect与 union 区别：

1.  union之前两个或多个流的类型必须是一样，connect可以不一样
2.  connect只能操作两个流，union可以操作多个。

### 3.9 简单滚动聚合算子 ###

常见的滚动聚合算子

*  sum, min,max
 *  minBy,maxBy  
    作用：KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream  
    参数：如果流中存储的是POJO或者scala的样例类, 参数使用字段名. 如果流中存储的是元组, 参数就是位置(基于0…)  
    返回：KeyedStream -> SingleOutputStreamOperator  
    示例
 *  示例1

DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4, 5);
    KeyedStream<Integer, String> kbStream = stream.keyBy(ele -> ele % 2 == 0 ? "奇数" : "偶数");
    kbStream.sum(0).print("sum");
    kbStream.max(0).print("max");
    kbStream.min(0).print("min");

*  示例2

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

**注意:**

分组聚合后, 理论上只**能取分组字段和聚合结果**, 但是Flink**允许其他的字段**也可以取出来, 其他字段默认情况是取的是这个**组内第一个元素**的字段值

*  示例3:

**注意:**

maxBy和minBy可以指定当出现相同值的时候,其他字段**是否取第一个**. true表示取第一个, false表示取与最大值(最小值)同一行的.

### 3.10 reduce ###

**作用**：  
一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

参数：interface ReduceFunction  
返回：KeyedStream -> SingleOutputStreamOperator  
示例

*  匿名内部类写法

ArrayList<WaterSensor> waterSensors = new ArrayList<>();
    waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
    waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
    waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
    waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
    waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
    
    KeyedStream<WaterSensor, String> kbStream = env
      .fromCollection(waterSensors)
      .keyBy(WaterSensor::getId);
    
    kbStream
      .reduce(new ReduceFunction<WaterSensor>() { 
          @Override
          public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception { 
              System.out.println("reducer function ...");
              return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
          }
      })
      .print("reduce...");
    
    env.execute();

*  Lambda表达式写法

kbStream
      .reduce((value1, value2) -> { 
          System.out.println("reducer function ...");
          return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
      })
      .print("reduce...");

注意:

1.  聚合后结果的类型, 必须和原来流中元素的类型保持一致!

### 3.11 process ###

**作用**  
process算子在Flink算是一个比较底层的算子, 很多类型的流上都可以调用, 可以从流中获取更多的信息(不仅仅数据本身)

示例1: 在keyBy之前的流上使用  
注意：在keyBy之前使用，如果想得到全局的结果，需要设置并行度为1。KeyBy之后使用使用将流处理为1。

env
      .fromCollection(waterSensors)
      .process(new ProcessFunction<WaterSensor, Tuple2<String, Integer>>() { 
          @Override
          public void processElement(WaterSensor value,
                                     Context ctx,
                                     Collector<Tuple2<String, Integer>> out) throws Exception { 
              out.collect(new Tuple2<>(value.getId(), value.getVc()));
          }
      })
      .print();

示例2: 在keyBy之后的流上使用

env
      .fromCollection(waterSensors)
      .keyBy(WaterSensor::getId)
      .process(new KeyedProcessFunction<String, WaterSensor, Tuple2<String, Integer>>() { 
          @Override
          public void processElement(WaterSensor value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { 
              out.collect(new Tuple2<>("key是:" + ctx.getCurrentKey(), value.getVc()));
          }
      })
      .print();

### 3.12 对流重新分区的几个算子 ###

*  KeyBy  
    先按照key分组, 按照key的双重hash来选择后面的分区
 *  shuffle  
    对流中的元素随机分区
 *  reblance  
    对流中的元素平均分布到每个区.当处理倾斜数据的时候, 进行性能优化
 *  rescale  
    同 rebalance一样, 也是平均循环的分布数据. 但是要比rebalance更高效, 因为rescale不需要通过网络, 完全走的"管道"  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]

## 四、Sink ##

![在这里插入图片描述][20210304140239351.png]

Sink有下沉的意思，在Flink中所谓的Sink其实可以表示为将数据存储起来的意思，也可以将范围扩大，表示将处理完的数据发送到指定的存储系统的输出操作.

之前我们一直在使用的print方法其实就是一种Sink，只不过是将结果sink到终端控制台上输出。

public DataStreamSink<T> print(String sinkIdentifier) { 
       PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier, false);
       return addSink(printFunction).name("Print to Std. Out");
    }

Flink内置了一些Sink, 除此之外的Sink需要用户自定义!  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][20210304140309735.png]

### 4.1 KafkaSink ###

*  添加Kafka Connector依赖

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>1.11.2</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.75</version>
    </dependency>

*  启动Kafka集群
 *  Sink到Kafka的示例代码

import com.alibaba.fastjson.JSON;
    import com.atguigu.flink.java.chapter_5.WaterSensor;
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
    
    import java.util.ArrayList;
    
    public class Flink01_Sink_Kafka { 
        public static void main(String[] args) throws Exception { 
            ArrayList<WaterSensor> waterSensors = new ArrayList<>();
            waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
            waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
            waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
            waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
            waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
            env
              .fromCollection(waterSensors)
              .map(JSON::toJSONString)
              .addSink(new FlinkKafkaProducer<String>("hadoop102:9092", "topic_sensor", new SimpleStringSchema()));
    
    
            env.execute();
        }
    }

在linux启动一个消费者, 查看是否收到数据

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_sensor

### 4.2 RedisSink ###

*  添加Redis Connector依赖

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

*  启动Redis服务器
 *  Sink到Redis的示例代码

import com.alibaba.fastjson.JSON;
    import com.atguigu.flink.java.chapter_5.WaterSensor;
    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;
    
    import java.util.ArrayList;
    
    public class Flink02_Sink_Redis { 
        public static void main(String[] args) throws Exception { 
            ArrayList<WaterSensor> waterSensors = new ArrayList<>();
            waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
            waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
            waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
            waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
            waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
    
            // 连接到Redis的配置
            FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
              .setHost("hadoop102")
              .setPort(6379)
              .setMaxTotal(100)
              .setTimeout(1000 * 10)
              .build();
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
            env
              .fromCollection(waterSensors)
              .addSink(new RedisSink<>(redisConfig, new RedisMapper<WaterSensor>() { 
                  /* key value(hash) "sensor" field value sensor_1 {"id":"sensor_1","ts":1607527992000,"vc":20} ... ... */
    
                  @Override
                  public RedisCommandDescription getCommandDescription() { 
                      // 返回存在Redis中的数据类型 存储的是Hash, 第二个参数是外面的key
                      return new RedisCommandDescription(RedisCommand.HSET, "sensor");
                  }
    
                  @Override
                  public String getKeyFromData(WaterSensor data) { 
                      // 从数据中获取Key: Hash的Key
                      return data.getId();
                  }
    
                  @Override
                  public String getValueFromData(WaterSensor data) { 
                      // 从数据中获取Value: Hash的value
                      return JSON.toJSONString(data);
                  }
              }));
    
            env.execute();
        }
    }

Redis查看是否收到数据  
![在这里插入图片描述][20210304140542253.png]

redis-cli --raw

注意:  
发送了5条数据, redis中只有2条数据. 原因是hash的field的重复了, 后面的会把前面的覆盖掉

### 4.3 ElasticsearchSink ###

*  添加Elasticsearch Connector依赖

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
        <version>1.12.0</version>
    </dependency>

*  启动Elasticsearch集群
 *  Sink到Elasticsearch的示例代码

import com.alibaba.fastjson.JSON;
    import com.atguigu.flink.java.chapter_5.WaterSensor;
    import org.apache.flink.api.common.functions.RuntimeContext;
    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.elasticsearch6.ElasticsearchSink;
    import org.apache.http.HttpHost;
    import org.elasticsearch.action.index.IndexRequest;
    import org.elasticsearch.client.Requests;
    import org.elasticsearch.common.xcontent.XContentType;
    
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.List;
    
    public class Flink03_Sink_ES { 
        public static void main(String[] args) throws Exception { 
            ArrayList<WaterSensor> waterSensors = new ArrayList<>();
            waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
            waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
            waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
            waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
            waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
    
            List<HttpHost> esHosts = Arrays.asList(
              new HttpHost("hadoop102", 9200),
              new HttpHost("hadoop103", 9200),
              new HttpHost("hadoop104", 9200));
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
            env
              .fromCollection(waterSensors)
              .addSink(new ElasticsearchSink.Builder<WaterSensor>(esHosts, new ElasticsearchSinkFunction<WaterSensor>() { 
    
                  @Override
                  public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) { 
                      // 1. 创建es写入请求
                      IndexRequest request = Requests
                        .indexRequest("sensor")
                        .type("_doc")
                        .id(element.getId())
                        .source(JSON.toJSONString(element), XContentType.JSON);
                      // 2. 写入到es
                      indexer.add(request);
                  }
              }).build());
    
            env.execute();
        }
    }

Elasticsearch查看是否收到数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]

注意  
如果出现如下错误:  
![在这里插入图片描述][20210304140650226.png]

添加log4j2的依赖:

<dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-to-slf4j</artifactId>
        <version>2.14.0</version>
    </dependency>

*  如果是无界流, 需要配置bulk的缓存

esSinkBuilder.setBulkFlushMaxActions(1);

### 4.4 自定义Sink ###

如果Flink没有提供给我们可以直接使用的连接器，那我们如果想将数据存储到我们自己的存储设备中，怎么办？（目前flink中提供的sink已经基本涵盖了所有常用的数据存储介质）

我们自定义一个到Mysql的Sink

*  在mysql中创建数据库和表

create database test;
    use test;
    CREATE TABLE `sensor` (
      `id` varchar(20) NOT NULL,
      `ts` bigint(20) NOT NULL,
      `vc` int(11) NOT NULL,
      PRIMARY KEY (`id`,`ts`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

*  导入Mysql驱动

<dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.49</version>
    </dependency>

*  写到Mysql的自定义Sink示例代码

import com.atguigu.flink.java.chapter_5.WaterSensor;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    import java.util.ArrayList;
    
    public class Flink04_Sink_Custom { 
        public static void main(String[] args) throws Exception { 
            ArrayList<WaterSensor> waterSensors = new ArrayList<>();
            waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
            waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
            waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
            waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
            waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
            env.fromCollection(waterSensors)
              .addSink(new RichSinkFunction<WaterSensor>() { 
    
                  private PreparedStatement ps;
                  private Connection conn;
    
                  @Override
                  public void open(Configuration parameters) throws Exception { 
                      conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false", "root", "aaaaaa");
                      ps = conn.prepareStatement("insert into sensor values(?, ?, ?)");
                  }
    
                  @Override
                  public void close() throws Exception { 
                      ps.close();
                      conn.close();
                  }
    
                  @Override
                  public void invoke(WaterSensor value, Context context) throws Exception { 
                      ps.setString(1, value.getId());
                      ps.setLong(2, value.getTs());
                      ps.setInt(3, value.getVc());
                      ps.execute();
                  }
              });
    
    
            env.execute();
        }
    }

## 五、执行模式(Execution Mode) ##

Flink在**1.12.0上对流式API新增一项特性**:可以根据你的使用情况和Job的特点, 可以**选择不同的运行时执行模式**(runtime execution modes)，目前还存在一些Bug，但后期会解决掉的。

流式API的传统执行模式我们称之为**STREAMING 执行模式, 这种模式一般用于无界流**, 需要持续的在线处理

1.12.0新增了一个**BATCH执行模式**, 这种执行模式在执行方式上类似于MapReduce框架. 这种执行模式一般用于**有界数据**.

默认是使用的STREAMING 执行模式

### 5.1 选择执行模式 ###

**BATCH**执行模式仅仅用于**有界数据**, 而**STREAMING** 执行模式可以用在有界数据和无界数据.  
一个公用的规则就是: 当你处理的数据是有界的就应该使用**BATCH**执行模式, 因为它**更加高效**. 当你的数据是**无界的, 则必须使用STREAMING 执行模式**, 因为只有这种模式才能处理持续的数据流.

### 5.2 配置BATH执行模式 ###

执行模式有3个选择可配:

1.  STREAMING(默认)
2.  BATCH
3.  AUTOMATIC

*  通过命令行配置

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

*  通过代码配置

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

建议: **不要**在运行时配置(代码中配置), 而是使用命令行配置, 引用这样会灵活: 同一个应用即可以用于无界数据也可以用于有界数据

### 5.3 有界数据用STREAMING和BATCH的区别 ###

STREAMING模式下, 数据是**来一条输出一次结果**.  
BATCH模式下, 数据处理完之后, **一次性输出结果**.  
下面展示WordCount的程序**读取文件内容**在不同执行模式下的执行结果对比:

*  流式模式  
    // 默认流式模式, 可以不用配置

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 5]

*  批处理模式

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

![在这里插入图片描述][20210304141413678.png]

*  自动模式

env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

![在这里插入图片描述][20210304141426841.png]

注意:  
批模式下, 如果单词**只有1个**, 则**不会输出**, 原因是因为没有保存状态. 关于状态的概念后面再讲.  
![在这里插入图片描述][20210304141442463.png]

[20210302204212557.png]: /images/20221023/e90be86e685b4d1cb43caf8843f08793.png
[20210302204223909.png]: /images/20221023/f52939220eb34a058ab52cad078353c6.png
[20210302204254901.png]: /images/20221023/1d18514cc47f467e9e57831157ebf333.png
[20210302204802783.png]: /images/20221023/b927bccc396c4d9b8974349b80214d23.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221023/34d0b75b1ea84f5f8af43b3e5176c3ea.png
[20210302205141986.png]: /images/20221023/68eabbbd945d414ab790928ca21f2a85.png
[20210302205134187.png]: /images/20221023/f8f2be3eb75b43a49c9a03f4670b5f44.png
[20210302205242790.png]: /images/20221023/63fd602175b240e5a8534ee61cb0a397.png
[20210302205413828.png]: /images/20221023/18fb7ec6d7304f4ab0393901efe7058d.png
[20210302205418865.png]: /images/20221023/2427863182af4ed99ed92d0185f00f45.png
[20210302205513292.png]: /images/20221023/d9799b6ee71a456ea5bf92527afa6a52.png
[20210302205604153.png]: /images/20221023/f2382bd03f1847fcad9650c497d4455e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221023/0fbdbb19a76549dca47e3e6dd93a3cf4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]: /images/20221023/091427ad32644d81b9d2525813f6f802.png
[20210304140239351.png]: /images/20221023/5644a1edd37046fbae567c8c7816d7cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]: /images/20221023/02a39610e09342a0b17c4518ac8989e1.png
[20210304140309735.png]: /images/20221023/3e6e4fce1de3474aae206fb28d2ea0af.png
[20210304140542253.png]: /images/20221023/ad420b1d081f4624b717b5d3b0c7cd8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]: /images/20221023/d047628d5c9f4718b16508db5ae698f3.png
[20210304140650226.png]: /images/20221023/c1dfda6bb7074b849cc6bd67d8eadb49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 5]: /images/20221023/b136f38d2da84beba69a15f9d59b4532.png
[20210304141413678.png]: /images/20221023/dfaf8c4c4fd7406c9212b7882033278d.png
[20210304141426841.png]: /images/20221023/bb588c7d482f42a593ee966e648a8d61.png
[20210304141442463.png]: /images/20221023/6c91ac3afbab40ba967b68a866efbf26.png