You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zentol <gi...@git.apache.org> on 2015/02/06 14:40:12 UTC
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
GitHub user zentol opened a pull request:
https://github.com/apache/flink/pull/370
[FLINK-785] Chained AllReduce
This a a preliminary PR to see whether I'm on the right track.
I'm wondering whether this would be everything needed to add a Chained AllReduce, before i continue with this issue. I tried it out and it appears to work, but wanted to make sure.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zentol/incubator-flink chained_all_reduce
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/370.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #370
----
commit 37b77670b517995498f83df452ed5b20754fc63e
Author: zentol <s....@web.de>
Date: 2015-02-06T13:38:05Z
[FLINK-785] Chained AllReduce
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74565386
I think we should use the chained reducer from this pull request and drop the chained combiner change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74236482
I think for the combiner, we cannot do this. The way it is done here easily blows up the memory, by collecting so many records. Bear in mind that a record is not necessarily small, in the ALS case, they were 500MB large. Times 10k, that is 5TB of buffered records. The non-chaining combiner variant makes sure that only two objects are ever around by passing records/fragments through the buffers in the local channels.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-73914381
Oh, just saw that you updated your PR.
Won't open a PR. You can have a look at my branch here: https://github.com/fhueske/flink/tree/chained_all_reduce
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-73404704
there's something funky going on with the tests here.
i got 2 failing tests in ObjectReuseITCase:
```
ObjectReuseITCase>JavaProgramTestBase.testJobWithoutObjectReuse:168->postSubmit:68->TestBaseUtils.compareResultsByLinesInMemory:223->TestBaseUtils.compareResultsByLinesInMemory:238 arrays first differed at element [0]; expected:<a,[10]0> but was:<a,[6]0>
ObjectReuseITCase>JavaProgramTestBase.testJobWithObjectReuse:120->postSubmit:68->TestBaseUtils.compareResultsByLinesInMemory:223->TestBaseUtils.compareResultsByLinesInMemory:238 arrays first differed at element [0]; expected:<a,[10]0> but was:<a,[6]0>
```
These two tests verify the wrong behaviour that occurs when object reuse is enabled but not accounted for. i thought this was generally treated as undefined behaviour, why are there tests for that?
the other 2 tests fail with NullPointerException when accessing the expected result.
```
ClosureCleanerITCase.after:52->TestBaseUtils.compareResultsByLinesInMemory:223->TestBaseUtils.compareResultsByLinesInMemory:234 » NullPointer
ClosureCleanerITCase.after:52->TestBaseUtils.compareResultsByLinesInMemory:223->TestBaseUtils.compareResultsByLinesInMemory:234 » NullPointer
```
i can't figure out why this occurs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/370#discussion_r24633778
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedAllGroupReduceCombineDriver.java ---
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.operators.chaining;
+
+import java.util.ArrayList;
+
+import org.apache.flink.api.common.functions.FlatCombineFunction;
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.operators.RegularPactTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ChainedAllGroupReduceCombineDriver<T> extends ChainedDriver<T, T> {
+ private static final Logger LOG = LoggerFactory.getLogger(ChainedAllGroupReduceCombineDriver.class);
+
+ // --------------------------------------------------------------------------------------------
+ private FlatCombineFunction<T> combiner;
+ private TypeSerializer<T> serializer;
+
+ private volatile boolean running = true;
+
+ private final ArrayList<T> values = new ArrayList();
+
+ // --------------------------------------------------------------------------------------------
+ @Override
+ public void setup(AbstractInvokable parent) {
+ @SuppressWarnings("unchecked")
+ final FlatCombineFunction<T> com = RegularPactTask.instantiateUserCode(this.config, userCodeClassLoader, FlatCombineFunction.class);
+ this.combiner = com;
+ FunctionUtils.setFunctionRuntimeContext(com, getUdfRuntimeContext());
+
+ this.objectReuseEnabled = this.executionConfig.isObjectReuseEnabled();
+
+ final TypeSerializerFactory<T> serializerFactory = this.config.getInputSerializer(0, this.userCodeClassLoader);
+ this.serializer = serializerFactory.getSerializer();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("ChainedAllGroupReduceCombineDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
+ }
+
+ @Override
+ public void openTask() throws Exception {
+ final Configuration stubConfig = this.config.getStubParameters();
+ RegularPactTask.openUserCode(this.combiner, stubConfig);
+ }
+
+ @Override
+ public void closeTask() throws Exception {
+ if (!this.running) {
+ return;
+ }
+ RegularPactTask.closeUserCode(this.combiner);
+ }
+
+ @Override
+ public void cancelTask() {
+ this.running = false;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ @Override
+ public Function getStub() {
+ return this.combiner;
+ }
+
+ @Override
+ public String getTaskName() {
+ return this.taskName;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ @Override
+ public void collect(T record) {
+ try {
+ values.add(objectReuseEnabled ? record : serializer.copy(record));
+ if (values.size() > 10000) {
--- End diff --
is there a more reasonable value for this? (basically anything whose reasoning goes beyond "i felt like using 10000")
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-73388209
Looks good in general.
You need to make sure though, that you obey the execution object re-usage settings.
That basically means you need to pay attention to the objects that you give to and receive from the user code and possibly copy them to new objects (see FLINK-1285 and @aljoscha's commits for details).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74247290
i know, so could we not do the same, except skipping the sorting part.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-73923184
sorry for not making it clear that i forced pushed some changes, forgot making a separate branch.
the changes you made to the driver are already in, sorry for wasting your time there.
i will include the changes to the test case.
The ChainedAllReduceDriver currently does not cause any problems.
I've added a ChainedAllGroupReduceCombineDriver aswell, which fails for one test case
```
ReplicatingDataSourceITCase.after:69->TestBaseUtils.compareResultsByLinesInMemory:223->TestBaseUtils.compareResultsByLinesInMemory:238 arrays first differed at element [0]; expected:<([50050]0)> but was:<([37525]0)>
```
may be related to FLINK-1521, once the newest test is complete I'll hopefully have a definite answer. when i force objectReuse in the global reduce all tests pass.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74247098
The synchronous combine driver stores them in the sort buffer in serialized fashion, which is save.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/370
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74236667
The chained reducer variant makes more sense. The question is, what is cheaper - serialized exchange through the in-mem-channel, or the copying. My gut feeling is that the copy is a bit better, so it may be worth it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-75030486
okay, I will the remove the chained combiner.
I'm curious though, the issue description specifically mentions the AllGroupReduce greatly benefiting from a chained version. did (or do) you have a specific approach in mind?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-73913500
Hi @zentol,
I had a look at your PR and the failing test case. The test was in place to ensure that objects are reused. For that the functions were implemented in a way that they will produce wrong results if objects are reused (therefore the unexpected result of 100 instead of 60). The problem was caused by the CollectionInputFormat which sends out different objects (instead of reusing the same object). When I changed that behavior and added some fixes to the ChainedAllReduceDriver, all tests pass.
I will open a PR against your branch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74247815
Then it is worse then the chained variant, because you save nothing and need buffer space. The chained variant forwards records/fragments as they are produces.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-785] Chained AllReduce / AllGroupReduce...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/370#issuecomment-74240805
the SynchronouesChainedCombineDriver stores multiple records as well doesn't it? except up to a fixed size instead of # of elements. could we not do the same here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---