075.尚硅谷_Flink-Table API和Flink SQL_更新模式

好，大家在这个写入到文件系统的过程当中，可能就会发现一个问题，那难道说我做了这样的一个聚合操作，做了里边更新操作之后，接下来我真的就没有办法写入到这个外部系统里面去了吗？诶，当前至少这个文件系统是没有办法写入进去了，因为我们看到csv table think这里边它只实现了一个aend stream table think，对吧？那你如果想要实现做了更新操作的这些，呃，当前的这这些表还能够写入到外部系统，那那得写入到什么样的地方去呢？啊，那就是大家能想到写入到一些数据库里面去，对吧？那这些数据库必须得支持什么呢？那就是它必须得实现我们这里边其他的一些table s，这里边有哪些呢？Retract stream table s。比方说像我们那个前面给大家讲的啊，就把这个聚合之后的这张表a j table，我们转换成流的时候，可以突tract stream，对吧，这就相当于是一个retract strip table think，那另外还可以有什么呢？有absur strip table think，那这里边的这些就是aend retract upset，这些到底又是什么概念呢？在这里就要给大家隆重的推出介绍一个重要的概念，叫做更新模式，哎，那什么叫做更新模式呢？啊，大家就会发现了，对于这个流式的查询，我们流式的处理而言，现在我们定义了一个CQL去做查询表的查询，对吧？啊，那这里边其实是我们必须就得定义一个，因为你涉及到最后我们拿到这个，呃，这个表的结果，它有可能是直接往后追加，也有可能是直接是改这个表里面的某一行数据，对吧，那对于这样的一张表，你最后要写入到外部系统的时候。

你到底怎么样才能告诉外部系统我当前的这个数据是要改一行数据还是说往后面追加呢？这就涉及到了一个更新模式的问题，更新模式主要是用来干什么呢？它就是让我们定义跟外部系统交换消息的类型是什么，就由更新模式所谓的update mode来制定。那这里面的更新模式呢，主要就是有这么三种，一种叫做追加模式end，另外一种撤回模式retract，那最后还有一种叫更新插入UPS模式，那那这三种刚才我们在这个，呃，最终cable think的实现方式上啊，都看到有有对应的这种模式的实现了，那这三种模式分别表示什么含义呢？首先最简单的就是追加模式PA模式，这个模式表示就是我们当前的table s和外部的连接器。

地之间，它只传递只交换一种信息，这种信息就是插入信息，Insert信息，也就是说表示我当前的这个表只做追加，往后面追加一行一行数据的这个操作，对吧？呃，就是或者说你只有一种消息的话，那就是消息就没有类型了，对吧？所有只要发给你的消息都是往后面追加，那大家想一想，对于流处理而言啊，流处理而言是不是我们外部输出的时候，正常来讲就应该是来一条数据，我们不是说流处理就是，就是什么呢？就是来一条数据，输出一条数据，我们在这里边有有一个中间有一个这个操作对吧？这里边来一条数据，输出一个处理的结果，那正常来讲其实就应该是，哎，就是在我们的这张输出，假如说有一张输出表的话，那是不是就应该是不停的在后面追加啊，诶大家觉得这个正常情况下就应该是这样不停的追加的一个状态，但。

那有时候呢，我们的定义又不一样，那你比方说你像我们的这个count啊，每来一个对应的sensor，我们要把它这个count值做一个更新，那他这个count值你如果存在一张表里面的话，只是不停的把这个更新之后的值往后面追加，这就起不到我们的作用了，对吧？哎，你后续如果说要去直接把你说我最后要的这个count值，你想往这个里面写，或者说往往里面写，我最后想拿的值是什么呢？你难道拿的是一大张表吗？呃，就是所有的这个每来一次数据之后，更新之后的这个所有的count值吗？其实不是的，我最后只想要拿到现在最新更新之后的那个就够了，对吧？啊，当然这个数据是所有的都会传递过来的，但是我外部的这个数据库，我需要知道你最新的这个值是什么，那怎么样能够告诉外部数据库说我现在的这个数据，哎，这这就有一个问题了，那你就是说你看你这里面。

输出的这个数据啊，都是之前来了一个，比方说3411对吧，后面来了一个3412，来了一个3413，那对于外部系统而言，我怎么知道你现在就是要更新341的这个数据，而不是追加了一条3412的这条数据呢？所以这里边就得涉及到，诶，我就得指定另外的模式传递不同的信息，就像我们当时在那个控制台输出的时候，大家看到的那样啊，就是retra stream，它输出的是什么呢？加上一个出false的一个标志位，表示当前这条信息到底是应该要删的，还是说我要新增的，所以大家看到接下来我们就有一个模式叫做撤回模式，Retra模式，它表示的含义是我向外部系统去发送的数据，发的这个消息呢，有两种类型，一种叫做A消息。

