Flink的sink实战之一:初探

摘要:
addSink是否也被调用?

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

关于sink

下图来自Flink官方,红框中就是sink,可见实时数据从Source处开始,在Transformation阶段完成业务逻辑后在sink结束,因此sink可以用来处理计算结果,例如控制台输出或者保存数据库:
在这里插入图片描述

关于《Flink的sink实战》系列文章

本文是《Flink的sink实战》的第一篇,旨在初步了解sink,通过对基本API和addSink方法的分析研究,为后续的编码实战打好基础;

全系列链接

  1. 《Flink的sink实战之一:初探》
  2. 《Flink的sink实战之二:kafka》
  3. 《Flink的sink实战之三:cassandra3》
  4. 《Flink的sink实战之四:自定义》

从一段实例代码开始

  1. 下面是个简单的flink应用代码,红框中的print方法就是sink操作:
    在这里插入图片描述
  2. 下图是官方给出的sink方式,都是DataStream类的API,直接调用即可实现sink,刚才代码中的print就是其中一个:
    在这里插入图片描述
  3. 接下来看看上图中API的源码,先看print方法,在DataStream.java中,如下,实际上是调用了addSink方法,入参是PrintSinkFunction:
    在这里插入图片描述
  4. 另一个常用API是writeAsText,源码如下,调用了writeUsingOutputFormat方法:
    在这里插入图片描述
  5. 追踪writeUsingOutputFormat发现也是调用了addSink,入参是OutputFormatSinkFunction
    在这里插入图片描述
  6. print和writeAsText背后都在调用addSink,那么另一个常用的writeAsCsv方法呢?莫非也是调用addSink?打开一看果然,和writeAsText一样调用了writeUsingOutputFormat,而该方法里面就是在调用addSink:
    在这里插入图片描述
  7. 综上所述,data sink的关键就是addSink的入参,即SinkFunction接口的实现,通过类图直观看到常见的sink能力是如何实现的:

在这里插入图片描述
8. 从上图可见抽象类RichSinkFunction与各种sink能力的关系十分紧密,我们应该重点关注它,在类图上展示方法签名,如下图:
在这里插入图片描述
9. 如上图所示,RichSinkFunction本身没有内容,但是它实现SinkFunction,继承AbstractRichFunction,是RichFunctionSinkFunction这两种特性的结合;
10. RichFunction的特性在前面的《Flink的DataSource三部曲》中已经了解,就是资源的open和close;
11. SinkFunction的特性呢?显然是用来处理计算结果的,类图上显示的是两个invoke方法,来看看官方的PrintSinkFunction.java
在这里插入图片描述
12. writer.write(record)的源码在PrintSinkOutputWriter.java,如下所示:
在这里插入图片描述

小结

至此,我们已经对Flink的sink有了基本了解:

  1. 负责实时计算结果的处理(如输出或持久化);
  2. 主要实现方式是调用DataStream.addSink方法;
  3. 各种sink能力的实现,主要途径是实现addSink方法的入参定义的接口;

后面的章节,一起进行sink方面的编码实战吧,实战的方向:体验官方提供的sink能力,自定义sink能力实现;

欢迎关注公众号:程序员欣宸

微信搜索「程序员欣宸」,我是欣宸,期待与您一同畅游Java世界...
https://github.com/zq2599/blog_demos

免责声明:文章转载自《Flink的sink实战之一:初探》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇cookie在同一域名下,不同路径下的共享,需要设置PathKibana(一张图片胜过千万行日志)下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

有赞 Flink 实时任务资源优化探索与实践

简介: 目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。 随着 Flink K8s 化以及实时集群迁移完成,有赞越来越多的 Flink 实时任务运行在 K8s 集群上,Flink K8s 化提升了实时集群在大促时弹性扩缩容能力,更好的降低大促期间机器扩缩容的成本。同时,由于 K8s 在公司内部有专门的团队进行维护, Flink K8s...

Flink1.13.2版本 Standalone 模式部署

Flink的部署有3种模式,分别是local模式、Standalone模式、yarn模式。其中local就是单机模式,一般来说用于本地开发测试;Standalone跟yarn模式都可以支撑集群部署、实现HA,但是两者在任务分配机制、内存管理等内容上有比较大的差异。一般在处理计算数据量级非常大的生产环境,使用flink on yarn的模式更多一些。 我们的...

Flink on Yarn模式启动流程分析

此文已由作者岳猛授权网易云社区发布。 欢迎访问网易云社区,了解更多网易技术产品运营经验。 Flink On Yarn 架构 Paste_Image.png 前提条件首先需要配置YARN_CONF_DIR, HADOOP_CONF_DIR ,HADOOP_CONF_PATH其中一个用来确保Flink能够访问HDFS和Yarn的RM。 主要启动流程 1....

Flink 概述 基本架构

1.Flink 概述 1.1 流处理技术的演变   在开源世界里,Apache Storm 项目是流处理的先锋。Storm 最早由 Nathan Marz 和创业公司 BackType 的一个团队开发,后来才被 Apache 基金会接纳。Storm 提供 了低延迟的流处理,但是它为实时性付出了一些代价:很难实现高吞吐,并且其正 确性没能达到通常所需的水平,...

大数据实时计算引擎 Flink

数据来源:系统中可以采集到的数据,如用户数据、业务数据等,也包含系统运行时产生的日志数据等。 数据采集:不同数据源生成数据类型格式存在差异,在数据采集前可能增加数据总线(如京东JBus)对业务进行解耦,Sqoop和Flume是常用的数据采集工具。 Sqoop:用于和关系型数据库进行交互,使用SQL语句在Hadoop和关系型数据库间传送数据,Sqoop...

idea搭建flink环境

1.打开idea,选择new project,跳转至如下界面: 2.单击next 下一步,输入项目名称,单击finish完成 3.在项目main项目下新建scala目录,并新建文件,文件名称为 hello.scala   4.新建文件之后,进入helllo.scala文件,会弹出添加sdk,添加自己版本的scala就可以。完成之后可以新建添加如下代码,...