00:00
在算子状态当中,有一类其实是比较特殊的,那就是广播状态。其实从概念上来讲,广播状态甚至比一般的算子状态还要容易理解,那就是说所有状态都一样,所有当前算子的并行子任务,他们获取到的状态实例。尽管不是同一份儿,但是我们应该保证他们里边的内容是完全相同的,这样的状态就叫做广播状态,那并行度如果要做调整的时候,我们之前说算子状态里边Lisa state uniona state,他们的区别,或者说他们的主要考虑的核心也都在这里啊,就是需要把当前的状态要打散,然后再去做重兄分配。而对于广播状态而言呢,哎,根本就不需要有任何的考量,直接复制一份,如果是增加的话,那多复制一份。发出去就可以了,如果减少一个并行度的话,那相当于多出来的那一个状态直接砍掉就可以了,因为别的地方都是完全一样的。
01:06
看起来很简单,但是在使用的过程当中呢,广播状态它其实是比较麻烦的,而且非常的特别,所以接下来我们就专门讲解一下广播状态到底是怎么回事。广播状态呢,其实在前面我们讲到connect,讲到河流操作的时候已经有过设计,那主要就是我们当时讲到了。所谓的。广播连接流就是一个data STEM,如果要连接上一个广播流的话,那么得到的就是一个广播连接流,然后再做对应的process操作,那么就可以进行两条流的合合并连接啊。那么在这个过程当中怎么样能够得到一个广播流呢?那就是我们说的需要去定义广播状态,然后调用data stream broadcast的方法,这样的话就可以得到一个广播。
02:06
那到底在什么场景下会有这样的一个应用的需求呢?简单来讲,那就是最普遍的场景,就是需要做动态配置,或者要定义动态规则的时候。我们在很多情况下啊,处理流数据的时候,我们可能需要为当前的这个数据定义一些配置的规则,比方说我们在获取用户行为的时候,可能我们想要检测当前用户,诶,先发生一个比方说先登录,然后再去浏览,或者先登录然后再去下单,我们可能想要检测一个用户这些行为的。一个匹配规则先发生什么后发生什么,但是这个匹配规则呢,有可能随着时间变化,或者随着我们环境的不同可能会变,我们之前可能考虑到的是想要检测用户登录之后下单的这样的行为。那接下来呢,有可能。
03:05
考虑的是用户登录之后啊,有可能是做了一些购买,或者说做了一些访问页面点击这样的一些简单行为。这些都属于我们可以配置的东西。对于不变的配置而言,假如说我们要检测的模式就已经定了,就是用户上来之后先登录,然后去访问一个页面。那就不要改了,我们可以直接把这个规则写死在代码里面,当然了,如果说要发生变化的话,我们也可以考虑到把它写到外部的一个配置文件里面去做一个读取,做一个加载。但是我们想到如果它是动态的不停的改变的话,写到外部文件,那就不是特别的方便,诶,因为如果我们写到外部文件的话,那还需要定期的去扫描这个配置文件,如果发现改变了,那我们需要去更新,然后还需要再去把这个应用整个去做一个重启。
04:01
重新去做一个应用。这个过程其实是代价非常高的,所以我们的解决办法其实还是流处理里边事件驱动的思路,哎,我们可以把这个动态的配置项或者动态的规则,它不也是数据吗?我们把这些数据当成一个流数据不就可以了吗?所以接下来我们要做的其实就是原始的数据流和这条动态的规则流。进行一个连接,然后接下来就可以实时的更新这些配置项,或者更新当前的规则,然后应用到数据流里边去做计算。这个过程其实就是考虑到动态配置或者动态规则实时更新的,这个过程就可以用到广播流和广播状态。那在这个过程当中,我们知道对于这个规则而言,或者是配置而言,它是全局有效的,我们不应该针对不同的病情、子任务发送不同的配置或者是规则,我们应该是实时的把它们全部更新。
05:12
当然这就涉及到需要把同一份儿数据广播给所有的下游的子任务,那这个广播出去的东西,我们应该用一个状态把它保存起来啊,那这个时候呢,这就是所谓的广播状态,那广播状态的底层。跟其他的算子状态是不一样的,并不是列表,而是一个key value的形式,所以本质上是一个映射状态,Maps。那我们回忆一下的话,在代码上其实调用的过程是首先针对一个data stream去调用点broadcast方法,然后这个方法里边需要传入一个map state的描述器,这个描述器说明我们当前广播状态的名称和类型,然后就可以声明出来接下来要把什么样的状态广播出去了,就得到了一个广播流,然后呢,再把原始的数据流跟广播流连接到一起,就可以得到广播连接流了。
06:16
那所以整个的代码应用其实是这样的一个过程,在之前我们讲到广播连接里讲到connect的时候,其实已经提到过,现在可以再复习一下啊,那就是首先我们基于当前的一个数据,这个是rulere,也就是说当前的这个流里边呢,全部都是我们动态的那些规则,或者说动态的那些配置项。作为数据放在这个data stream里面,然后这个data stream去调用一个broadcast方法传入一个map的描述器。定义好了广播状态的类型,然后就可以得到一个广播流,然后再把原始的数据数据流跟当前的广播流做一个connect,再去做process处理,就可以得到我们想要的。
07:07
最终的结果。那在具体处理计算的过程当中,这里的processt方法应该要传入一个broadcast process function,或者一个key的broadcast process,就是我们说的。处理函数process家族里边的最后两,那么他们在定义的过程当中呢?关键是要实现里边的process element和process broadcast element2个方法,类似于我们讲的Co process方式,里边一国两制的那个process element1和process element2,这也是分别处理两条流,只不过一条是真正的数据流,另外一条相当于是广播流啊,我们是从这个广播状态里边需要拿出对应的数据来,然后去进行处理计算。那这里边我们会发现啊,就会涉及到对应的上下文了,这里面都有ctx在广播broadcast element处理广播流数据的,这个里边我们拿到的就是一个,这个是可以获取当前的广播状态,也可以去更改,也可以去处理的。
08:19
而上面的这个process element呢,它的上下文是一个read only,也就是说它可以获取到当前的广播状态,但是呢,只能读取,不能更改。啊,这样的话我们就知道了,对于广播状态的改变,只能通过广播流里的数据来更换我的规则,更更新我们的配置,而对于普通数据流呢,只能是基于规则去做计算,去做处理,这就是完整的处理流程。
我来说两句