Hi Tim, 1. The filter is stored within the JobGraph which is persisted to a persistent storage if HA is enabled. Usually, this is either HDFS, S3 or any other highly available file system. It is just a serialized POJO. If you want to store your filter's state you would need to use Flink's state API [1]. 2. Unless you use Flink's state API, Flink won't be able to recover the numElementsSeen field. 3. I think stateful filters are ok to use if your filter needs to be stateful. Statefulness usually complicates things so if your function can be stateless, then I would recommend to make it stateless. However, there are some applications which strictly require statefulness.
[1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html Cheers, Till On Thu, Nov 7, 2019 at 2:11 PM Timothy Victor <vict...@gmail.com> wrote: > I have a FilterFunction implementation which accepts an argument in its > constructor which it stores as an instance member. For example: > > class ThresholdFilter implements FilterFunction { > > private final MyThreshold threshold; > > private int numElementsSeen; > > public ThresholdFilter(MyThreshold threshold) { > this.threshold = threshold; > } > > <more code> > > } > > The filter uses the threshold in deciding whether or not to filter the > incoming element. > > All this works but I have some gaps in my understanding. > > 1. How is this filter stored and recovered in the case of a failure. > Is it just serialized to a POJO and stored in the configured state backend? > > 2. When recovered will it maintain the state of all members (e.g. note > that I have a numElementsSeen member in the filter which will keep > incrementi for each element recevied). > > 3. Is this sort of thing even advisable for a filter? I'm guessing > Filters are meant to be reusable across operator instances. In which case > the state could be wrong after recovery? > > Thanks in advance > > Tim >
