022__尚硅谷_Flink理论_Flink DataStream API（八）Kafka Sink

好，接下来就开启我们这个API这一部分的最后一部分内容，那就是SK这部分API，因为我们说了整个在flink的流失处理程序这个流程过程当中分成三大块，对吧，Source transform，最后是SK，呃，就是对于这个flink而言呢，呃，它不像这个Spark，我们Spark里边是怎么样，就拿到每一个RDD，对吧，For each啊，就可以做这个迭代的操作，然后之后我们是不是想要输出到哪里，你自己去定义啊，对吧？啊，你你自己去做操作，去去连接，去处理，去输出就可以了。在flink里边呢，它对这个流程的要求非常的严格，就是直观的这种输出，正常的对外输出操作都要用最后的think来完成，呃，所以说呃，那这里边我们就要讲这个thinkk的API了，SAPI其实很简单啊，最常见，或者说我们一般情况用的是什么呢？

就是这一句，一个stream要去输出的时候点ADD think，对啊，当然了，大家平常最目前最熟悉的其实是print，那那么大家可能知道print其实就是一种，对，就是一种特殊的think方法啊，因为大家看它是调用了我们这里边的这个data data stream下面的这个print方法对不对，它其实返回的是一个什么，是不是一个S啊，Data stream SK，对，所以这个我们看起来是像一个打印的操作，但实际上它是一个S操作啊，所以呃，这里边我们就直接给大家讲这个I think了，那在这里当然我们知道它是不是可以在里边直接传入一个。啊，这不是think function式，呃，不是s function，不是map function了，这个叫做一个s function，对我们可以自己去自定义一个think function传进来啊，啊所以直观来讲的话，做thinkk其实就是这么一步，但是具体到我们要往什么地方去输出的时候，那就有不同的实现了啊，因为对于一些常用的输出的一些这个工具啊，我们外部的这个存储设备，官方link，官方其实就提供了一部分框架的支持，呃，提提供了一些think的支持，比如说。

大家看这里边啊，这是官网里边截出来两张图，比方说已经官方提供的一些这个连接器，Connector提供哪些呢？啊，卡夫卡对吧？呃，卡夫卡是S和thinkk都提供啊，然后像这个casandra是只提供thinkink的，对吧？像这个ES也是只提供think的啊，那么呃，这里边还有这个system h FS啊，只提供think啊，另外还有其他的一些，大家可以看这个，还有Twitter streaming API，这个只提供south啊，也有不同的这个官方的支持，如果有这个连接器的话，那我们如果想要用它做S，做thinkk，是不是直接引入就可以了，对吧？有对应的包引入，然后那这个就相当于我们不用去自己手写了，对吧？New的时候你直接引他的那个对应的那个对象就可以了。呃，那除了这个之外，大家看到这里边我们常用的好像就只有卡夫卡ES，可能HTFS也算是我们一个常用的，对吧，呃，可能就就这些还常用一点，那比方说我还想。

写到里呢，呃，大家就会想到red其实也是大数据领域常见的一个，呃，数据输出的一个地方，对吧？那如果要往red里面写怎么办呢？诶，这个没关系，大家看，除了这个官方提供的连接器之外。大家可能知道阿帕奇下面还有一个项目叫bahi bahi这个项目它就是专门为啊Spark或者flink这些大数据框架提供外部支持的一些工具的，那大家看有一些连接器是BA here给我们提供了官方支持的，比如说flu、阿卡对吧，Na这些和的这个think他们都是。

巴尔已经支持了的啊，那大家自然就会又又想到那假，假如说我就那么别扭，我最后的输出我是想要写到MYSQL里面，对吧，我我写到关系数据库里边可不可以呢。啊，当然了，也是可以的，只不过这里没有官方提供的连接器，那我们就只好对只好自定义自己去手写了，对吧？啊，所以接下来我们就分几类给大家把这些典型的think操作，呃跟外部的这个连接给大家做一个讲解，呃，首先是卡夫卡，呃夫卡的话，这个就对于大家而言就非常简单了，对吧？呃，这个卡夫卡前面这个连接器我们在做S的时候也已经引入了，所以现在我们就不用再次引入了，那接下来我们就直接在这个啊这样啊，我们还是先去新建一个，因为这个S操作的这个类可能会比较多一点，我们先去在下面新建一个package吧，这个叫做sink test。

