Flink（三）Flink的主力APIDataStreamAPI

　　大家好，我是 杰哥
　　今天，继续来探索  Flink  ，今天我们主要来看看  Flink  为我们所提供的 分层 API 一 Flink 的分层 API
　　Flink 不仅具有高吞吐、低延迟、高可用等优秀特性，而且还提供了 易于使用的分层 API ，所以它也是一个 易于开发 的框架，它的  API 分层 如图所示
　　Flink  包含三层  API : 低级 APIs  - 有状态处理、 核心 APIs  -  DataStream API  以及  高级 APIs  -  Table API & SQL 1、有状态处理
　　最底层级的抽象仅仅提供了 有状态流 ，它将处理函数（Process Function）嵌入到了 DataStream API  中。通过实现  ProcessFunction  接口来进行操作。 ProcessFunction  可以处理一或两条输入数据流中的单个事件或者归入一个特定窗口内的多个事件。它提供了对于时间和状态的细粒度控制。开发者可以在其中任意地修改状态，也能够注册定时器用以在未来的某一时刻触发回调函数 2、核心 APIs-DataStream API
　　DataStream ？是不是感觉还有点儿熟悉呢？
　　是的，上一节的例子里面，它就出现过，我们使用  DataStream  类来存放  Flink  程序中的数据集合，供 Flink 进行处理 // 2-  读取文本流 （nc -lk 7777） DataStreamSource lineDataStream = env.socketTextStream(＂localhost＂, 7777);
　　其中  DataStreamSource  其实就是集成自  DataStream 而这里所提到的  DataStream API  的名字就来自于  DataStream  类。这个类中，可以存放包含重复项的不可变数据集合。集合中的数据可以是有限的，也可以是无界的
　　DataStream  在使用方面类似于常规的  Java  中的集合，但在一些关键方面有很大的不同，比如它们是不可变的，这意味着一旦创建了它们，就不能添加或删除元素。此外，我们也不能查看里面的元素，只能使用  DataStream API   操作 (也称为转换)对它们进行一些变换操作
　　可以通过在  Flink  程序中添加源来创建初始  DataStream 。然后，再可以从中派生新的流，并使用映射、过滤器等  API  方法将它们组合在一起 3、Table API & SQL API
　　Table API  是一个针对  Java、Scala  和  Python  的语言集成查询  API ，它允许以非常直观的方式组合来自关系操作符(如选择、筛选和连接)的查询。 Flink  的  SQL API  基于  Apache Calcite ，它实现了  SQL  标准的语法
　　Flink  提供的最高层级的抽象是  SQL 。这一层抽象在语法与表达能力上与  Table API  类似， 但是是以  SQL  查询表达式的形式表现程序。 SQL  抽象与  Table API  交互密切，同时  SQL  查询 可以直接在  Table API  定义的表上执行
　　Flink  的 分层  API  中，最主要、最常用的是  DataStream API  和  Table API & SQL  。我们可以将两者单独使用，也可以混合使用，取决于我们具体的应用场景 二 DataStream API 认识
　　我想，从它的 操作步骤 来认识它，应该是比较直观的
　　类似于我们曾经了解到的  JDBC  会有固定的 7 个步骤，使用  Flink  的  DataStream API  进行数据的处理，实际上也包含以下  5  个固定步骤，而  DataStream API  也可以随之被分为  5  类： 1、设置执行环境    2、读取输入流    3、转换操作    4、输出到一个或多个数据汇中    5、执行程序
　　我们来分别看一下 1、设置执行环境
　　在  Flink  中，可以使用  StreamExecutionEnvironment  的下列三种方式进行执行环境的创建 getExecutionEnvironment(); createLocalEnvironment(); createRemoteEnvironment(String host, int port, String... jarFiles);
　　其中前两个表示创建的是 本地环境 ，即表示  Flink  程序运行在 本地机器 。还可以通过指定远程机器的 主机名 、 端口号 以及 程序本身所生成的 jar 包最终拷贝到的路径 ，创建一个 远程执行环境 ，使得  Flink  程序运行在所指定的主机上
　　2、读取输入流
　　StreamExecutionEnvironment  为我们提供了一系列创建流式数据源的方法，使得我们可以将数据流读取到应用中。这些数据流可以来自于 文件、数据库、消息队列 等 读取的数据，就可以统一放入  DataStream  对象中
　　1）从集合中读取数据 ArrayList user = new ArrayList<>(); user.add(new User(＂Mary＂,＂./home＂,1000L)); user.add(new User(＂Bob＂,＂./cart＂,2000L)); DataStream stream = env.fromCollection(user);
　　2）从文件中读取数据 DataStream stream = env.readTextFile(＂words.txt＂);
　　3）从 Socket 读取数据 DataStream stream = env.socketTextStream(＂localhost＂, 7777);
　　4）从消息中心读取数据 DataStreamSource stream = env.addSource(new  FlinkKafkaConsumer( ＂clicks＂, new SimpleStringSchema(), properties ));
　　5）自定义 Source 如果我们需要读取数据的数据源， Flink  没有为其提供读取数据源的方法，我们就可以通过  Flink  提供的自定义  Source  的方式进行了 那就只好自定义实现  SourceFunction  了。接下来我们创建一个自定义的数据源，实现  SourceFunction  接口。主要重写两个关键方法： run ()和  cancel () run ()方法：使用运行时上下文对象（SourceContext）向下游发送数据；cancel ()方法：通过标识位控制退出循环，来达到中断数据源的效果。3、转换操作
　　当数据被存入  DataStream  对象中后，我们就可以对它进行转换操作了 转换的类型有多种多样。有些会重新组织一下  DataStream  中的数据流，比如对这些数据进行一次分组或者分区；有些呢，就会生成一个新的  DataStream  （类型可能会发生变化）
　　比如，使用  map（） 可以将一个输入流中的所有正方形全部转换为圆形
　　今天在这里就不细说了，下次文章会专门为大家介绍  DataStream API  的转换操作 4、输出结果
　　一般场景下，我们将数据处理的结果总是会发送到某些外部系统，比如 文件、数据库，以及消息中心 中等，当然测试时，可以直接输出至 标准输出 中
　　比如，输出到文件的方式如下： StreamingFileSink fileSink = StreamingFileSink .forRowFormat(new Path(＂./output＂), new SimpleStringEncoder<>(＂UTF-8＂)) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(TimeUnit.MINUTES.toMillis(15) )  .withInactivityInterval(TimeUnit.MINUTES.toMillis(5 )) .withMaxPartSize(1024 * 1024 * 1024) .build()) .build(); // 将 Event 转换成 String 写入文件 stream.map(Event::toString).addSink(fileSink);
　　此外，还可以输出到  Kafka ， mysql , pulsar  等
　　最新版本的  Flink  所支持的所有输入输出如下：
　　5、执行程序
　　在应用定义完成之后，我们就可以通过  StreamExecutionEnvironment  类中的  execute () 方法进行执行
　　Flink  被设计为了 延迟计算 的方式执行，在 执行  execute（）  方法之前，前面的所有程序，只是在执行环境中构建了一个执行计划。也就是说，只有执行了  execute（）  方法之后，前面定义的  Flink  应用程序才会被真正执行
　　这 5 个步骤看完了，我们再将那个简单的实例拿过来看看，确认一下是否是这样 public class StreamWordCount {     public static void main(String[] args) throws Exception{ //     1-创建执行环境         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //     2-读取文本流 （nc -lk 7777）         DataStreamSource lineDataStream = env.socketTextStream(＂localhost＂, 7777); //       3-转换操作 //       3.1 收集各个单词，定义为二元组         SingleOutputStreamOperator> streamOperator = lineDataStream.flatMap((String line, Collector> out) -> {             String words[] = line.split(＂ ＂);             for (String word :words) {                 out.collect(Tuple2.of(word, 1L));             }         }).returns(Types.TUPLE(Types.STRING, Types.LONG)); //       3.2 分组         KeyedStream, String> keyedStream = streamOperator.keyBy(data -> data.f0); //       3.3-统计         SingleOutputStreamOperator> sum = keyedStream.sum(1); //      4-打印         sum.print(); //      5-启动执行         env.execute();     } }
　　这个其实就是严格按照我们上述所提到的步骤来走的
　　其中第 2 步骤，是通过一行一行地读取  socket  文本流内容，进行数据源的输入的，将所输入的数据流存入  lineDataStream  中
　　而第 3 步骤，则是对  lineDataStream  中的数据进行转换处理操作。首先将各个数据通过  flatMap（） 方法转换为二元组，然后使用  keyBy()  方法，对其元素进行分组，接着再使用  sum（） 方法，对分组之后的元素进行求和操作
　　随着数据流的不断输入，所输入的每个单词出现的次数便会通过程序被源源不断地输出 总结
　　Flink  提供了 分层 API ，便于我们进行开发。在实际开发中，我们使用最多的就是  DataStream API  ，而怎样认识  DataStream API  呢？从它的操作步骤来看，可以分为  5  个步骤，也就是说  DataStream   API  可以分为  5  大类：
　　1、设置执行环境
　　2、读取输入流
　　3、转换操作
　　4、输出到一个或多个数据汇中
　　5、执行程序
　　嗯，就这样。每天学习一点，时间会见证你的强大
　　欢迎大家关注我们的公众号【青梅主码】，一起持续性学习吧~
　　往期精彩回顾
　　总结复盘
　　架构设计读书笔记与感悟总结
　　带领新人团队的沉淀总结
　　复盘篇：问题解决经验总结复盘
　　网络篇
　　网络篇（四）：《图解 TCP/IP》读书笔记
　　网络篇（一）：《趣谈网络协议》读书笔记（一）
　　事务篇章
　　事务篇（四）：Spring事务并发问题解决
　　事务篇（三）：分享一个隐性事务失效场景
　　事务篇（一）：毕业三年，你真的学会事务了吗？
　　Docker篇章
　　Docker篇（六）：Docker Compose如何管理多个容器？
　　Docker篇（二）：Docker实战，命令解析
　　Docker篇（一）：为什么要用Docker？
　　..........
　　Spring  Cloud篇章
　　Spring Cloud（十三）：Feign居然这么强大？
　　Spring Cloud（十）：消息中心篇-Kafka经典面试题，你都会吗？
　　Spring Cloud（九）：注册中心选型篇-四种注册中心特点超全总结
　　Spring Cloud（四）：公司内部，关于Eureka和zookeeper的一场辩论赛
　　..........
　　Spring Boot篇章
　　Spring Boot（十二）：陌生又熟悉的 OAuth2.0 协议，实际上每个人都在用
　　Spring Boot（七）：你不能不知道的Mybatis缓存机制！
　　Spring Boot（六）：那些好用的数据库连接池们
　　Spring Boot（四）：让人又爱又恨的JPA
　　SpringBoot（一）：特性概览
　　..........
　　翻译
　　[译]用 Mint 这门强大的语言来创建一个 Web 应用
　　【译】基于 50 万个浏览器指纹的新发现
　　使用 CSS 提升页面渲染速度
　　WebTransport 会在不久的将来取代 WebRTC 吗？
　　.........
　　职业、生活感悟
　　你有没有想过，旅行的意义是什么？
　　程序员的职业规划
　　灵魂拷问：人生最重要的是什么？
　　如何高效学习一个新技术？
　　如何让自己更坦然地度过一天？
　　..........