Hi all, I have a use case where i need to enrich incoming events on Kafka topics 100msg/sec with a lookup on around 10 million entries in a table. Any suggestions for this architecture around usage of flink and/or db/cache?

I often chose to load in the enrichment data as a DataStream and join it on to the event stream with a

CoProcessFunction

. For example:

var eventsWithCustomerEnrichment = events
    .connect(enrichmentCustomerData)
    .keyBy(Event::userId, CustomerData::userId)
    .process(new EnrichmentEventWithCustomerData())

But as always the devil is in the details. Depends on whether you're able to preload all of the data into a DataStream, and whether you have enough resources to hold all of that enrichment state

Hi @Dr Ravi Tomar Enrichment is a super common and practical use-case for Flink and the 10m entries aren't terribly scary at all (obv. depending on how large the enrichment objects are). I think a few of the initial questions would be: • How frequently are the same keys encountered or are they repeated at all? This could influence how long entries need to stay in state/cache (via a TTL, forever, etc.) • Are the enrichment keys updated frequently? If so, this may require a separate stream for handling updates to avoid stale values being used for enrichment (this can be done via a totally separate stream, potentially a broadcast stream, likely populated via CDC) • Is this a "rolling" change? What I mean by that is does the job already exist and this is being added to it? Or is this a net-new Flink job? This could influence if you need to pre-load the state or if it would be acceptable to gradually build the state and enrich once the corresponding data is available. In my experience when I've needed to "load" lots of state (like 10m+ records), I've typically used a bootstrap process via the State Processor API which will seed my enrichment state with all of the necessary keys prior to data flowing.

Thanks @Andreas Bube and @Rion Williams for your inputs, @Rion Williams Frequency of updates: maybe 100-1000 records a day, mostly upserts. Key do not change mostly the value would change. This would be a complete new solution, i am skeptical on how to load 30gb of data during bootstrap (considering 3kb per record) Also this could be starting point and lookup data may even go upto 3x in a year. How can we make this system scalable. 100msg/sec is nothing for kafka or flink butlookup seems challenge to me. So, loading into RAM vs using some cache system vs distributing it across multiple nodes!! Any tested set of tool stack would help!!

Sure! This doesn't seem like a terribly scary amount of volume (even the 30GB of data), but I'll let others chime in as well. With it being a net-new job, bootstrapping the state does make sense to ensure that on your first run, you'll have everything that you need. You could introduce a separate stream of data to support the update use-case as well similar to the example that @Andreas Bube provided (connect your update stream responsible for updating the values in state with the operator responsible for enrichment). > Frequency of updates: maybe 100-1000 records a day, mostly upserts. ... mostly value would change This is pretty common. I'd say the biggest complication would be getting all 30GB of your records into a state that's consumable for your bootstrap process (e.g., if it's in a database consider leveraging a CDC connector to bring it into a Kafka topic via a snapshot). This same CDC connector would be used within your job to listen to updates and update state accordingly. I know I've bootstrapped jobs with ~100M+ records and it worked like a charm. > Also this could be starting point and lookup data may even go upto 3x in a year. How can we make this system scalable. 100msg/sec is nothing for kafka or flink butlookup seems challenge to me. So keep in mind, if you are storing the lookup data in Flink state, the lookup is going to be trivial. Think of it closer to an "in memory" lookup from a hashmap or cache (in this case backed by RocksDB, so "technically" disk). You aren't going to be issuing external calls to your database for each record, which is typically where people will be a bit more fearful in terms of increasing volumes. Depending on your deployment model -- you can configure things like autoscaling as well. Assuming your job is designed in a matter that doesn't prohibit parallelization (e.g., keying by low cardinality keys), scaling out should not be a problem. > So, loading into RAM vs using some cache system vs distributing it across multiple nodes!! Yup this is what Flink excels at. While having adequate RAM is important, RocksDB is going to typically be how your state is backed (disk instead of RAM). It's highly scalable, configurable and ... far less expensive than say a proper distributed cache like Redis, etc. All of your state isn't going to be loaded into memory at any given time.

Thankyou very much @Rion Williams , i will give it a try and definitely share my experience here😊

I can only second @Rion Williams. I don't think Flink with RocksDB and SSDs will have any issues having 30GB in state. Heck, my current project has 24GB of geo enrichment data in state and in runs just fine on my little MacBook laptop when testing locally. An alternative to the State Processor API is to use the Hybrid Source. Give it a look. We're using this on our current project to process 4 years of historical and live data in the same job.

I agree @Andreas Bube! I’ve been tempted to do a write up for a few of these common patterns to help folks figure out the best approach to solve this riddle. I’ve seen Hybrid Source a bit but haven’t had the opportunity to put it into practice (I’ve also seen other patterns where bootstrapping is initially done in the same job; I.e; job runs and seeds all enrichment data and stops, starts back up, detects it’s been bootstrapped, and runs as normal). But overall — @Dr Ravi Tomar you should be totally fine with the proposed state and even under much larger potential growth.