然后接下来我们去new一个object，先去实现卡夫卡的啊，这个叫卡夫卡s test。好，呃，那么这个主函数里边大家会想到我们首先还是标准流程，是不是先要创建执行环境啊，Stream execution environment get，这个environment，我们直接把这个引入进来，对吧？然后为了方便我们的这个就是就是这个实现，我们还是把这个全局的。并行路直接设成一，然后接下来是数据源，对吧？啊，现在有数据源，那我们定义一个input。

Stream啊，这个我就是不是直接抄就可以了呀，是不是我这里边的这个stream还可以直接从我们之前的那个文件里面直接读啊，对吧，得把它抄过来，甚至大家会想到这个基本的map操作是不是也得做啊，这个其实都都没什么区别啊，我就直接把这个map操作做完吧。啊，所以大家看这个就是。说这个天下文章一大抄，天下代码也是一大抄，对吧，但是我们得清楚啊，你这个抄过来之后，得得去明确它到底是什么情况，呃，这里大家要注意一下，我这个做了map转换的时候呢，我不要把它直接就是输出这个s reading，因为我们最后要往卡夫卡里写，对吧？输出成什么样子比较简单一些呢？因为后面涉及到序列化的一个过程，所以这里边我是不是直接把它to string变成string，方便序列化输出啊，所以这里转成string方便序列化。

输出。呃，然后呃，大家不要忘记啊，在下边一定要有一个，诶，这是在main里边啊，最后一定不要忘了要有是不是要execute把它执行起来啊，对吧，卡夫卡think。Test，那现在我们已经有这个S了，有了接下来是不是应该要有think啊，对吧？我们这里的think直接就应该是。应该是怎么样，是不是data stream可以直接。ADD的一个think，对，然后这里边是不是就直接可以去你有一个think方式了，因为已经有官方的连接器支持，我们是不是直接从里边引就可以了，大家想这里边我们要去new的一个是什么东西呢？

肯定还是啊，就像我们之前那个S一样，当时我们不是那个，呃，叫什么flink卡夫卡，对不是consumer吗？大家想现在我还是consumer吗？对，当时我们是要从那个卡夫卡里边消费数据，所以是一个消费者，现在我们是不是要往卡夫卡里边写数据，其实对，应该是有一个生产者才对，对吧？哎，那大家就会想到这里我应该有一个卡夫卡，对，是不是应该有一个producer啊啊，同样还是011引用这个版本对吧？那同样这里边我们还是string类型，因为大家看我这里边是不是已经把它涂string了呀啊，所以这里边就直接就是string string类型啊，然后里边啊，那同样大家看一下这个他要传什么东西啊。

好，我们看一眼下面这个实现啊，那这里面可能得有一个broke list对吧，然后得有一个topic ID，最后还得有一个序列化器对吧，序列化的STEM啊，所以大家看这个就比较简单，我们也不用那个五花八门那些配置了，因为我把那个序列化和这个就是这个broke list都已经直接传进去了，就不用再配置properties了。那这里边我们直接给。Local host9092对吧，Topic，我这里边呃给一个think test吧，然后这里边还是同样你一个simple string STEM码，这样是不是就可以了，所大家看这个就这么简单，这个跟fli flink跟卡夫卡确实是连接非常的方便，已经实现的非常的，呃，就是这个实现看起来很优雅了，对吧，因为非常简单一起啊，直接就连接起来了。

好呃，那我们给大家简单的测一下吧，这个大家会想到这里边我我那边卡夫卡是直接已经起来的，对吧？啊，这里边这个有一个processor，这个就没什么关系了，我们把这个NC先停了吧。呃，那那大家会想到我怎么样能看到这里边的卡卡已经有东西了呢？对，那这里边我是不是得去创建一个消费者啊，对卡夫卡consumer，诶这个拼错了啊，杠杠，这个是BOO。Server local host 9092，呃，是这叫什么来着。

哦，Sink test，好，我们把这个消费者创建出来啊，然后这里边大家可能会想到，这里边如果I think的话，那我这边控制台是不是就相当于没有输出了，对吧？如果大家还想在这里边看到的话，那我是不是也可以就相当于再多一个think对不对？那我再把它print出来，直接打印输出，这个也是可以的，对吧？好，那么我们现在就可以直接把这个代码运行一下，他把s test。大家看这边我们已经读完了，把这边数据都已经输出了，我们看一下这边，诶大家看这边是不是也已经把我们的数据都已经读起来输出了呀，对吧，因为我提前没有单独的去创建那个topic，所以这里面有warning对吧？啊，但大家看到这个是正常按照这个顺序一条一条输出出来的啊，所以这个是看起来是没有什么问题的啊。

