039__尚硅谷_Flink理论_Flink状态编程（下）

其实前面我们在实现的过程当中，大家会看到当时其实还有另外一种实现方式。因为我们在写这里边点flat map的时候，其实已经有同学看到了，大家看到还有一种什么实现啊，我在定义一个process。STREAM3DATA stream k也是要先K对吧，ID。大家会发现我如果做这个flat map的话，它是不是可以有一个with啊，所以这个，呃，那那这个是干什么的用的呢？啊，它其实就是我们前面点flat map，然后再去自己实现一个rich flat map function里边又自定义state这样的一种简单的简写形式，它用这种方式直接可以实现，好那我们大家也学习一下这个怎么用这种方式来实现这个状态编程啊，啊大家看一下，呃，Fla map with state这里边要传什么东西呢？哎哟，这个东西有点奇怪，我们点进去看一下吧。

大家看一下这个flat map with state啊，这个当时我们其实看这个kid stream API的时候，大家就看到了，就有这两个对吧？Map with state和Fla map with state，那别的我们是不是就没看到啊，你看到有那个，呃，什么reduce with state吗？Aggregate with state没看到对不对？呃，那为什么呢？对那些本来就是有状态的，而map和flat map本身是没有状态，对无状态的，所以这里边它就专门提供了这两种简便方式啊，快速去操作状态的一些方式，那家看一下它这里边的这个定义我们得怎么写呢？大家这里看一看啊，这个这个稍微有点麻烦大家看一眼，这里边它得传类型，类型是一个R一个S，哎哟，这个看不懂了，这是什么玩意儿啊，之前我们这个不是流里边它那个数据类型都叫T吗。

那这里边这这好像跟那个输入的类型都不一样啊，那这又是什么东西呢。哎，这里大家注意看看它的输出是什么，它的输出是一个data stream对，然后data stream输出之后的数据类型是具体的类型是R对，所以这里边它首先给的这个R是个输出的类型，好，然后大家又会想到，那这个S又是什么呢？哦，有些同学已经想到了，我们这里边除了输入和输出，应该还涉及到一个什么，还涉及到一个状态，对不对，S指的就是state，所以这里的S指的是定义的状态的数据类型，好，那然后接下来大家再看啊，这这里边是预预先定义好的这个数据类型，我们先写好吧。

这里边得得定一个什么玩意儿啊，输入是什么？哎，这里面不要输入对吧，这里面不要输入，我们要输出输出是什么？对，输出是我们定义好的那个三元组string double double。然后还有一个状态，对我们的状态是什么类型呢？我们现在存的状态是不是就是当前的那个值啊，对吧，上一个的那个温度值，所以它是一个double类型对不对啊，这里面当然就这里定义，当然就不应该定义value state了，对吧？当然就是直接给一个double就可以了，好，然后大家看这里外面就不报错了，里边的实现得怎么去实现呢？还是回过头来看一下，他要传的是一个一个方式。

那么这个function，这应该是一个什么东西呢？这个function是一个，诶大家看这个function，这个function的输入参数又是什么。一个TT是输入的数据类型，哎，这个大家知道了对吧？是我们整条流里边主主流里边的这个输入类型，然后还有一个optionss是状态的数据类型，对，这个大家一定要搞清楚啊，这里边是，呃，这个所谓的我们状态的数据类型，所以这里边它是什么呢？相当于我知道了当前的输入是什么，还得知道当前的状态是什么，对吧？诶，为什么这里边来了一个option呢？是不是有可能一开始没有状态啊，对吧，所以这里边它还提供了这种方便的这种，呃，判断这个状态是不是没有，所以它是一个optional option的一个类型，所以接下来它会把它转换成一个什么输出呢，大家看。

于是。转换成一个trasible ones r，那R本身是一个输出的数据类型，所以这是不是最后就是来了一个输入，是不是把它转换成对应的输出啊，哎，Flat map嘛，你做map操作，那当然是输入要对应的一个输出，对吧？另外是不是还得告诉我状态有什么变化呀？大家想想是不是这样，想想我们当时一开始给大家画的那个有状态处理的那个图，对吧，中间是我们的那个算子，算子任务，这里边是不是来一个数据，然后走一个，这里边是不是去读取状态，然后去做处理啊。所以我们这里边是不是就是来的时候，我们要做的处理是输入的什么数据，还有状态是什么，然后输出是不是就相当于输出是什么数据，还有状态变成什么样子呀？哎，所以它是这样的一个flat map的操作，所以这里边我们怎么定义，大家看它既然有这个option，可以判断这个状态是不是有没有，所以我这里边可以去做一个。

大家想一想，是不是可以直接做一个这个模式匹配啊，对吧，可以直接kiss一下。那。这里边首先这个input的这个这个data啊，它应该是什么类型，是不是一个sensor reading啊，呃，另外还有一个是我的那个状态对吧？呃，那个如果这里边假如说没东西的话，没有是一个闹对吧，因为这个它本身是一个option嘛，Option是不是可以是some，可以是none啊，所以这里边假如没有的话，我输出一个什么。

