Hi Flink Users, I'm using Flink to process a stream of records containing a text field. The records are sourced from a message queue, enriched as they flow through the pipeline based on business rules and finally written to a database. We're using the Ververica platform so it's running on Kubernetes.
The initial business rules were straightforward, e.g. if field X contains a certain word then set field Y to a certain value. For the implementation I began by looking at https://flink.apache.org/news/2020/01/15/demo-fraud-detection.html for inspiration. I ended up implementing a business rule as a Java class with a match-predicate & an action. The records enter the pipeline on a data stream which is joined with the rules in a broadcast stream and a ProcessFunction checks each record to see if it matches any rule predicates. If the record doesn't match any business rule predicates it continues on in the pipeline. If the record does match one or more business rule predicates it is sent to a side output with the list of business rules that it matched. The side output data stream goes through a RichAsyncFunction which loops through the matched rules and applies each one's action to the record. At the end, that enriched side-output record stream is unioned back with the non-enriched record stream. This all worked fine. I have some new business rules which are more complicated and require sending the record's text field to different pre-trained NLP models for prediction, e.g. if a model predicts the text language is X then update field Y to that value, if another model predicts the sentiment is positive then set some other field to another value. I'm planning on using seldon-core to serve these pre-trained models, so they'll also be available in the k8s cluster. I'm not sure about the best way to set up these model prediction calls in Flink. I could add in a new ProcessFunction in my pipeline before my existing enrichment-rule-predicate ProcessFunction and have it send the text to each of the prediction models and add the results for each one to the record so it's available for the enrichment step. The downside of this is that in the future I'm anticipating having more and more models, and not necessarily wanting to send each record to every model for prediction. e.g. I might have a business rule which says if the author of the text is X then get the sentiment (via the sentiment model) and update field Z, so it would be a waste of time doing that for all records. I had a look at stateful functions. There's an example in the statefun.io overview which shows having a stateful function for doing a fraud model prediction based on if an account has had X number of frauds detected in the last 30 days, so the key for the state is an account number. In my case, these model predictions don't really have any state - they just take input and return a prediction, they're more like a stateless lambda function. Also, I was wondering if I implemented these as stateful functions would I be able to make them available to other Flink jobs within the cluster, as opposed to having them as individual RichAsyncFunctions defined within a single Flink job and only available to that. The last thing which made stateful functions sound good was that at the moment all my business rules happen to be orthogonal, but I can imagine in the future where I might want one rule to be based on another one, and whereas regular dataflows have to be an acyclic graph stateful functions could support that. So, in summary: * Does this sound like a good use case for stateful functions? * Are stateful functions available to all Flink jobs within a cluster? Thanks, John.
