Implement the convenience methods count and collect in DataSet

[mxm]

These methods provide convenience for the API user to get intermediate results into the program.

[zentol]

Does this work when using count() or collect() multiple times in the same plan?

you use this.getExecutionEnvironment().getIdString() to identify the accumulators, and as far as i see this value is the same for all operators in a plan. As such, multiple usages of count() will return the sum count of all datasets, and collect() will throw an exception since the same key is used when calling getRuntimeContext().addAccumulator(id, accumulator)

[zentol]

well nevermind, just saw that you're calling this.getExecutionEnvironment().execute(); within these methods, so when using it, count() or collect() would be the last method called. I find that very odd behaviour, i can't think of another method that executes than right away.

[zentol]

Why didn't you implement these as an identity-(map)function and save the information to unique accumulators (maybe identified by a user-supplied ID? you could insert them into any plan at any point any number of times, i can see that being useful.

[mxm]

Why didn't you implement these as an identity-(map)function and save the information to unique accumulators (maybe identified by a user-supplied ID?

Thank you for your feedback. The idea is to have intermediate results available to the user program within the execution of a Flink program. Currently, this is only possible through RichFunction which provide access to the RuntimeContext.

[StephanEwen]

I think executing right away makes sense, because we need the long value or the List<T> immediately. This is a new type of methods, where you fetch back data to the driver program.

As for IDs, I agree that a random generated ID per method makes sense.

[uce]

Very cool! I think this is a great addition and we will improve on the performance soon :)

@zentol, as @StephanEwen said it makes sense to execute right away. You can think of this as "actions" from Spark, which get results back to the user. He can't use a map function, because he doesn't emit any records, but just adds them to the accumulator.

I've tried this out locally and it works, but you have to add the random IDs as you run into problem with multiple actions in the same program otherwise.

With this change, we also should allow to have prorams without sinks. Currently, you get a warning if you only have "actions" like collect/count without a sink.

Before we can merge this, you should also add test cases for both operators and make sure that it works with multiple actions per program.

[zentol]

hmm...alright, i can see the point.

Doesn't executing right away carries the risk of it being inefficient when using them multiple times though? Since it effectively means executing multiple jobs within the same program (i think ... ), any common part of the jobs are done an extra time. (if I'm wrong here skip the rest)

example:

List l1 = A.map(X).map(Y).collect();
List l2 = A.map(X).map(Z).collect();

<some user code using l1 & l2>

this would result in 2 jobs being executed, with map(X) being executed twice. whereas

B = A.map(X);
B.map(Y).collect("c1");
B.map(Z).collect("c2");

JobExecutionResult jre = env.execute()
List l1 = jre.getAccumulatorResult("c1");
List l2 = jre.getAccumulatorResult("c2);

<some user code using l1 & l2>

would only be 1 job, with map(X) done only once. it is not as pretty (by a fair margin i admit), but in line with the current API.

[mxm]

I changed the code to generate a unique identifier for each call of count or collect.

@zentol Executing in a lazy fashion could be implemented but would require additional changes to the API. The methods should then return something like a LocalDataSet which lets the user retrieve the accumulator result once the job has been executed.

[zentol]

why additional changes? you can already retrieve the accumulators by using the JobExecutionResult.

the only necessary change i see is removing the execute() within collect() and count(), and now that you generate the ID's yourself returning the ID, so the accumulators can be accessed later.

B = A.map(X);
String id1 = B.map(Y).collect();
String id2 = B.map(Z).collect();

JobExecutionResult jer = env.execute()
List l1 = jer.getAccumulatorResult(id1);
List l2 = jer.getAccumulatorResult(id2);

<some user code using l1 & l2>

why would this not be sufficient?

[StephanEwen]

Let us change this to close()

[StephanEwen]

With the scheduler and intermediate data set enhancements coming up for 0.9 soon, this is now quite feasible to use. I suggest to merge it once the inline comments are addressed.

[mxm]

I fixed the proposed changes and rebased to the current master. The changes are only reflected in the Java API and need to be added to the Scala API as well.

[mxm]

I've implemented count and collect in the Scala API. There is still a problem with the ListAccumulator for non-primitive Objects (e.g. not Integer or Long) probably due to Object reuse.

[mxm]

Looks like this is now ready to merge.

@zentol I understand your concern. However, I think that it is much easier to execute in this way. Most of the times, the user probably wants just one accumulator result and not multiple. This is supposed to be a convenience function.

[StephanEwen]

@zentol You are right, for the time being, that this results in parts in repeated execution. While not totally unavoidable in all cases, the code going in soon about caching intermediate results will help there big time.

[mxm]

I squashed the commits and rebased to the current master. Any objections against merging this?

[rmetzger]

Why are you moving netty to flink-core ?

[mxm]

Simply because AbstractID depends on io.netty.buffer.ByteBuf

[rmetzger]

Allright, that makes sense

[uce]

Actually, only a single convenience method in AbstractID depends on this. I would propose to refactor that method into one of the existing utility classes in flink-runtime and keep the Netty dependency in flink-runtime.

[StephanEwen]

Let us not move netty to flink-core. Let us rather pull that method out of the AbstractID, or, not use AbstractID at in your code, but the Java UUID.

[mxm]

I moved the writeInto(ByteBuf buf) method to the subclasses of AbstractID used by NettyMessage. This is the only context where the method is being used and the subclasses already have another fromByteBuf(ByteBuf buf) method.

[mxm]

If there are no objections, I will merge this once Travis passes.