Hi John, Thanks a lot for opening the JIRA ticket! If you are interested in contributing that to StateFun, I'm also happy to guide you with the contribution.
On Mon, Oct 5, 2020 at 10:24 PM John Morrow <johnniemor...@hotmail.com> wrote: > Thanks for the response Gordon, and that FlinkForward presentation - it's > been very helpful. > > I put in a JIRA ticket for it: > https://issues.apache.org/jira/browse/FLINK-19507 > > I did find this page: > https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.2/io-module/flink-connectors.html > and there are source/sink connectors for Pulsar ( > https://github.com/streamnative/pulsar-flink) - I'm guessing that's how I > should approach using Pulsar as an ingress/egress? > That is correct! The `SourceFunctionSpec` and `SinkFunctionSpec` are a means for users to bridge existing Flink sources and sinks to StateFun ingress / egress. The downside to that approach, is that even if you're purely using remote functions, you'd still have to provide an embedded module to add ingresses / egresses this way. Eventually it would be best (if we have several users requesting Pulsar) to have native support like Kinesis and Kafka so that users can define them textually in `module.yaml` definition files, but this approach you pointed definitely works for the time being. Cheers, Gordon > > Cheers, > John. > > ------------------------------ > *From:* Tzu-Li (Gordon) Tai <tzuli...@apache.org> > *Sent:* Monday 5 October 2020 03:21 > *To:* John Morrow <johnniemor...@hotmail.com>; user <user@flink.apache.org > > > *Subject:* Re: Stateful Functions + ML model prediction > > Hi John, > > It is definitely possible to use Apache Pulsar with StateFun. Could you > open a JIRA ticket for that? > It would be nice to see how much interest we can gather on adding that as > a new IO module, and consider adding native support for Pulsar in future > releases. > > If you are already using StateFun and want to start using Pulsar as an > ingress/egress already for current versions, there's also a way to do that > right now. > If that's the case, please let me know and I'll try to provide some > guidelines on how to achieve that. > > Cheers, > Gordon > > > On Fri, Oct 2, 2020, 1:38 AM John Morrow <johnniemor...@hotmail.com> > wrote: > > Hi Flink Users, > > I was watching Tzu-Li Tai's talk on stateful functions from Flink Forward ( > https://www.youtube.com/watch?v=tuSylBadNSo) which mentioned that Kafka & > Kinesis are supported, and looking at > https://repo.maven.apache.org/maven2/org/apache/flink/ I can see IO > packages for those two: statefun-kafka-io & statefun-kinesis-io > > > Is it possible to use Apache Pulsar as a Statefun ingress & egress? > > Thanks, > John. > > ------------------------------ > *From:* John Morrow <johnniemor...@hotmail.com> > *Sent:* Wednesday 23 September 2020 11:37 > *To:* Igal Shilman <i...@ververica.com> > *Cc:* user <user@flink.apache.org> > *Subject:* Re: Stateful Functions + ML model prediction > > Thanks very much Igal - that sounds like a good solution! > > I'm new to StateFun so I'll have to dig into it a bit more, but this > sounds like a good direction. > > Thanks again, > John. > > ------------------------------ > *From:* Igal Shilman <i...@ververica.com> > *Sent:* Wednesday 23 September 2020 09:06 > *To:* John Morrow <johnniemor...@hotmail.com> > *Cc:* user <user@flink.apache.org> > *Subject:* Re: Stateful Functions + ML model prediction > > Hi John, > > Thank you for sharing your interesting use case! > > Let me start from your second question: > > Are stateful functions available to all Flink jobs within a cluster? > > > Yes, the remote functions are some logic exposed behind an HTTP endpoint, > and Flink would forward any message addressed to them via an HTTP request. > The way StateFun works is, for every invocation, StateFun would attach the > necessary context (any previous state for a key, and the message) to the > HTTP request. > So practically speaking the same remote function can be contacted by > different Jobs, as the remote functions are effectively stateless. > > Does this sound like a good use case for stateful functions? > > > The way I would approach this is, I would consider moving the > business rules and the enrichment to the remote function. > This would: > > a) Eliminate the need for a broadcast stream, you can simply deploy a new > version of the remote function container, as they can be independy > restarted (without the need to restart the Flink job that contacts them) > b) You can perform the enrichment immediately without going through > an RichAsyncFunction, as StateFun, by default, invokes many remote > functions in parallel (but never for the same key) > c) You can contact the remote service that hosts the machine learning > model, or even load the model in the remote function's process on startup. > > So, in kubernetes terms: > > 1. You would need a set of pods (a deployment) that are able to serve HTTP > traffic and expose a StateFun endpoint. > 2. You would need a separate deployment for Flink that runs a StateFun job > 3. The StateFun job would need to know how to contact these pods, so you > would also need a kubernetes service (or a LoadBalancer) that > balances the requests from (2) to (1). > > If you need to change your business rules, or the enrichment logic you can > simply roll a new version of (1). > > > Good luck, > Igal. > > On Tue, Sep 22, 2020 at 10:22 PM John Morrow <johnniemor...@hotmail.com> > wrote: > > 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. > > >