大家注意我要输出的是不是这样的一个二元组啊，对吧，一个是输出类型的一个传播ones啊，那大家会想到这个什么样的是一个传播one呢。可以序列化一次的这个呃，数据类型对吧，那是不是相当于list的一个列表就可以传传传播ones对吧？所以这里边我可以直接给一个二元组，给一个list是不是可以啊。呃，当然了，如果一开始就没有状态，那我这里边是不是干脆就什么东西都不要输出啊。对吧，我可以直接这么去定义，所以我直接给一个list empty，这不就完事了，然后另外还得要一个要一个状态，注意状态还是要一个option类型，那是不是这里面就变成some了？对some什么呢？状态变成什么呢？

状态，我们要定义的是上一次的那个温度值对不对，你既然已经有输入了，我是不是可以把温度值拿出来啊，存进状态对不对？input.temperature存进状态是不是就完事了？好，所以大家注意啊，这里边如果没有状态的话，也就是没有。数据来过，那么就将当前数据温度值存入状态，这是我们第一种情况，那当然还有第二种情况，对不对？第二种情况是如果有状态的话。

那么就应该怎么样，对，那是不是就应该与上次的温度值，也就是状态里面的那个值对吧，比较差值。如果大于阈值就输出报警好，所以这里边我们要判断的就是它的类型应该是input还是一个sensor reading，那这里边应该是什么？对，这里边就变成了一个some。啊，里边。应该是什么？哎，这这不是some啊，这里边是应该是我们是要定义这个输入参数对不对，所以这里边其实是last time对吧，它是一个some double，对这样的一个数据类型，那他应该输出什么呢。啊，这个时候那就得去case，就得做做操作了，对吧？啊，这里面就得做操作了，比方说我先定义一个diff，那那这个流程是不是跟下面一样啊，再想想是不是完全一样，只不过这里边要变成input.temperature去减last time对吧，然后再点ABS。

诶，这里边哦，这里面不能直接last time，因为它本身是一个sum类型，对，呃，是本身是一个那个option类型，对吧，所以我们需要get，然后在ABS，那当然这里边继续判断，如果diff要大于我们定义好的这个就不是，虽然是后的了，因为没有传入参数了，对吧，那只好写死了。大于十。这里面应该传输什么，是不是直接输出我们的那个就是输出的类型啊，不是报警信息，是我们定义好的三元组，所以这里边是注意这里边要输出的呢，大家注意是什么，是要先给一个二元组，对吧？哎，这个大家要先注意一下，是先要是不是先来一个例子呀。

List里边这才是那个三元组，对不对，然后外边还得有一个状态，状态是不是又是一个sum。那么some现在应该变成什么？是不是还是更新成input的，对temperature把它做一个更新对不对啊，那这里边的三元组是不是要写input，诶大家大家注意，这是list里边是不是还得有三元组的括号啊，input.id然后呃，这last temp，呃，没没有点了对吧？诶点get是不是得。

然后还有input.temperature对，大家看是不是这样的一个过程啊，对，所以我们看一下这里边报错哦，对，因为现在是不是只有if没有else啊，对吧，必须还得有else的实现，那else就比较简单了，ELSE4啊，当然我们也可以闹，也可以就把它直接正常输出对不对？呃，或者说我们直接输出一个空空是不是还是MT啊，Empty的时候注意状态是不是也得也得保存进去啊，因为即使是没有输出，所以这里边不能直接浪对吧，是不是状态这里边至少这个状态也得把当前的那个音input的状态更新进去啊，上一次的温度值更新进去，所以input.temperature放进去，诶大家看这样就实现完了。

这个过程大家看这个代码上好像简洁了一些，但是好像还有点儿难理解是吧？啊，当然这就是另外一种实现方式，大家如果要是觉得这种方式还不如我们直接写那个，呃，Rich Fla function觉得那个来的舒服的话，那大家就就就是一般就就用那种方式写就好了，但是这种方式大家也至少要知道，这也是状态编程的一种一种模式，好，那我们还是把它也简单测一下吧，那比方说我这里边。Process，呃的STREAM3，我们看看它的输出结果是不是符合预期。代码写起来，我们来看一下这里边的输出。好，首先诶，大家看这里边第一个出现的时候，是不是相当于他就不会报警了，为什么不会报警了，因为是不是我们指定了none时候的那种情况啊，哎，所以这种情况如果我们在呃之前的rich function里边，如果要去处理的话，是不是还得单独去处理一次啊，对吧，假如是第一次来的时候怎么办啊，那还得去单独处理，这里边就省得我们再去做这个操作了啊，所以这个就稍微的轻松一点啊，然后比方说我直接给一个25吧，给一个25大家想是不是直接报警了对吧，已经比这个大了，那如果我再来一个25.8，那当然这个不会报警对不对，如果再来一个36。

是不是又报警了，不管是大时大于超过十还是小于超过十，他都会报警，符合我们定义的行为啊，这就是这部分内容，大家可以把不同的这种状态编程的方法再好好的复习复习，自己实现一下。