Implement bunch of transformations applied to same source stream in Apache Flink in parallel and combine result

1
votes

Could you please help me - I'm trying to use Apache Flink for machine learning tasks with external ensemble/tree libs like XGBoost, so my workflow will be like this:

  • receive single stream of data which atomic event looks like a simple vector event=(X1, X2, X3...Xn) and it can be imagined as POJO fields so initially we have DataStream<event> source=...
  • a lot of feature extractions code applied to the same event source: feature1 = source.map(X1...Xn) feature2 = source.map(X1...Xn) etc. For simplicity lets DataStream<int> feature(i) = source.map() for all features
  • then I need to create a vector with extracted features (feature1, feature2, ...featureK) for now it will be 40-50 features, but I'm sure it will contain more items in future and easily can contains 100-500 features and more
  • put these extracted features to dataset/table columns by 10 minutes window and run final machine learning task on such 10 minutes data

In simple words I need to apply several quite different map operations to the same single event in stream and then combine result from all map functions in single vector.

So for now I can't figure out how to implement final reduce step and run all feature extraction map jobs in parallel if possible. I spend several days on flink docs site, youtube videos, googling, reading Flink's sources but it seems I'm really stuck here.

The easy solution here will be to use single map operation and run each feature extraction code sequentially one by one in huge map body, and then return final vector (Feature1...FeatureK) for each input event. But it should be crazy and non optimal.

Another solution for each two pair of features use join since all feature DataStreams has same initial event and same key and only apply some transformation code, but it looks ugly: write 50 joins code with some window. And I think that joins and cogroups developed for joining different streams from different sources and not for such map/reduce operations.

As for me for all map operations here should be a something simple which I'm missing.

Could you please point me how you guys implement such tasks in Flink, and if possible with example of code?

Thanks!

1
apache-flinkflink-streaming
Your first approach seems optimal to me, even though you would have one huge map operation which extracts all the features for a single data sample, this process would be applied in parallel to all the events coming in. So, it would be parallelly processed anyway. Also, in your second approach, the entire process of join would depend on the slowest feature extractor, so it not optimal compared to first. - Biplob Biswas
Thanks, make sense! another solution is to use some instrument with sequence barriers implemented like LMAX Disruptor. I already use Aeron as communication layer, so all messages move through Aeron first, then they come to Flink, and it seems make sense to put map calculation result to ring buffer again to speed up throughput here. And when all features will be calculated - pack result to vector and save to final table. - Andrey Salnikov
One more common to Flink approach: union(feature1, feature2...featureK) sources, so stream can be imaged as f13 f21 f32 f41 f12 f11 - elements from all events and all feature parts and then use process function which will put all unordered parts of vectors to some state, and flush completed vectors to ctx.collect() in right order which finally will be pointed to the features sink. - Andrey Salnikov
@AndreySalnikov, any new updates on this? How did you finally approached this? - anaray

1 Answers

0
votes

What is the number of events per second that you wish to process? If it’s high enough (~number of machines * number of cores) you should be just fine processing more events simultaneously. Instead of scaling with number of features, scale with number of events. If you have a single data source you still could randomly shuffle events before applying your transformations.

Another solution might be to:

  1. Assign unique eventId and split the original event using flatMap into tuples: <featureId, Xi, eventId>.
  2. keyBy(featureId, eventId) (or maybe do random partitioning with shuffle()?).
  3. Perform your transformations.
  4. keyBy(eventId, ...).
  5. Window and reduce back to one record per event.