请教初始化系统缓存的问题

["haoxinyue@163.com" <ha...@163.com>]

大家好，初学flink，版本1.8.1。想请教一个思路问题：在物联网系统中会实时处理一些设备上来的状态数据，但是设备元数据信息或者基础数据会存储在类似MySQL数据库中，当实时流数据上来的时候需要这些基础数据进行计算。但是因为性能问题，不能实时的去数据库获取，所以需要在系统启动的时候缓存起来，然后再开始收数据或者开始处理数据。数据来源kafka。
请教2个问题：

        1. 有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
        2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）

谢谢。



haoxinyue@163.com

Re: Re: 请教初始化系统缓存的问题

["haoxinyue@163.com" <ha...@163.com>]

非常感谢，大家同行。
我们目前是确实按照类似你说的这些方式去完成的。但是我们始终觉得应该有更加flink的方式优雅完成，就像维表join。之前一直没有细看，谢谢提醒。



haoxinyue@163.com
 
发件人： 高博
发送时间： 2019-09-24 16:03
收件人： user-zh
主题： Re: 请教初始化系统缓存的问题
你好，
 
我这里提供几个思路，我们公司做车联网的，目前线上运行的程序，都需要处理你说的这些场景。
 
1.
有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
     目前我们需要使用的外部缓存数据，包含需要调用接口的，需要读取数据库的基础数据。
     针对调用接口，我们使用的guava的异步缓存刷新策略
     针对数据库中的基础数据，我们类似懒加载，当第一条数据过来的时候，我们会锁住流，等待所有的数据都读取完了，整个流才继续执行。
 
 
 
2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）
     针对基础数据的刷新，可以分两个阶段来做。
     阶段一，可以启动一个定时刷新的线程，定时增量从数据库中读取数据来刷新缓存中的数据就OK。
 
阶段二，可以基于canal+kafka+广播来做。简单来说，就是用canal监听数据库的变换，然后推送到kafka中，你的流中消费这个kafka
topic，然后获取到基础数据后，通过广播的形式来传递到各个算子。
 
另外，社区里有人使用的是流表与维表join的方式来做的，维表就是基础数据，维表的加载和刷新都是通过异步算子来实现的。具体的，你可以搜一下。
感觉这个更符合Flink的体系。我前面提到的，其实在java程序和在Flink中都会这么处理。
 
 
haoxinyue@163.com <ha...@163.com> 于2019年9月24日周二 上午10:45写道：
 
>
> 大家好，初学flink，版本1.8.1。想请教一个思路问题：在物联网系统中会实时处理一些设备上来的状态数据，但是设备元数据信息或者基础数据会存储在类似MySQL数据库中，当实时流数据上来的时候需要这些基础数据进行计算。但是因为性能问题，不能实时的去数据库获取，所以需要在系统启动的时候缓存起来，然后再开始收数据或者开始处理数据。数据来源kafka。
> 请教2个问题：
>
>         1.
> 有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
>         2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）
>
> 谢谢。
>
>
>
> haoxinyue@163.com
>

Re: 请教初始化系统缓存的问题

[高博 <ga...@gmail.com>]

你好，

我这里提供几个思路，我们公司做车联网的，目前线上运行的程序，都需要处理你说的这些场景。

1.
有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
     目前我们需要使用的外部缓存数据，包含需要调用接口的，需要读取数据库的基础数据。
     针对调用接口，我们使用的guava的异步缓存刷新策略
     针对数据库中的基础数据，我们类似懒加载，当第一条数据过来的时候，我们会锁住流，等待所有的数据都读取完了，整个流才继续执行。



2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）
     针对基础数据的刷新，可以分两个阶段来做。
     阶段一，可以启动一个定时刷新的线程，定时增量从数据库中读取数据来刷新缓存中的数据就OK。

 阶段二，可以基于canal+kafka+广播来做。简单来说，就是用canal监听数据库的变换，然后推送到kafka中，你的流中消费这个kafka
topic，然后获取到基础数据后，通过广播的形式来传递到各个算子。

另外，社区里有人使用的是流表与维表join的方式来做的，维表就是基础数据，维表的加载和刷新都是通过异步算子来实现的。具体的，你可以搜一下。
感觉这个更符合Flink的体系。我前面提到的，其实在java程序和在Flink中都会这么处理。


haoxinyue@163.com <ha...@163.com> 于2019年9月24日周二 上午10:45写道：

>
> 大家好，初学flink，版本1.8.1。想请教一个思路问题：在物联网系统中会实时处理一些设备上来的状态数据，但是设备元数据信息或者基础数据会存储在类似MySQL数据库中，当实时流数据上来的时候需要这些基础数据进行计算。但是因为性能问题，不能实时的去数据库获取，所以需要在系统启动的时候缓存起来，然后再开始收数据或者开始处理数据。数据来源kafka。
> 请教2个问题：
>
>         1.
> 有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
>         2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）
>
> 谢谢。
>
>
>
> haoxinyue@163.com
>

Re: 请教初始化系统缓存的问题

[Terry Wang <zj...@gmail.com>]

你好，可以考虑在open方法里启动一个定时的线程去取mysql里去数据和进行缓存更新。
当有新数据流入到你的系统中时，可以判断定时线程数据加载是否完成，当数据加载完成后再进行数据处理。
希望能有帮助~

Best,
Terry Wang



> 在 2019年9月24日，上午10:45，haoxinyue@163.com 写道：
> 
> 大家好，初学flink，版本1.8.1。想请教一个思路问题：在物联网系统中会实时处理一些设备上来的状态数据，但是设备元数据信息或者基础数据会存储在类似MySQL数据库中，当实时流数据上来的时候需要这些基础数据进行计算。但是因为性能问题，不能实时的去数据库获取，所以需要在系统启动的时候缓存起来，然后再开始收数据或者开始处理数据。数据来源kafka。
> 请教2个问题：
> 
>        1. 有什么方式能否保证数据开始处理的时候，基础数据已经缓存好了，可以在流处理中获取到（类似valueState），算子的open函数似乎不能写到valueState里面去，只能初始化state。
>        2. 当基础数据在数据库中有变化的时候，如何实时的通知到处理流中，更新缓存？（广播流？）
> 
> 谢谢。
> 
> 
> 
> haoxinyue@163.com