再跟大家多说一句，就是在实际工作当中，实际应用当中，我们往往是怎么样呢？其实很多情况不是从文件里边去读取数据流的，对吧？哎，往往我们的数据源也应该是一个，呃，也应该是一个流逝的一个数据源，对吧？最常见的是什么？其实就是卡夫卡进，进来之后做一堆操作，然后最后在卡夫卡出，对，所以这就是一个完整的所谓的数据管道，我们之前给大家讲对吧，Data pipe plan，所以接下来我们给大家稍微做一些调整，可以把它转换成这样的一个卡夫卡进，卡夫卡出的一条管道，那比方说这个卡普卡镜，我们是不是在这个S，这个test里边已经给大家讲过了，我就直接把这个copy过来就完事，对不对？好。

这个。Connector，对，这个consumer也给放进来啊，然后我们检查一下，看一看，呃，这里边我们就直接可以把这个input stream把它注掉，这里我们的input stream就定义成a source，用这个卡夫卡去输入，对吧？呃，然后这里边我们定义的这个输入的这个topic叫sensor，然后后边我们输出的这个topic叫s test是吧？啊就是这样的一个状态，好，我们把这个代码机器来看一下，而且下面我们中间是会把它打印输出的，对吧？在处理的过程当中会把它打印输出。把这个代码提起来，然后接下来我们的要求是不是应该得有一个producer，哎，之前我们这个还在对吧，另外是不是还得有一个对，还得有一个消费者啊，这就是前面我们给的这个think text这个消费者对吧？然后这个生产者呢，他是应该是三四这个topic下面的一个生产者，好所以接下来这样啊。

因为我们可能还需要看到这里边的一个输出，呃，那接下来我们怎么输出呢。就把这个数据一条一条。诶，大家看这里边，诶，我把这一部分已经从这里这个topic的输入，然后这里边已经打印出来了是吧？呃，这里可能是因为诶大家看我这里边这个状态没有没有没有解开啊，所以说没显示出来，大家看到这里边输出的我们本来输入的是一条这个string，就是啊逗号分割的这个数据对吧？这里边输出的是不是就变成包装好的样例类的这个string了，对吧？啊sensor reading啊大家看就是这样的，所以大家可以感受一下这个数据管道的这个状态啊，这里面是不是又输出了一条啊？呃，大家看到就是我们这里面从一个topic进来，消费数据经过处理转换之后从另外一个topic输出，这就是我们说的卡不卡进，卡夫卡出啊，有这样的一个数据管道，而且在中间，如果我们想要看他的状态的话，其实在这里面有它的那个中间状态，对吧。

啊，可以做各种各样的处理，这就是这个卡夫卡这一部分，再给大家补充一点内容，就是怎么样去构建一个数据管道啊，当然既然提到这里，就是大家会想到前面我们讲卡夫卡跟弗link做连接，源头那边如果要保证它那个数据不丢的话，得有一些操作，对吧？啊，那大家可能会想到，那还有一个问题，就是说你如果往卡卡里边写数据的话。那是不是也有可能，诶前面写对吧，往里面写写着写着中间某一步给挂了，挂了之后是不是还就是你接下来这个数据是不是要重放，有一个这个问题啊，你如果这个数据不重放的话，那相当于这个数据要就是之前的那个，呃，我们的那个操作有可能就要丢，对吧，如果要重放的话，是不是有可能输出到卡不卡就有可能要重复再输出一次啊，诶所以这又有一个问题，就是所谓的就是从头到尾，如果我们要保证它状态一致。

在出现故障的时候回滚保证它状态一致，这有一个问题叫端到端的状态一致性，呃，先给大家就按to end对吧，先给大家提出这样一个概念啊，大家可以思考一下，就是怎么样可以保证它这个做到端到端状态一致性啊，这个后续我们讲过checkpoint这些机制之后，给大家详细讲解跟卡夫卡连接的端到端状态一致性保持啊好，那这一部分我们就先讲到这儿。