flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Flavio Pompermaier <pomperma...@okkam.it>
Subject Case of possible join optimization
Date Tue, 08 Sep 2015 07:59:38 GMT
Hi to all,

I have a case where I don't understand why flink is not able to optimize
the join between 2 datasets.

My initial code was basically this:

DataSet<Tuple2<String, List<ThriftObj>>> bigDataset = ...;//5.257.207
elements
DataSet<Tuple2<String,List<MyObject>>> attrToExpand = ...;//65.000 elements

DataSet<Tuple2<String, IndexAttributeToExpand>> tmp =
attrToExpand.joinWithHuge(subset).where(0).equalTo(0).projectFirst(0,1).projectSecond(1);

This job wasn't able to complete on my local machine (from Eclipse) because
Flink was giving me the following error:

Hash join exceeded maximum number of recursions, without reducing
partitions enough to be memory resident. Probably cause: Too many duplicate
keys.

This was because in attrToExpand the List<MyObject> could be quite big.
Indeed, changing that code to the following make everything work like a
charm:

DataSet<Tuple2<String, List<ThriftObj>>> subset =

attrToExpand.project(0).joinWithHuge(bigDataset).where(0).equalTo(0).getInput2();

DataSet<Tuple3<String, List<MyObject>, List<ThriftObj>>> atomSubset
=

attrToExpand.join(subset).where(0).equalTo(0).projectFirst(0,1).projectSecond(1);


Isn't something impossible for Flink to optimize my initial code into the
second? I was convinced that Flink was performing a join only on the keys
before grabbing also the other elements of the Tuples into memory..am I
wrong?

Best,
Flavio

Mime
View raw message