You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/01/15 19:00:14 UTC
[2/2] flink git commit: [FLINK-2716] [gelly] [apis] New checksum
method on DataSet and Graph
[FLINK-2716] [gelly] [apis] New checksum method on DataSet and Graph
This closes #1462
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a8be52a3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a8be52a3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a8be52a3
Branch: refs/heads/master
Commit: a8be52a3309c5e68fe865a421fb60fdb3cc56d16
Parents: 98fad04
Author: Greg Hogan <co...@greghogan.com>
Authored: Tue Dec 15 14:25:54 2015 -0500
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Jan 15 18:58:57 2016 +0100
----------------------------------------------------------------------
.../java/org/apache/flink/api/java/Utils.java | 93 ++++++++++++++++++++
.../flink/api/java/utils/DataSetUtils.java | 21 +++++
.../flink/graph/scala/utils/package.scala | 50 +++++++++++
.../scala/test/util/GraphUtilsITCase.scala | 46 ++++++++++
.../apache/flink/graph/utils/GraphUtils.java | 38 ++++++++
.../flink/graph/test/util/GraphUtilsITCase.java | 58 ++++++++++++
.../apache/flink/api/scala/utils/package.scala | 21 +++++
.../flink/test/util/DataSetUtilsITCase.java | 13 +++
.../api/scala/util/DataSetUtilsITCase.scala | 12 +++
9 files changed, 352 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index 038b58c..2edc533 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -20,7 +20,9 @@ package org.apache.flink.api.java;
import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.SerializedListAccumulator;
+import org.apache.flink.api.common.accumulators.SimpleAccumulator;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType;
@@ -158,6 +160,97 @@ public final class Utils {
}
}
+ public static class ChecksumHashCode implements SimpleAccumulator<ChecksumHashCode> {
+
+ private static final long serialVersionUID = 1L;
+
+ private long count;
+ private long checksum;
+
+ public ChecksumHashCode() {}
+
+ public ChecksumHashCode(long count, long checksum) {
+ this.count = count;
+ this.checksum = checksum;
+ }
+
+ public long getCount() {
+ return count;
+ }
+
+ public long getChecksum() {
+ return checksum;
+ }
+
+ @Override
+ public void add(ChecksumHashCode value) {
+ this.count += value.count;
+ this.checksum += value.checksum;
+ }
+
+ @Override
+ public ChecksumHashCode getLocalValue() {
+ return this;
+ }
+
+ @Override
+ public void resetLocal() {
+ this.count = 0;
+ this.checksum = 0;
+ }
+
+ @Override
+ public void merge(Accumulator<ChecksumHashCode, ChecksumHashCode> other) {
+ this.add(other.getLocalValue());
+ }
+
+ @Override
+ public ChecksumHashCode clone() {
+ return new ChecksumHashCode(count, checksum);
+ }
+
+ @Override
+ public String toString() {
+ return "ChecksumHashCode " + this.checksum + ", count " + this.count;
+ }
+ }
+
+ @SkipCodeAnalysis
+ public static class ChecksumHashCodeHelper<T> extends RichOutputFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String id;
+ private long counter;
+ private long checksum;
+
+ public ChecksumHashCodeHelper(String id) {
+ this.id = id;
+ this.counter = 0L;
+ this.checksum = 0L;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public void open(int taskNumber, int numTasks) {}
+
+ @Override
+ public void writeRecord(T record) throws IOException {
+ counter++;
+ // convert 32-bit integer to non-negative long
+ checksum += record.hashCode() & 0xffffffffL;
+ }
+
+ @Override
+ public void close() throws IOException {
+ ChecksumHashCode update = new ChecksumHashCode(counter, checksum);
+ getRuntimeContext().addAccumulator(id, update);
+ }
+ }
+
+
// --------------------------------------------------------------------------------------------
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-java/src/main/java/org/apache/flink/api/java/utils/DataSetUtils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/utils/DataSetUtils.java b/flink-java/src/main/java/org/apache/flink/api/java/utils/DataSetUtils.java
index 9a1e952..01d801e 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/utils/DataSetUtils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/utils/DataSetUtils.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.java.utils;
import com.google.common.collect.Lists;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.api.common.functions.RichMapPartitionFunction;
import org.apache.flink.api.java.DataSet;
@@ -30,6 +31,7 @@ import org.apache.flink.api.java.operators.GroupReduceOperator;
import org.apache.flink.api.java.operators.MapPartitionOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.AbstractID;
import org.apache.flink.util.Collector;
import java.util.Collections;
@@ -246,6 +248,25 @@ public final class DataSetUtils {
return new GroupReduceOperator<>(mapPartitionOperator, input.getType(), sampleInCoordinator, callLocation);
}
+ // --------------------------------------------------------------------------------------------
+ // Checksum
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Convenience method to get the count (number of elements) of a DataSet
+ * as well as the checksum (sum over element hashes).
+ *
+ * @return A ChecksumHashCode that represents the count and checksum of elements in the data set.
+ */
+ public static <T> Utils.ChecksumHashCode checksumHashCode(DataSet<T> input) throws Exception {
+ final String id = new AbstractID().toString();
+
+ input.output(new Utils.ChecksumHashCodeHelper<T>(id)).name("ChecksumHashCode");
+
+ JobExecutionResult res = input.getExecutionEnvironment().execute();
+ return res.<Utils.ChecksumHashCode> getAccumulatorResult(id);
+ }
+
// *************************************************************************
// UTIL METHODS
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/utils/package.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/utils/package.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/utils/package.scala
new file mode 100644
index 0000000..954e6c3
--- /dev/null
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/utils/package.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.graph.scala
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.Utils.ChecksumHashCode
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.utils._
+
+import scala.reflect.ClassTag
+
+package object utils {
+ /**
+ * This class provides utility methods for computing checksums over a Graph.
+ *
+ * @param self Graph
+ */
+ implicit class GraphUtils[K: TypeInformation : ClassTag, VV: TypeInformation : ClassTag, EV:
+ TypeInformation : ClassTag](val self: Graph[K, VV, EV]) {
+
+ /**
+ * Computes the ChecksumHashCode over the Graph.
+ *
+ * @return the ChecksumHashCode over the vertices and edges.
+ */
+ @throws(classOf[Exception])
+ def checksumHashCode(): ChecksumHashCode = {
+ val checksum: ChecksumHashCode = self.getVertices.checksumHashCode()
+ checksum.add(self.getEdges checksumHashCode())
+ checksum
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/util/GraphUtilsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/util/GraphUtilsITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/util/GraphUtilsITCase.scala
new file mode 100644
index 0000000..c6d3c58
--- /dev/null
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/util/GraphUtilsITCase.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.graph.scala.test.util
+
+import org.apache.flink.api.scala._
+import org.apache.flink.graph.scala._
+import org.apache.flink.graph.scala.utils._
+import org.apache.flink.graph.scala.test.TestGraphUtils
+import org.apache.flink.test.util.MultipleProgramsTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+
+@RunWith(classOf[Parameterized])
+class GraphUtilsITCase(mode: MultipleProgramsTestBase.TestExecutionMode) extends
+MultipleProgramsTestBase(mode) {
+
+ @Test
+ @throws(classOf[Exception])
+ def testChecksumHashCodeVerticesAndEdges() {
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
+ .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
+ val checksum = graph.checksumHashCode()
+ assertEquals(checksum.getCount, 12L)
+ assertEquals(checksum.getChecksum, 19665L)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/utils/GraphUtils.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/utils/GraphUtils.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/utils/GraphUtils.java
new file mode 100644
index 0000000..e7d3a16
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/utils/GraphUtils.java
@@ -0,0 +1,38 @@
+/*
+ * 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.graph.utils;
+
+import org.apache.flink.api.java.Utils;
+import org.apache.flink.api.java.Utils.ChecksumHashCode;
+import org.apache.flink.api.java.utils.DataSetUtils;
+import org.apache.flink.graph.Graph;
+
+public class GraphUtils {
+
+ /**
+ * Computes the checksum over the Graph
+ *
+ * @return the checksum over the vertices and edges.
+ */
+ public static Utils.ChecksumHashCode checksumHashCode(Graph graph) throws Exception {
+ ChecksumHashCode checksum = DataSetUtils.checksumHashCode(graph.getVertices());
+ checksum.add(DataSetUtils.checksumHashCode(graph.getEdges()));
+ return checksum;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/util/GraphUtilsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/util/GraphUtilsITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/util/GraphUtilsITCase.java
new file mode 100644
index 0000000..51602bc
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/util/GraphUtilsITCase.java
@@ -0,0 +1,58 @@
+/*
+ * 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.graph.test.util;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.Utils.ChecksumHashCode;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.test.TestGraphUtils;
+import org.apache.flink.graph.utils.GraphUtils;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class GraphUtilsITCase extends MultipleProgramsTestBase {
+
+ public GraphUtilsITCase(TestExecutionMode mode){
+ super(mode);
+ }
+
+ @Test
+ public void testChecksumHashCodeVerticesAndEdges() throws Exception {
+ /*
+ * Test checksum hashcode
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(
+ TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env),
+ env);
+
+ ChecksumHashCode checksum = GraphUtils.checksumHashCode(graph);
+
+ assertEquals(checksum.getCount(), 12L);
+ assertEquals(checksum.getChecksum(), 19665L);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-scala/src/main/scala/org/apache/flink/api/scala/utils/package.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/utils/package.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/utils/package.scala
index 82d4394..d11cf8c 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/utils/package.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/utils/package.scala
@@ -20,7 +20,9 @@ package org.apache.flink.api.scala
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.java.Utils
+import org.apache.flink.api.java.Utils.ChecksumHashCode
import org.apache.flink.api.java.utils.{DataSetUtils => jutils}
+import org.apache.flink.util.AbstractID
import _root_.scala.language.implicitConversions
import _root_.scala.reflect.ClassTag
@@ -103,6 +105,25 @@ package object utils {
: DataSet[T] = {
wrap(jutils.sampleWithSize(self.javaSet, withReplacement, numSamples, seed))
}
+
+ // --------------------------------------------------------------------------------------------
+ // Checksum
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Convenience method to get the count (number of elements) of a DataSet
+ * as well as the checksum (sum over element hashes).
+ *
+ * @return A ChecksumHashCode with the count and checksum of elements in the data set.
+ *
+ * @see [[org.apache.flink.api.java.Utils.ChecksumHashCodeHelper]]
+ */
+ def checksumHashCode(): ChecksumHashCode = {
+ val id = new AbstractID().toString
+ self.javaSet.output(new Utils.ChecksumHashCodeHelper[T](id))
+ val res = self.javaSet.getExecutionEnvironment.execute()
+ res.getAccumulatorResult[ChecksumHashCode](id)
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-tests/src/test/java/org/apache/flink/test/util/DataSetUtilsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/util/DataSetUtilsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/util/DataSetUtilsITCase.java
index 478354a..4ccc6e2 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/util/DataSetUtilsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/util/DataSetUtilsITCase.java
@@ -23,8 +23,10 @@ import com.google.common.collect.Sets;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.DataSetUtils;
+import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -81,4 +83,15 @@ public class DataSetUtilsITCase extends MultipleProgramsTestBase {
Assert.assertEquals(expectedSize, result.size());
}
+
+ @Test
+ public void testIntegerDataSetChecksumHashCode() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Integer> ds = CollectionDataSets.getIntegerDataSet(env);
+
+ Utils.ChecksumHashCode checksum = DataSetUtils.checksumHashCode(ds);
+ Assert.assertEquals(checksum.getCount(), 15);
+ Assert.assertEquals(checksum.getChecksum(), 55);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8be52a3/flink-tests/src/test/scala/org/apache/flink/api/scala/util/DataSetUtilsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/util/DataSetUtilsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/util/DataSetUtilsITCase.scala
index 7fff8ff..25ecc9c 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/util/DataSetUtilsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/util/DataSetUtilsITCase.scala
@@ -18,6 +18,7 @@
package org.apache.flink.api.scala.util
+import org.apache.flink.api.java.Utils.ChecksumHashCode
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.utils._
import org.apache.flink.test.util.MultipleProgramsTestBase
@@ -61,4 +62,15 @@ class DataSetUtilsITCase (
Assert.assertEquals(expectedSize, result.size)
}
+
+ @Test
+ def testIntegerDataSetChecksumHashCode(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ val ds = CollectionDataSets.getIntDataSet(env)
+
+ val checksum: ChecksumHashCode = ds.checksumHashCode()
+ Assert.assertEquals(checksum.getCount, 15)
+ Assert.assertEquals(checksum.getChecksum, 55)
+ }
}