Luke Hutchison created FLINK-6110:
-------------------------------------
Summary: Flink unnecessarily repeats shared work triggered by different blocking sinks, leading to massive inefficiency
Key: FLINK-6110
URL:
https://issues.apache.org/jira/browse/FLINK-6110 Project: Flink
Issue Type: Bug
Components: Core
Affects Versions: 1.2.0
Reporter: Luke Hutchison
After a blocking sink (collect() or count()) is called, all already-computed intermediate DataSets are thrown away, and any subsequent code that tries to make use of an already-computed DataSet will require the DataSet to be computed from scratch. For example, the following code prints the elements a, b and c twice in succession, even though the DataSet ds should only have to be computed once:
{code}
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> ds = env.fromElements("a", "b", "c").map(s -> { System.out.println(s); return s + s;});
List<String> c1 = ds.map(s -> s).collect();
List<String> c2 = ds.map(s -> s).collect();
{code}
This is problematic because not every operation is easy to express in Flink using joins and filters -- sometimes for smaller datasets (such as marginal counts) it's easier to collect the values into a HashMap, and then pass that HashMap into subsequent operations so they can look up the values they need directly. A more complex example is the need to sort a set of values, then use the sorted array for subsequent binary search operations to identify rank -- this is a lot easier to do using an array of sorted values, as long as that array fits easily in RAM.
However, any collect() or count() operation causes immediate execution of the Flink pipeline, which throws away *all* intermediate values that could be reused for future executions. As a result, code can be extremely inefficient, recomputing the same values over and over again unnecessarily.
I believe that intermediate values should not be released or garbage collected until after env.execute() is called.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
Locked
