【Flink-1.17-教程】-【四】Flink DataStream API（1）源算子（Source）

【Flink-1.17-教程】-【四】Flink DataStream API（1）源算子（Source）

• 1）执行环境（Execution Environment）
• • 1.1.创建执行环境
  • 1.2.执行模式（Execution Mode）
  • 1.3.触发程序执行
  • 2）源算子（Source）
  • • 2.1.准备工作
    • 2.2.从集合中读取数据
    • 2.3.从文件读取数据
    • 2.4.从 Socket 读取数据
    • 2.5.从 Kafka 读取数据
    • 2.6.从数据生成器读取数据
    • 2.7.Flink 支持的数据类型

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

      1）执行环境（Execution Environment）

      Flink 程序可以在各种上下文环境中运行：我们可以在本地 JVM 中执行程序，也可以提交到远程集群上运行。

      不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前 Flink 的运行环境，从而建立起与 Flink 框架之间的联系。

      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.执行模式（Execution Mode）

      从 Flink 1.12 开始，官方推荐的做法是直接使用 DataStream API，在提交任务时通过将执行模式设为 BATCH 来进行批处理。不建议使用 DataSet API。

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

      DataStream API 执行模式包括：流执行模式、批执行模式和自动模式。

      • 流执行模式（Streaming）

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

      • 批执行模式（Batch）

        专门用于批处理的执行模式。

      • 自动模式（AutoMatic）

        在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。批执行模式的使用。主要有两种方式：

        （1）通过命令行配置

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

        在提交作业时，增加 execution.runtime-mode 参数，指定值为 BATCH。

        （2）通过代码配置

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

        在代码中，直接基于执行环境调用 setRuntimeMode 方法，传入 BATCH 模式。

        实际应用中一般不会在代码中配置，而是使用命令行，这样更加灵活。

        1.3.触发程序执行

        需要注意的是，写完输出（sink）操作并不代表程序已经结束。因为当 main() 方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据——因为数据可能还没来。Flink 是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为“延迟执行”或“懒执行”。

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

        env.execute();
        

        2）源算子（Source）

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

        在 Flink1.12 以前，旧的添加 source 的方式，是调用执行环境的 addSource()方法：

        DataStream stream = env.addSource(...);
        

        方法传入的参数是一个“源函数”（source function），需要实现 SourceFunction 接口。

        从 Flink1.12 开始，主要使用流批统一的新 Source 架构：

        DataStreamSource stream = env.fromSource(…)
        

        Flink 直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的 Source，通常情况下足以应对我们的实际需求。

        2.1.准备工作

        具体代码如下：

        public class WaterSensor { public String id;
            public Long ts;
            public Integer vc;
            // 一定要提供一个 空参 的构造器
            public WaterSensor() { }
            public WaterSensor(String id, Long ts, Integer vc) { this.id = id;
                this.ts = ts;
                this.vc = vc;
            }
            public String getId() { return id;
            }
            public void setId(String id) { this.id = id;
            }
            public Long getTs() { return ts;
            }
            public void setTs(Long ts) { this.ts = ts;
            }
            public Integer getVc() { return vc;
            }
            public void setVc(Integer vc) { this.vc = vc;
            }
            @Override
            public String toString() { return "WaterSensor{" +
                        "id='" + id + '\'' +
                        ", ts=" + ts +
                        ", vc=" + vc +
                        '}';
            }
            @Override
            public boolean equals(Object o) { if (this == o) { return true;
                }
                if (o == null || getClass() != o.getClass()) { return false;
                }
                WaterSensor that = (WaterSensor) o;
                return Objects.equals(id, that.id) &&
                        Objects.equals(ts, that.ts) &&
                        Objects.equals(vc, that.vc);
            }
            @Override
            public int hashCode() { return Objects.hash(id, ts, vc);
            }
        }
        

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

        • 类是公有（public）的
        • 有一个无参的构造方法
        • 所有属性都是公有（public）的
        • 所有属性的类型都是可以序列化的

          Flink 会把这样的类作为一种特殊的 POJO（Plain Ordinary Java Object 简单的 Java 对象，实际就是普通 JavaBeans）数据类型来对待，方便数据的解析和序列化。另外我们在类中还重写了 toString 方法，主要是为了测试输出显示更清晰。

          我们这里自定义的 POJO 类会在后面的代码中频繁使用，所以在后面的代码中碰到，把这里的 POJO 类导入就好了。

          2.2.从集合中读取数据

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

          public class CollectionDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                  // TODO 从集合读取数据
                  DataStreamSource source = env
                          .fromElements(1,2,33); // 从元素读
          //                .fromCollection(Arrays.asList(1, 22, 3));  // 从集合读
                  source.print();
                  env.execute();
              }
          }
          

          2.3.从文件读取数据

          真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

          读取文件，需要添加文件连接器依赖：

          org.apache.flinkflink-connector-files${flink.version}

          示例如下：

          public class FileSourceDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                  env.setParallelism(1);
                  // TODO 从文件读： 新Source架构
                  FileSource fileSource = FileSource
                          .forRecordStreamFormat(
                                  new TextLineInputFormat(),
                                  new Path("input/word.txt")
                          )
                          .build();
                  env
                  		.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource")
                     		.print();
                  env.execute();
              }
          }
          /**
           *
           * 新的Source写法：
           *   env.fromSource(Source的实现类，Watermark，名字)
           *
           */
          

          说明：

          • 参数可以是目录，也可以是文件；还可以从 HDFS 目录下读取，使用路径 hdfs://…；
          • 路径可以是相对路径，也可以是绝对路径；
          • 相对路径是从系统属性 user.dir 获取路径：idea 下是 project 的根目录，standalone 模式下是集群节点根目录；

            2.4.从 Socket 读取数据

            不论从集合还是文件，我们读取的其实都是有界数据。在流处理的场景中，数据往往是无界的。

            我们之前用到的读取 socket 文本流，就是流处理场景。但是这种方式由于吞吐量小、稳定性较差，一般也是用于测试。

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

            2.5.从 Kafka 读取数据

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

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

            org.apache.flinkflink-connector-kafka${flink.version}

            代码如下：

            public class KafkaSourceDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                    env.setParallelism(1);
                    // TODO 从Kafka读： 新Source架构
                    KafkaSource kafkaSource = KafkaSource.builder()
                            .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092") // 指定kafka节点的地址和端口
                            .setGroupId("atguigu")  // 指定消费者组的id
                            .setTopics("topic_1")   // 指定消费的 Topic
                            .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value
                            .setStartingOffsets(OffsetsInitializer.latest())  // flink消费kafka的策略
                            .build();
                    env
            //                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkasource")
                            .fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource")
                            .print();
                    env.execute();
                }
            }
            /**
             *   kafka消费者的参数：
             *      auto.reset.offsets
             *          earliest: 如果有offset，从offset继续消费; 如果没有offset，从 最早 消费
             *          latest  : 如果有offset，从offset继续消费; 如果没有offset，从 最新 消费
             *
             *   flink的kafkasource，offset消费策略：OffsetsInitializer，默认是 earliest
             *          earliest: 一定从 最早 消费
             *          latest  : 一定从 最新 消费
             *
             *
             *
             */
            

            2.6.从数据生成器读取数据

            Flink 从 1.11 开始提供了一个内置的 DataGen 连接器，主要是用于生成一些随机数，用于在没有数据源的时候，进行流任务的测试以及性能测试等。1.17 提供了新的 Source 写法，需要导入依赖：

            org.apache.flinkflink-connector-datagen${flink.version}

            代码如下：

            public class DataGeneratorDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                    // 如果有n个并行度， 最大值设为a
                    // 将数值 均分成 n份，  a/n ,比如，最大100，并行度2，每个并行度生成50个
                    // 其中一个是 0-49，另一个50-99
                    env.setParallelism(2);
                    /**
                     * 数据生成器Source，四个参数：
                     *     第一个： GeneratorFunction接口，需要实现， 重写map方法， 输入类型固定是Long
                     *     第二个： long类型， 自动生成的数字序列（从0自增）的最大值(小于)，达到这个值就停止了
                     *     第三个： 限速策略， 比如 每秒生成几条数据
                     *     第四个： 返回的类型
                     */
                    DataGeneratorSource dataGeneratorSource = new DataGeneratorSource<>(
                            new GeneratorFunction() { @Override
                                public String map(Long value) throws Exception { return "Number:" + value;
                                }
                            },
                            100,
                            RateLimiterStrategy.perSecond(1),
                            Types.STRING
                    );
                    env
                            .fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "data-generator")
                            .print();
                    env.execute();
                }
            }
            

            2.7.Flink 支持的数据类型

            1、Flink 的类型系统

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

            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 模式的类。

              （4）辅助类型

              Option、Either、List、Map 等。

              （5）泛型类型（GENERIC）

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

              在这些类型中，元组类型和 POJO 类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO 还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为 Flink 的 POJO 类型。

              Flink 对 POJO 类型的要求如下：

              • 类是公有（public）的
              • 有一个无参的构造方法
              • 所有属性都是公有（public）的
              • 所有属性的类型都是可以序列化的

                3、类型提示（Type Hints）

                Flink 还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。但是，由于 Java 中泛型擦除的存在，在某些特殊情况下（比如 Lambda 表达式中），自动提取的信息是不够精细的——只告诉 Flink 当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。

                为了解决这类问题，Java API 提供了专门的“类型提示”（type hints）。

                回忆一下之前的 word count 流处理程序，我们在将 String 类型的每个词转换成（word，count）二元组后，就明确地用 returns 指定了返回的类型。因为对于 map 里传入的 Lambda 表达式，系统只能推断出返回的是 Tuple2 类型，而无法得到 Tuple2。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。

                .map(word -> Tuple2.of(word, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));
                

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

                returns(new TypeHint>(){})