I want to join two stream. One of them is events kind of data, and the other one is dimensional data (i.e. users) and the size of it is hundreds of GBs ~ TBs. What’s the recommended way to join them? Look up join?

This is the sort of use case for which lookup joins were designed. You'll probably want to turn on caching to make the performance tolerable. One alternative is to stream in the dimensional data and have it held in Flink state. In fact, there are many ways to approach this -- see https://github.com/knaufk/enrichments-with-flink for a deep dive into this topic. https://docs.immerok.cloud/docs/explainers/flink/enriching-data-for-machine-learning-with-apache-flink/ is another take on this topic.

Thanks David! Will take a look!

Eric, you have to make a decision whether to use the table api or the stream api. We decided to use the stream api - in this case a

KeyedBroadcastProcessFunction

- we found it easier to reason about during the join function, what to do with elements that don't match and also updating the broadcast state at runtime when the external data set (in your case Users) changes.

Thanks for the suggestion Basil! With

KeyedBroadcastProcessFunction

, have you seen any issues when the dimension data sometimes arrive later than the events data which causes some of the records in the event stream not get enriched by the dimension data?

If you are concerned about that, you can buffer the early arriving events in keyed state, and then when the broadcast data arrives, use applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function) to do the enrichment then.

Thanks David. I have another question for a special case. In one pipeline, we want to join three dimension streams in real-time. In this case, does above solutions apply here? Maybe my question can be breakdown into two parts: • How to enrich a stream with 2+ dimension streams? • How to do that if all streams being joined are dimensional?

Under the covers, Flink has an n-ary operator that Flink SQL can use to implement n-way joins, but it's not currently exposed via the DataStream API. So you are forced to either merge (union) streams before joining them, or perform the joins in a cascading fashion (2 streams at a time).

Thanks David! Is there any link that has some more detailed explanation of that? I guess the Reducing Latency with Parallel Enrichments is similar to what you suggested, but I didn’t get how exactly the

MERGE

(union) helps here… and I couldn’t find the n-ary operator in Flink SQL docs…

The n-ary operator isn't user visible. I'm not really sure in what cases it gets used; it may not apply to the case you are interested in. You would just do an n-way SQL join, and hope for the best. The idea behind using union would be to merge all of the dimension streams into one broadcast stream, and then cope with that in the KeyedBroadcastProcessFunction doing the join. You'd have to come up with some (probably awkward) unified type to use as the output of the union.

Hello, reusing this thread because I have a very concrete question about a strategy for stream extensions. Is it possible to use rocksdb as a cache for LookupFunctions that are backed up by an external service (i.e. a gRPC service)