就是另外一种呢，叫做retra撤回消息，就是一种是添加，一种是撤回，撤回之前的某一个数据对吧？那所以接下来大家都知道，我们对于表里面的操作，这个增删改查查询就不说了，呃，我们就是没有没有任何的改变嘛，所以我们主要关心的就是增删改对吧？啊，那对于这个之前我们这个呃追加模式而言，大家就会发现了，只能插入，不能做删除和update做更新，对吧？那现在如果有这个retra的模式的话，你可以发送不同类型消息的话，那其实就是做增删改都可以了，如果要是增的话，很简单，你发送一条消息，然后告诉这个外部系统说，诶，我当前类型是A，那就表示什么呢？就是追加嘛，对吧，往后面追加就可以了，这具体到我们的那个，呃里边，是不是就相当于当前的那前前面那个尔类型是处啊，对吧，如果是。

啊，就表示诶，我当前那条消息是追加进来的，然后另外还有一个，如果要是说删除一条消息怎么办呢？我就编码成一个retract消息，就假如说我收到一条消息是false的话，就像我们那个对吧，那个前面一个标志位是false，就表示当前是一个rere消息，就要撤回，就要删除，对吧，之前那条消息没用了，那如果说我要是更新一条数据怎么办呢？大家看到它就是连续发送两条消息，一条消息是把之前的原来的那要更新的那条数据撤回对吧，把它删掉，然后呢，再加来，再加下一条更新之后的那条数据的APP，把它再添加进去，所以就相当于是用两条消息表达以一次数据的更新，这样的话外部系统就非常的清晰了，对吧，你接下来你传过来的消息我就知道，诶，你到底是要做这个表里边的增山。

哎，到底是做哪个操作我就都知道了，这样的话我们就可以直接操作这个关系数据库了，对吧，就非常的容易了。那另外还有一种模式叫做更新插入up sir的模式，它跟撤回模式有些类似，都是有两种不同的消息，但是呢，它的编码方式不一样，诶upsur大家知道本来没有这个词，Updates是什么意思呢？UPS其实就是update加insert对吧，就合在一起，这种所有的update和insert合在一起叫做UPS，这有点像什么？呃，就是前面像我们这个之前做自定义think往这个MYSQL里边啊呃，自定义写入数据的时候，我们在什么呢？直接上来之后就去做一个更新操作，对吧？啊就是如果说它没更新成功的话，我们再执行那个插入啊，这这是这样的一个逻辑，那这里边的这个这个模式呢，也类似，它就是更新和插入。

都叫做UPS，那它相当于什么呢？就是在做这个UPS的时候，我就要先去找这个数据到底有没有对吧？呃，所以在这种模式下呢，我是必须要得知道一个K的，就相当于我得定义一个当前的一个键，对吧？这个K如果有的话，我就直接更新它后边的那个value，就相当于是一个update，如果这个key没有的话，哎，那我就相当于是一个insert，直接把它插入就完了啊。所以这样的话，所有的更新和插入我都编，编辑成编码成一个UPS消息，那如果删除的话，这是单独的一个delete消息，那它的这样做好处是什么呢？大家看，好处就是撤回模式下更新一次我要发两条消息，那现在是不是我更新和插入，不管什么情况，我都只发一条消息就完了呀，对吧？所以整个的这个呃，系统的性能就会大大的提升，整个的这个代码的，呃，就是我们当。

填这个消息传递机制啊，就会更加的简洁，那当然它有缺点，缺点就是你必须外部的这个存储系统啊，数据库必须得支持我们这个K的设定，对吧？而且我必须能够找到这个K，这样的话我就可以直接去做这个UPS统一做操作了啊，所以大家会发现不同的外部连接系统，不同的数据库，它可能会支持不同的模式，对吧？对于我们当前的这个文件系统而言，很遗憾，它只支持当前的这个ipad模式，所以说如果说我们做了一些更新操作，做了一些删除操作这样的一张表里边，呃，得到了一个结果表的话，你想把它直接插入到这个文件系统，那就做不到了，只能是把它插入到支持其他的两种模式里边的这样的外部系统才可以，这就是关于更新模式。