You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ik...@apache.org on 2015/07/03 01:29:02 UTC
[2/2] git commit: updated refs/heads/trunk to d7e4bde
[GIRAPH-1013] Adding prepare graph library in Java8
Summary:
Adding simple graph preparation:
- symmetric
- removal of isolated edges
- normalizing
- connected components
Creating a new module, only used for Phadoop_facebook, which is written in Java8.
Tests are new/modified from what is in our repo, the rest is identical.
Test Plan: mvn clean install
Reviewers: maja.kabiljo, dionysis.logothetis, sergey.edunov
Reviewed By: sergey.edunov
Differential Revision: https://reviews.facebook.net/D40719
Project: http://git-wip-us.apache.org/repos/asf/giraph/repo
Commit: http://git-wip-us.apache.org/repos/asf/giraph/commit/d7e4bde1
Tree: http://git-wip-us.apache.org/repos/asf/giraph/tree/d7e4bde1
Diff: http://git-wip-us.apache.org/repos/asf/giraph/diff/d7e4bde1
Branch: refs/heads/trunk
Commit: d7e4bde11fdbfdef13e9316702c914e749a7757f
Parents: f9dc6b5
Author: Igor Kabiljo <ik...@fb.com>
Authored: Thu Jun 25 14:13:18 2015 -0700
Committer: Igor Kabiljo <ik...@fb.com>
Committed: Thu Jul 2 16:04:27 2015 -0700
----------------------------------------------------------------------
checkstyle-relaxed-8.xml | 275 +++++++++++
findbugs-exclude.xml | 2 +-
giraph-block-app-8/pom.xml | 137 ++++++
.../src/main/assembly/compile.xml | 39 ++
.../block_app/library/ReusableSuppliers.java | 120 +++++
.../giraph/block_app/library/package-info.java | 21 +
.../prepare_graph/PrepareGraphPieces.java | 400 ++++++++++++++++
.../UndirectedConnectedComponents.java | 472 +++++++++++++++++++
.../WeaklyConnectedComponents.java | 163 +++++++
.../library/prepare_graph/package-info.java | 22 +
.../vertex/ConnectedComponentVertexValue.java | 56 +++
.../WeaklyConnectedComponentVertexValue.java | 57 +++
...WeaklyConnectedComponentVertexValueImpl.java | 84 ++++
.../prepare_graph/vertex/package-info.java | 21 +
.../prepare_graph/TestConnectedComponents.java | 192 ++++++++
.../library/prepare_graph/TestPrepareGraph.java | 150 ++++++
giraph-block-app/pom.xml | 4 +
.../reducers/array/HugeArrayUtils.java | 20 +-
.../block_app/test_setup/NumericTestGraph.java | 9 +-
.../test_setup/graphs/Small1GraphInit.java | 29 +-
.../test_setup/graphs/Small2GraphInit.java | 29 +-
.../graphs/SmallDirectedForestGraphInit.java | 76 +++
.../graphs/SmallDirectedTreeGraphInit.java | 75 +++
.../giraph/object/MultiSizedReusable.java | 113 +++++
.../org/apache/giraph/object/package-info.java | 21 +
.../block_app/framework/BlockExecutionTest.java | 36 +-
.../combiner/DoubleSumMessageCombiner.java | 6 +-
.../combiner/FloatSumMessageCombiner.java | 6 +-
.../apache/giraph/combiner/MessageCombiner.java | 6 +-
.../combiner/MinimumDoubleMessageCombiner.java | 3 +-
.../combiner/MinimumIntMessageCombiner.java | 2 +-
.../combiner/SimpleSumMessageCombiner.java | 5 +-
.../giraph/combiner/SumMessageCombiner.java | 73 +++
.../giraph/reducers/impl/MaxPairReducer.java | 87 ++++
.../master/TestComputationCombinerTypes.java | 2 +-
.../apache/giraph/master/TestSwitchClasses.java | 4 +-
.../giraph/vertex/TestComputationTypes.java | 4 +-
pom.xml | 28 +-
38 files changed, 2790 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/checkstyle-relaxed-8.xml
----------------------------------------------------------------------
diff --git a/checkstyle-relaxed-8.xml b/checkstyle-relaxed-8.xml
new file mode 100644
index 0000000..88a95cf
--- /dev/null
+++ b/checkstyle-relaxed-8.xml
@@ -0,0 +1,275 @@
+<?xml version="1.0"?>
+
+<!--
+ 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.
+-->
+
+<!--
+ This version of checkstyle is based on the Hadoop and common-math
+ checkstyle configurations. It is a best effort attempt to try to match
+ the CODE_CONVENTIONS and Oracle "Code Conventions for the Java
+ Programming Language". See the following link:
+
+ http://www.oracle.com/technetwork/java/codeconvtoc-136057.html
+
+ The documentation for checkstyle is available at
+
+ http://checkstyle.sourceforge.net
+-->
+
+<!DOCTYPE module PUBLIC "-//Puppy Crawl//DTD Check Configuration 1.1//EN" "http://www.puppycrawl.com/dtds/configuration_1_1.dtd">
+
+<!-- Apache giraph customization of default Checkstyle behavior -->
+<module name="Checker">
+ <property name="localeLanguage" value="en"/>
+
+ <!-- Checks for headers -->
+ <!-- See http://checkstyle.sf.net/config_header.html -->
+ <!-- Verify that EVERY source file has the appropriate license -->
+ <module name="Header">
+ <property name="headerFile" value="${checkstyle.header.file}"/>
+ <property name="fileExtensions" value="java"/>
+ </module>
+
+ <!-- Checks for Javadoc comments (checker). -->
+ <!-- See http://checkstyle.sf.net/config_javadoc.html -->
+ <!-- Require package javadoc -->
+ <module name="JavadocPackage"/>
+
+ <!-- Miscellaneous other checks (checker). -->
+ <!-- See http://checkstyle.sf.net/config_misc.html -->
+ <!-- Require files to end with newline characters -->
+ <module name="NewlineAtEndOfFile"/>
+
+ <!-- Checks for whitespace (tree walker) -->
+ <!-- See http://checkstyle.sf.net/config_whitespace.html -->
+ <!-- No tabs allowed! -->
+ <module name="FileTabCharacter"/>
+
+ <module name="TreeWalker">
+ <property name="cacheFile" value="target/checkstyle-cachefile"/>
+
+ <!-- Checks for blocks. You know, those {}'s -->
+ <!-- See http://checkstyle.sf.net/config_blocks.html -->
+ <!-- No empty blocks (i.e. catch) -->
+ <module name="EmptyBlock"/>
+ <module name="AvoidNestedBlocks"/>
+ <!-- No if/else/do/for/while without braces -->
+ <module name="NeedBraces"/>
+ <module name="LeftCurly"/>
+ <module name="RightCurly"/>
+
+ <!-- Checks for class design -->
+ <!-- See http://checkstyle.sf.net/config_design.html -->
+ <!-- Utility class should not be instantiated, they must have a
+ private constructor -->
+ <module name="HideUtilityClassConstructor"/>
+ <!-- Interfaces must be types (not just constants) -->
+ <module name="InterfaceIsType"/>
+ <!-- No public fields -->
+ <module name="VisibilityModifier">
+ <property name="protectedAllowed" value="true"/>
+ <property name="publicMemberPattern" value="^$"/>
+ </module>
+
+ <!-- Checks for common coding problems -->
+ <!-- See http://checkstyle.sf.net/config_coding.html -->
+ <module name="EmptyStatement"/>
+ <!-- Require hash code override when equals is -->
+ <module name="EqualsHashCode"/>
+ <!-- Method parameters and local variables should not hide
+ fields, except in constructors and setters -->
+ <module name="HiddenField">
+ <property name="ignoreConstructorParameter" value="true" />
+ <property name="ignoreSetter" value="true" />
+ <property name="tokens" value="VARIABLE_DEF"/>
+ </module>
+ <!-- Disallow unnecessary instantiation of Boolean, String -->
+ <module name="IllegalInstantiation">
+ <property name="classes" value="java.lang.Boolean, java.lang.String"/>
+ </module>
+ <module name="InnerAssignment"/>
+ <!-- Switch statements should be complete and with independent cases -->
+ <module name="FallThrough" />
+ <module name="MissingSwitchDefault" />
+ <module name="SimplifyBooleanExpression"/>
+ <module name="SimplifyBooleanReturn"/>
+ <!-- Only one statment per line allowed -->
+ <module name="OneStatementPerLine"/>
+ <!-- Use a consistent way to put declarations -->
+ <module name="DeclarationOrder" />
+ <!-- Don't add up parentheses when they are not required -->
+ <module name="UnnecessaryParentheses" />
+ <!-- Don't use too widespread catch (Exception, Throwable,
+ RuntimeException) -->
+ <module name="IllegalCatch" />
+ <!-- Don't use = or != for string comparisons -->
+ <module name="StringLiteralEquality" />
+ <!-- Don't declare multiple variables in the same statement -->
+ <module name="MultipleVariableDeclarations" />
+ <!-- String literals more than one character long should not be
+ repeated several times -->
+ <!-- the "unchecked" string is also accepted to allow
+ @SuppressWarnings("unchecked") -->
+ <!-- Disabling for now until we have a better ignoreStringsRegexp -->
+ <!--
+ <module name="MultipleStringLiterals" >
+ <property name="ignoreStringsRegexp" value='^(("")|(".")|("unchecked"))$'/>
+ </module>
+ -->
+
+ <!-- Checks for imports -->
+ <!-- See http://checkstyle.sf.net/config_import.html -->
+ <module name="RedundantImport"/>
+ <!-- Import should be explicit, really needed and only from pure
+ java packages -->
+ <module name="AvoidStarImport" />
+ <module name="UnusedImports" />
+ <module name="IllegalImport" />
+
+ <!-- Checks for Javadoc comments (tree walker). -->
+ <!-- See http://checkstyle.sf.net/config_javadoc.html -->
+ <!-- Javadoc must be formatted correctly -->
+ <module name="JavadocStyle">
+ <property name="checkFirstSentence" value="false"/>
+ </module>
+ <!-- Must have class / interface header comments -->
+ <module name="JavadocType"/>
+ <!-- Require method javadocs, allow undeclared RTE, allow missing
+ javadoc on getters and setters -->
+ <module name="JavadocMethod">
+ <property name="allowMissingJavadoc" value="true"/>
+ <property name="allowUndeclaredRTE" value="true"/>
+ <property name="allowMissingThrowsTags" value="true"/>
+ <property name="allowMissingPropertyJavadoc" value="true"/>
+ <property name="allowMissingParamTags" value="true"/>
+ <property name="allowMissingReturnTag" value="true"/>
+ </module>
+
+ <!-- Miscellaneous other checks (tree walker). -->
+ <!-- See http://checkstyle.sf.net/config_misc.html -->
+ <!-- Java style arrays -->
+ <module name="ArrayTypeStyle"/>
+
+ <!-- Indentation -->
+<!-- Ignore indentation before https://github.com/checkstyle/checkstyle/issues/281 is fixed
+ <module name="Indentation">
+ <property name="basicOffset" value="2"/>
+ <property name="braceAdjustment" value="0"/>
+ <property name="caseIndent" value="0"/>
+ <property name="throwsIndent" value="2"/>
+ <property name="lineWrappingIndentation" value="0"/>
+ <property name="arrayInitIndent" value="2"/>
+ </module>
+-->
+ <!-- Turn this on to see what needs to be done
+ <module name="TodoComment"/>
+ -->
+ <module name="UpperEll"/>
+
+ <!-- Modifier Checks -->
+ <!-- See http://checkstyle.sf.net/config_modifiers.html -->
+ <!-- Use a consistent way to put modifiers -->
+ <module name="ModifierOrder"/>
+ <module name="RedundantModifier"/>
+
+ <!-- Checks for Naming Conventions. -->
+ <!-- See http://checkstyle.sf.net/config_naming.html -->
+ <!-- Constant names should obey the traditional all uppercase
+ naming convention -->
+ <module name="ConstantName"/>
+ <module name="LocalFinalVariableName"/>
+ <module name="LocalVariableName"/>
+ <module name="MemberName"/>
+ <module name="MethodName"/>
+ <module name="PackageName"/>
+ <module name="ParameterName"/>
+ <module name="StaticVariableName">
+ <property name="format" value="^[A-Z0-9_]*$"/>
+ </module>
+ <module name="TypeName"/>
+
+ <!-- Checks for regexp expressions. -->
+ <!-- See http://checkstyle.sf.net/config_regexp.html -->
+ <!-- No trailing whitespace -->
+ <module name="Regexp">
+ <property name="format" value="[ \t]+$"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Trailing whitespace"/>
+ </module>
+ <!-- No System.out.println() statements -->
+ <module name="Regexp">
+ <!-- No sysouts -->
+ <property name="format" value="System\.out\.println"/>
+ <property name="illegalPattern" value="true"/>
+ </module>
+ <!-- Authors should be in pom.xml file -->
+ <module name="Regexp">
+ <property name="format" value="@author"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="developers names should be in pom file"/>
+ </module>
+
+ <!-- Checks for Size Violations. -->
+ <!-- See http://checkstyle.sf.net/config_sizes.html -->
+ <!-- Lines cannot exceed 80 chars -->
+ <module name="LineLength">
+ <property name="max" value="80"/>
+ <property name="ignorePattern" value="^import"/>
+ </module>
+ <!-- Over time, we will revised this down -->
+ <module name="MethodLength">
+ <property name="max" value="200"/>
+ </module>
+ <module name="ParameterNumber">
+ <property name="max" value="8"/>
+ </module>
+
+ <!-- Checks for whitespace (tree walker) -->
+ <!-- See http://checkstyle.sf.net/config_whitespace.html -->
+ <module name="EmptyForIteratorPad"/>
+ <!-- Spacing around methods -->
+ <module name="MethodParamPad">
+ <property name="option" value="nospace"/>
+ <property name="allowLineBreaks" value="true"/>
+ </module>
+ <!-- No whitespace before a token -->
+ <module name="NoWhitespaceBefore"/>
+ <!-- Whitespace after tokens is required -->
+ <module name="WhitespaceAfter"/>
+ <!-- Whitespace around tokens is required -->
+ <module name="WhitespaceAround"/>
+ <module name="ParenPad"/>
+ <module name="TypecastParenPad"/>
+ <!-- No extra whitespace around types -->
+ <module name="GenericWhitespace"/>
+ <!-- Operator must be at end of wrapped line -->
+ <module name="OperatorWrap">
+ <property name="option" value="eol"/>
+ </module>
+
+ <!-- Required for SuppressionCommentFilter below -->
+ <module name="FileContentsHolder"/>
+ </module>
+
+ <!-- Setup special comments to suppress specific checks from source files -->
+ <module name="SuppressionCommentFilter">
+ <property name="offCommentFormat" value="CHECKSTYLE\: stop ([\w\|]+)"/>
+ <property name="onCommentFormat" value="CHECKSTYLE\: resume ([\w\|]+)"/>
+ <property name="checkFormat" value="$1"/>
+ </module>
+</module>
+
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/findbugs-exclude.xml b/findbugs-exclude.xml
index cc1a05a..afdf041 100644
--- a/findbugs-exclude.xml
+++ b/findbugs-exclude.xml
@@ -103,6 +103,6 @@
<Match>
<!-- Java Serialization is not used, so this is never an actual issue.
On the other hand, Kryo needs lambdas to be Serializable to work. -->
- <Bug pattern="SE_BAD_FIELD,SE_COMPARATOR_SHOULD_BE_SERIALIZABLE" />
+ <Bug pattern="SE_BAD_FIELD,SE_COMPARATOR_SHOULD_BE_SERIALIZABLE,SE_NO_SERIALVERSIONID" />
</Match>
</FindBugsFilter>
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/pom.xml
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/pom.xml b/giraph-block-app-8/pom.xml
new file mode 100644
index 0000000..6645eed
--- /dev/null
+++ b/giraph-block-app-8/pom.xml
@@ -0,0 +1,137 @@
+<!--
+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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.giraph</groupId>
+ <artifactId>giraph-parent</artifactId>
+ <version>1.2.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>giraph-block-app-8</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Apache Giraph Blocks Framework in Java 8</name>
+ <url>http://giraph.apache.org/giraph-block-app/</url>
+ <description>Giraph Blocks Framework and utilities for writing applications in Java 8</description>
+
+ <properties>
+ <top.dir>${project.basedir}/..</top.dir>
+ <checkstyle.config.path>${top.dir}/checkstyle-relaxed-8.xml</checkstyle.config.path>
+ <project.build.targetJdk>1.8</project.build.targetJdk>
+ <project.build.javaHome>${env.JAVA_8_HOME}</project.build.javaHome>
+ </properties>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-javadocs</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ <configuration>
+ <additionalparam>-Xdoclint:none</additionalparam>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-site-plugin</artifactId>
+ <configuration>
+ <siteDirectory>${project.basedir}/src/site</siteDirectory>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.6</version>
+ <configuration>
+ <skip>${surefire.skip}</skip>
+ <systemProperties>
+ <property>
+ <name>prop.jarLocation</name>
+ <value>${top.dir}/giraph-core/target/giraph-${project.version}-${forHadoop}-jar-with-dependencies.jar</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+ <!-- compile dependencies. sorted lexicographically. -->
+ <dependency>
+ <groupId>it.unimi.dsi</groupId>
+ <artifactId>fastutil</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.giraph</groupId>
+ <artifactId>giraph-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.giraph</groupId>
+ <artifactId>giraph-block-app</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+
+ <!-- runtime dependency -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+
+ <!-- test dependencies. sorted lexicographically. -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/assembly/compile.xml
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/assembly/compile.xml b/giraph-block-app-8/src/main/assembly/compile.xml
new file mode 100644
index 0000000..fcaffa6
--- /dev/null
+++ b/giraph-block-app-8/src/main/assembly/compile.xml
@@ -0,0 +1,39 @@
+<!--
+ 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.
+-->
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+ <id>jar-with-dependencies</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+
+ <dependencySets>
+ <dependencySet>
+ <useProjectArtifact>true</useProjectArtifact>
+ <outputDirectory>/</outputDirectory>
+ <unpackOptions>
+ <excludes>
+ <exclude>META-INF/LICENSE</exclude>
+ </excludes>
+ </unpackOptions>
+ <unpack>true</unpack>
+ <scope>runtime</scope>
+ </dependencySet>
+ </dependencySets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/ReusableSuppliers.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/ReusableSuppliers.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/ReusableSuppliers.java
new file mode 100644
index 0000000..cadbf2f
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/ReusableSuppliers.java
@@ -0,0 +1,120 @@
+/*
+ * 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.giraph.block_app.library;
+
+import org.apache.giraph.function.primitive.Obj2DoubleFunction;
+import org.apache.giraph.function.primitive.Obj2FloatFunction;
+import org.apache.giraph.function.primitive.Obj2IntFunction;
+import org.apache.giraph.function.primitive.Obj2LongFunction;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.graph.Vertex;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.FloatWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * SupplierFromVertex that wrap other functions, providing common
+ * pattern for reusing objects, and minimizing GC overhead.
+ */
+public class ReusableSuppliers {
+ /** Hide constructor */
+ private ReusableSuppliers() { }
+
+ /**
+ * Transforms primitive long supplier into
+ * LongWritable supplier, with object being reused,
+ * to minimize GC overhead.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ SupplierFromVertex<I, V, E, LongWritable>
+ fromLong(Obj2LongFunction<Vertex<I, V, E>> supplier) {
+ LongWritable reusable = new LongWritable();
+ return (vertex) -> {
+ reusable.set(supplier.apply(vertex));
+ return reusable;
+ };
+ }
+
+ /**
+ * Transforms primitive int supplier into
+ * IntWritable supplier, with object being reused,
+ * to minimize GC overhead.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ SupplierFromVertex<I, V, E, IntWritable>
+ fromInt(Obj2IntFunction<Vertex<I, V, E>> supplier) {
+ IntWritable reusable = new IntWritable();
+ return (vertex) -> {
+ reusable.set(supplier.apply(vertex));
+ return reusable;
+ };
+ }
+
+ /**
+ * Transforms primitive float supplier into
+ * FloatWritable supplier, with object being reused,
+ * to minimize GC overhead.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ SupplierFromVertex<I, V, E, FloatWritable>
+ fromFloat(Obj2FloatFunction<Vertex<I, V, E>> supplier) {
+ FloatWritable reusable = new FloatWritable();
+ return (vertex) -> {
+ reusable.set(supplier.apply(vertex));
+ return reusable;
+ };
+ }
+
+ /**
+ * Transforms primitive double supplier into
+ * DoubleWritable supplier, with object being reused,
+ * to minimize GC overhead.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ SupplierFromVertex<I, V, E, DoubleWritable>
+ fromDouble(Obj2DoubleFunction<Vertex<I, V, E>> supplier) {
+ DoubleWritable reusable = new DoubleWritable();
+ return (vertex) -> {
+ reusable.set(supplier.apply(vertex));
+ return reusable;
+ };
+ }
+
+ /**
+ * Creates SupplierFromVertex, by passing reusable value into
+ * consumer argument, and returning same reusable every time.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable,
+ T extends Writable>
+ SupplierFromVertex<I, V, E, T> withReusable(
+ final T reusable, final ConsumerWithVertex<I, V, E, T> consumer) {
+ return (vertex) -> {
+ consumer.apply(vertex, reusable);
+ return reusable;
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/package-info.java
new file mode 100644
index 0000000..801150c
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Core library of Pieces and Suppliers, providing most common usages.
+ */
+package org.apache.giraph.block_app.library;
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/PrepareGraphPieces.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/PrepareGraphPieces.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/PrepareGraphPieces.java
new file mode 100644
index 0000000..b4d40dc
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/PrepareGraphPieces.java
@@ -0,0 +1,400 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph;
+
+import java.util.Iterator;
+
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.delegate.DelegatePiece;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.ReusableSuppliers;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.edge.EdgeFactory;
+import org.apache.giraph.edge.MutableEdge;
+import org.apache.giraph.edge.ReusableEdge;
+import org.apache.giraph.factories.MessageValueFactory;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.primitive.Int2ObjFunction;
+import org.apache.giraph.function.primitive.Obj2DoubleFunction;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.object.MultiSizedReusable;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.NoMessage;
+import org.apache.giraph.types.ops.NumericTypeOps;
+import org.apache.giraph.types.ops.PrimitiveIdTypeOps;
+import org.apache.giraph.types.ops.collections.Basic2ObjectMap;
+import org.apache.giraph.types.ops.collections.BasicSet;
+import org.apache.giraph.writable.tuple.PairWritable;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.FloatWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.log4j.Logger;
+
+/**
+ * Utility class for Pieces and Blocks that prepare graph by processing edges.
+ */
+@SuppressWarnings("rawtypes")
+public class PrepareGraphPieces {
+ private static final Logger LOG = Logger.getLogger(PrepareGraphPieces.class);
+
+ /** Hide constructor */
+ private PrepareGraphPieces() {
+ }
+
+ /**
+ * Cleans symmetric unweighted graph, by removing duplicate edges,
+ * adding opposite edges for asymetric edges, and removing standalone
+ * vertices
+ *
+ * @param idTypeOps Vertex id type ops
+ * @param <I> Vertex id type
+ * @return Block that cleans the graph
+ */
+ public static
+ <I extends WritableComparable>
+ Block cleanSymmetricUnweightedGraph(PrimitiveIdTypeOps<I> idTypeOps) {
+ return new SequenceBlock(
+ removeDuplicateEdges(idTypeOps),
+ makeSymmetricUnweighted(idTypeOps),
+ removeStandAloneVertices());
+ }
+
+ /**
+ * Creates a piece that removes edges that exist only in one direction.
+ * If edge exists in both directions, but with different edge value,
+ * both edges are left intact.
+ *
+ * @param idTypeOps Vertex id type ops
+ * @param <I> Vertex id type
+ * @return Piece that removes asymmetrical edges
+ */
+ public static
+ <I extends WritableComparable>
+ Block removeAsymEdges(PrimitiveIdTypeOps<I> idTypeOps) {
+ ConsumerWithVertex<I, Writable, Writable, Iterable<I>>
+ removeEdges = (vertex, neighbors) -> {
+ BasicSet<I> set = idTypeOps.createOpenHashSet();
+ for (I message : neighbors) {
+ set.add(message);
+ }
+
+ for (Iterator<MutableEdge<I, Writable>> iter =
+ vertex.getMutableEdges().iterator();
+ iter.hasNext();) {
+ MutableEdge<I, Writable> edge = iter.next();
+ if (!set.contains(edge.getTargetVertexId())) {
+ iter.remove();
+ }
+ }
+ };
+ return Pieces.sendMessageToNeighbors(
+ "RemoveAsymEdges",
+ idTypeOps.getTypeClass(),
+ VertexSuppliers.vertexIdSupplier(),
+ removeEdges);
+ }
+
+ /**
+ * Remove duplicate edges, for each vertex only leaving
+ * instance of first outgoing edge to each target vertex.
+ *
+ * If graph is weighted, you might want to prefer summing the
+ * weights, instead of removing later occurrences.
+ */
+ public static <I extends WritableComparable>
+ Block removeDuplicateEdges(PrimitiveIdTypeOps<I> idTypeOps) {
+ Int2ObjFunction<BasicSet<I>> reusableSets =
+ MultiSizedReusable.createForBasicSet(idTypeOps);
+ return Pieces.<I, Writable, Writable>forAllVertices(
+ "RemoveDuplicateEdges",
+ (vertex) -> {
+ BasicSet<I> set = reusableSets.apply(vertex.getNumEdges());
+ for (Iterator<MutableEdge<I, Writable>> iter =
+ vertex.getMutableEdges().iterator();
+ iter.hasNext();) {
+ MutableEdge<I, Writable> edge = iter.next();
+ if (!set.add(edge.getTargetVertexId())) {
+ iter.remove();
+ }
+ }
+ });
+ }
+
+ /**
+ * Creates a piece that will take an unweighted graph (graph with edge value
+ * being NullWritable), and make it symmetric, by creating all edges
+ * that are not present and opposite edge exists.
+ *
+ * @param idTypeOps Vertex id type ops
+ * @return Piece that makes unweighted graph symmetric
+ */
+ public static <I extends WritableComparable>
+ Block makeSymmetricUnweighted(PrimitiveIdTypeOps<I> idTypeOps) {
+ Int2ObjFunction<BasicSet<I>> reusableSets =
+ MultiSizedReusable.createForBasicSet(idTypeOps);
+ ConsumerWithVertex<I, Writable, NullWritable, Iterable<I>>
+ addEdges = (vertex, neighbors) -> {
+ BasicSet<I> set = reusableSets.apply(vertex.getNumEdges());
+
+ for (Edge<I, NullWritable> edge : vertex.getEdges()) {
+ set.add(edge.getTargetVertexId());
+ }
+ for (I neighbor : neighbors) {
+ if (!set.contains(neighbor)) {
+ Edge<I, NullWritable> edge =
+ EdgeFactory.create(idTypeOps.createCopy(neighbor));
+ vertex.addEdge(edge);
+ set.add(neighbor);
+ }
+ }
+ };
+ return Pieces.sendMessageToNeighbors(
+ "MakeSymmetricUnweighted",
+ idTypeOps.getTypeClass(),
+ VertexSuppliers.vertexIdSupplier(),
+ addEdges);
+ }
+
+ /**
+ * Make weighted graph symmetric - by making edge weight in both directions
+ * equal to the sum of weights in both directions.
+ *
+ * This means if graph is already symmetric - resulting graph will have
+ * doubled weights.
+ * We are not taking average, in order to work with integer numbers - because
+ * division by two might not be integer number.
+ */
+ public static <I extends WritableComparable, V extends Writable,
+ E extends Writable>
+ Block makeSymmetricWeighted(
+ PrimitiveIdTypeOps<I> idTypeOps, NumericTypeOps<E> edgeTypeOps) {
+ MessageValueFactory<PairWritable<I, E>> messageFactory =
+ () -> new PairWritable<>(idTypeOps.create(), edgeTypeOps.create());
+ return new Piece<I, V, E, PairWritable<I, E>, Object>() {
+ @Override
+ public VertexSender<I, V, E> getVertexSender(
+ BlockWorkerSendApi<I, V, E, PairWritable<I, E>> workerApi,
+ Object executionStage) {
+ PairWritable<I, E> message = messageFactory.newInstance();
+ return (vertex) -> {
+ idTypeOps.set(message.getLeft(), vertex.getId());
+ for (Edge<I, E> edge : vertex.getEdges()) {
+ edgeTypeOps.set(message.getRight(), edge.getValue());
+ workerApi.sendMessage(edge.getTargetVertexId(), message);
+ }
+ };
+ }
+
+ @Override
+ public
+ VertexReceiver<I, V, E, PairWritable<I, E>> getVertexReceiver(
+ BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ // TODO: After t5921368, make edges also primitive
+ ReusableEdge<I, E> reusableEdge =
+ EdgeFactory.createReusable(null, null);
+ return (vertex, messages) -> {
+ Basic2ObjectMap<I, E> map =
+ idTypeOps.create2ObjectOpenHashMap(null);
+ for (PairWritable<I, E> message : messages) {
+ if (map.containsKey(message.getLeft())) {
+ edgeTypeOps.plusInto(
+ map.get(message.getLeft()), message.getRight());
+ } else {
+ // create edge copy, since it is not stored as primitive yet
+ map.put(message.getLeft(),
+ edgeTypeOps.createCopy(message.getRight()));
+ }
+ }
+
+ for (MutableEdge<I, E> edge : vertex.getMutableEdges()) {
+ E receivedWeight = map.remove(edge.getTargetVertexId());
+ if (receivedWeight != null) {
+ edgeTypeOps.plusInto(edge.getValue(), receivedWeight);
+ edge.setValue(edge.getValue());
+ }
+ }
+
+ // TODO: add foreach, or entry iterator to Basic Maps
+ for (Iterator<I> iter = map.fastKeyIterator(); iter.hasNext();) {
+ I neighbor = iter.next();
+ reusableEdge.setTargetVertexId(neighbor);
+ reusableEdge.setValue(map.get(neighbor));
+ vertex.addEdge(reusableEdge);
+ }
+ };
+ }
+
+ @Override
+ protected MessageValueFactory<PairWritable<I, E>> getMessageFactory(
+ ImmutableClassesGiraphConfiguration conf) {
+ return messageFactory;
+ }
+
+ @Override
+ public String toString() {
+ return "MakeSymmetricWeighted";
+ }
+ };
+ }
+
+
+ /**
+ * Removing vertices from the graph takes effect in the next superstep.
+ * Adding an empty piece is to prevent calling the next send inner piece
+ * before vertices are being actually removed.
+ */
+ public static Block removeStandAloneVertices() {
+ return Pieces.removeVertices(
+ "RemoveStandaloneVertices",
+ (vertex) -> vertex.getNumEdges() == 0);
+ }
+
+ public static Block normalizeDoubleEdges() {
+ ObjectTransfer<DoubleWritable> sumEdgeWeights =
+ new ObjectTransfer<>();
+ ObjectTransfer<LongWritable> countEdges = new ObjectTransfer<>();
+
+ return new DelegatePiece<>(
+ calcSumEdgesPiece(DoubleWritable::get, sumEdgeWeights),
+ countTotalEdgesPiece(countEdges),
+ new Piece<WritableComparable, Writable, DoubleWritable, NoMessage,
+ Object>() {
+ private double averageEdgeWeight;
+
+ @Override
+ public void masterCompute(
+ BlockMasterApi master, Object executionStage) {
+ averageEdgeWeight =
+ sumEdgeWeights.get().get() / countEdges.get().get();
+ LOG.info("Averge edge weight " + averageEdgeWeight);
+ }
+
+ @Override
+ public VertexReceiver<WritableComparable, Writable, DoubleWritable,
+ NoMessage> getVertexReceiver(
+ BlockWorkerReceiveApi<WritableComparable> workerApi,
+ Object executionStage) {
+ DoubleWritable doubleWritable = new DoubleWritable();
+ return (vertex, messages) -> {
+ for (MutableEdge<WritableComparable, DoubleWritable> edge :
+ vertex.getMutableEdges()) {
+ doubleWritable.set(
+ (float) (edge.getValue().get() / averageEdgeWeight));
+ edge.setValue(doubleWritable);
+ }
+ };
+ }
+
+ @Override
+ public String toString() {
+ return "NormalizeDoubleEdges";
+ }
+ });
+ }
+
+ public static Block normalizeFloatEdges() {
+ ObjectTransfer<DoubleWritable> sumEdgeWeights =
+ new ObjectTransfer<>();
+ ObjectTransfer<LongWritable> countEdges = new ObjectTransfer<>();
+
+ return new DelegatePiece<>(
+ calcSumEdgesPiece(FloatWritable::get, sumEdgeWeights),
+ countTotalEdgesPiece(countEdges),
+ new Piece<WritableComparable, Writable, FloatWritable, NoMessage,
+ Object>() {
+ private double averageEdgeWeight;
+
+ @Override
+ public void masterCompute(
+ BlockMasterApi master, Object executionStage) {
+ averageEdgeWeight =
+ sumEdgeWeights.get().get() / countEdges.get().get();
+ LOG.info("Averge edge weight " + averageEdgeWeight);
+ }
+
+ @Override
+ public VertexReceiver<WritableComparable, Writable, FloatWritable,
+ NoMessage> getVertexReceiver(
+ BlockWorkerReceiveApi<WritableComparable> workerApi,
+ Object executionStage) {
+ FloatWritable floatWritable = new FloatWritable();
+ return (vertex, messages) -> {
+ for (MutableEdge<WritableComparable, FloatWritable> edge :
+ vertex.getMutableEdges()) {
+ floatWritable.set(
+ (float) (edge.getValue().get() / averageEdgeWeight));
+ edge.setValue(floatWritable);
+ }
+ };
+ }
+
+ @Override
+ public String toString() {
+ return "NormalizeFloatEdges";
+ }
+ });
+ }
+
+ /**
+ * Piece that calculates total number of edges,
+ * and gives result to the {@code countEdges} consumer.
+ */
+ public static
+ Piece<WritableComparable, Writable, Writable, NoMessage, Object>
+ countTotalEdgesPiece(Consumer<LongWritable> countEdges) {
+ return Pieces.reduce(
+ "CountTotalEdgesPiece",
+ SumReduce.LONG,
+ ReusableSuppliers.fromLong((vertex) -> vertex.getNumEdges()),
+ countEdges);
+ }
+
+ /**
+ * Piece that calculates total edge weight in the graph,
+ * and gives result to the {@code sumEdgeWeights} consumer.
+ */
+ public static <E extends Writable>
+ Piece<WritableComparable, Writable, E, NoMessage, Object> calcSumEdgesPiece(
+ Obj2DoubleFunction<E> edgeValueF,
+ Consumer<DoubleWritable> sumEdgeWeights) {
+ return Pieces.reduce(
+ "CalcSumEdgesPiece",
+ SumReduce.DOUBLE,
+ ReusableSuppliers.fromDouble((vertex) -> {
+ double sum = 0;
+ for (Edge<WritableComparable, E> edge : vertex.getEdges()) {
+ sum += edgeValueF.apply(edge.getValue());
+ }
+ return sum;
+ }),
+ sumEdgeWeights);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/UndirectedConnectedComponents.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/UndirectedConnectedComponents.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/UndirectedConnectedComponents.java
new file mode 100644
index 0000000..2610436
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/UndirectedConnectedComponents.java
@@ -0,0 +1,472 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.CreateReducersApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.SendMessageChain;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.block_app.reducers.map.BasicMapReduce;
+import org.apache.giraph.combiner.MessageCombiner;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.Supplier;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.reducers.impl.MaxPairReducer;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.NoMessage;
+import org.apache.giraph.types.ops.LongTypeOps;
+import org.apache.giraph.types.ops.TypeOps;
+import org.apache.giraph.writable.tuple.LongLongWritable;
+import org.apache.giraph.writable.tuple.PairWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Iterators;
+
+/**
+ * Class for creating utility blocks for calculating and processing
+ * connected components.
+ *
+ * Graph is expected to be symmetric before calling any of the methods here.
+ */
+public class UndirectedConnectedComponents {
+ private static final Logger LOG =
+ Logger.getLogger(UndirectedConnectedComponents.class);
+
+ private UndirectedConnectedComponents() { }
+
+ /** Initialize vertex values for connected components calculation */
+ private static <I extends WritableComparable, V extends Writable>
+ Piece<I, V, Writable, NoMessage, Object> createInitializePiece(
+ TypeOps<I> idTypeOps,
+ Consumer<Boolean> vertexUpdatedComponent,
+ ConsumerWithVertex<I, V, Writable, I> setComponent,
+ SupplierFromVertex<I, V, Writable,
+ ? extends Iterable<? extends Edge<I, ?>>> edgeSupplier) {
+ I result = idTypeOps.create();
+ return Pieces.forAllVerticesOnReceive("InitializeCC", (vertex) -> {
+ idTypeOps.set(result, vertex.getId());
+ boolean updated = false;
+ for (Edge<I, ?> edge : edgeSupplier.get(vertex)) {
+ if (result.compareTo(edge.getTargetVertexId()) > 0) {
+ idTypeOps.set(result, edge.getTargetVertexId());
+ updated = true;
+ }
+ }
+ setComponent.apply(vertex, result);
+ vertexUpdatedComponent.apply(updated);
+ });
+ }
+
+ /** Propagate connected components to neighbor pieces */
+ private static class PropagateConnectedComponentsPiece
+ <I extends WritableComparable, V extends Writable>
+ extends Piece<I, V, Writable, I, Object> {
+ private final TypeOps<I> idTypeOps;
+ private final Supplier<Boolean> vertexToPropagate;
+ private final Consumer<Boolean> vertexUpdatedComponent;
+ private final Consumer<Boolean> converged;
+ private final SupplierFromVertex<I, V, Writable, I> getComponent;
+ private final ConsumerWithVertex<I, V, Writable, I> setComponent;
+ private final SupplierFromVertex<I, V, Writable,
+ ? extends Iterable<? extends Edge<I, ?>>> edgeSupplier;
+
+ private ReducerHandle<LongWritable, LongWritable> propagatedAggregator;
+
+ PropagateConnectedComponentsPiece(
+ TypeOps<I> idTypeOps,
+ Supplier<Boolean> vertexToPropagate,
+ Consumer<Boolean> vertexUpdatedComponent,
+ Consumer<Boolean> converged,
+ SupplierFromVertex<I, V, Writable, I> getComponent,
+ ConsumerWithVertex<I, V, Writable, I> setComponent,
+ SupplierFromVertex<I, V, Writable,
+ ? extends Iterable<? extends Edge<I, ?>>> edgeSupplier) {
+ this.idTypeOps = idTypeOps;
+ this.vertexToPropagate = vertexToPropagate;
+ this.vertexUpdatedComponent = vertexUpdatedComponent;
+ this.converged = converged;
+ this.getComponent = getComponent;
+ this.setComponent = setComponent;
+ this.edgeSupplier = edgeSupplier;
+ }
+
+ @Override
+ public void registerReducers(
+ CreateReducersApi reduceApi, Object executionStage) {
+ propagatedAggregator = reduceApi.createLocalReducer(SumReduce.LONG);
+ }
+
+ @Override
+ public VertexSender<I, V, Writable> getVertexSender(
+ final BlockWorkerSendApi<I, V, Writable, I> workerApi,
+ Object executionStage) {
+ final LongWritable one = new LongWritable(1);
+ return new InnerVertexSender() {
+ @Override
+ public void vertexSend(Vertex<I, V, Writable> vertex) {
+ if (vertexToPropagate.get()) {
+ workerApi.sendMessageToMultipleEdges(
+ Iterators.transform(
+ edgeSupplier.get(vertex).iterator(),
+ new Function<Edge<I, ?>, I>() {
+ @Override
+ public I apply(Edge<I, ?> edge) {
+ return edge.getTargetVertexId();
+ }
+ }),
+ getComponent.get(vertex));
+ propagatedAggregator.reduce(one);
+ }
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ converged.apply(propagatedAggregator.getReducedValue(master).get() == 0);
+ LOG.info("Undirected CC: " +
+ propagatedAggregator.getReducedValue(master).get() +
+ " many vertices sent in this iteration");
+ }
+
+ @Override
+ public VertexReceiver<I, V, Writable, I> getVertexReceiver(
+ BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ return new InnerVertexReceiver() {
+ private final I received = idTypeOps.create();
+
+ @Override
+ public void vertexReceive(Vertex<I, V, Writable> vertex,
+ Iterable<I> messages) {
+ boolean first = true;
+ for (I value : messages) {
+ if (first) {
+ idTypeOps.set(received, value);
+ first = false;
+ } else {
+ if (received.compareTo(value) > 0) {
+ idTypeOps.set(received, value);
+ }
+ }
+ }
+
+ I cur = getComponent.get(vertex);
+ if (!first && cur.compareTo(received) > 0) {
+ setComponent.apply(vertex, received);
+ vertexUpdatedComponent.apply(true);
+ } else {
+ vertexUpdatedComponent.apply(false);
+ }
+ }
+ };
+ }
+
+ @Override
+ public Class<I> getMessageClass() {
+ return idTypeOps.getTypeClass();
+ }
+ }
+
+ /** Calculates number of components, and the number of active vertices */
+ public static final class CalculateNumberOfComponents<V extends Writable>
+ extends Piece<LongWritable, V, Writable, LongWritable, Object> {
+ private final Consumer<LongWritable> numActiveConsumer;
+ private final Consumer<LongWritable> numComponentsConsumer;
+ private final
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent;
+
+ private ReducerHandle<LongWritable, LongWritable> numComponentsAggregator;
+ private ReducerHandle<LongWritable, LongWritable> numActiveAggregator;
+
+ public CalculateNumberOfComponents(
+ Consumer<LongWritable> numActiveConsumer,
+ Consumer<LongWritable> numComponentsConsumer,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable>
+ getComponent) {
+ this.numActiveConsumer = numActiveConsumer;
+ this.numComponentsConsumer = numComponentsConsumer;
+ this.getComponent = getComponent;
+ }
+
+ @Override
+ public void registerReducers(
+ CreateReducersApi reduceApi, Object executionStage) {
+ numComponentsAggregator = reduceApi.createLocalReducer(SumReduce.LONG);
+ numActiveAggregator = reduceApi.createLocalReducer(SumReduce.LONG);
+ }
+
+ @Override
+ public VertexSender<LongWritable, V, Writable> getVertexSender(
+ BlockWorkerSendApi<LongWritable, V, Writable, LongWritable> workerApi,
+ Object executionStage) {
+ final LongWritable one = new LongWritable(1);
+ return new InnerVertexSender() {
+ @Override
+ public void vertexSend(Vertex<LongWritable, V, Writable> vertex) {
+ numActiveAggregator.reduce(one);
+ // Only aggregate if you are the minimum of your CC
+ if (vertex.getId().get() == getComponent.get(vertex).get()) {
+ numComponentsAggregator.reduce(one);
+ }
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ numActiveConsumer.apply(numActiveAggregator.getReducedValue(master));
+ numComponentsConsumer.apply(
+ numComponentsAggregator.getReducedValue(master));
+ LOG.info("Num active is : " +
+ numActiveAggregator.getReducedValue(master));
+ LOG.info("Num components is : " +
+ numComponentsAggregator.getReducedValue(master));
+ }
+
+ @Override
+ protected
+ MessageCombiner<? super LongWritable, LongWritable> getMessageCombiner(
+ ImmutableClassesGiraphConfiguration conf) {
+ return SumMessageCombiner.LONG;
+ }
+ }
+
+ /**
+ * Calculate connected components, doing as many iterations as needed,
+ * but no more than maxIterations.
+ *
+ * Graph is expected to be symmetric.
+ */
+ public static <I extends WritableComparable, V extends Writable>
+ Block calculateConnectedComponents(
+ int maxIterations,
+ TypeOps<I> idTypeOps,
+ final SupplierFromVertex<I, V, Writable, I> getComponent,
+ final ConsumerWithVertex<I, V, Writable, I> setComponent,
+ SupplierFromVertex<I, V, Writable,
+ ? extends Iterable<? extends Edge<I, ?>>> edgeSupplier) {
+ ObjectTransfer<Boolean> converged = new ObjectTransfer<>();
+ ObjectTransfer<Boolean> vertexUpdatedComponent = new ObjectTransfer<>();
+
+ return new SequenceBlock(
+ createInitializePiece(
+ idTypeOps,
+ vertexUpdatedComponent,
+ setComponent,
+ edgeSupplier),
+ new RepeatUntilBlock(
+ maxIterations,
+ new PropagateConnectedComponentsPiece<>(
+ idTypeOps,
+ vertexUpdatedComponent,
+ vertexUpdatedComponent,
+ converged, getComponent, setComponent,
+ edgeSupplier
+ ),
+ converged
+ )
+ );
+ }
+
+ /**
+ * Default block, which calculates connected components using the
+ * vertex's default edges.
+ */
+ public static <I extends WritableComparable, V extends Writable>
+ Block calculateConnectedComponents(
+ int maxIterations,
+ TypeOps<I> idTypeOps,
+ final SupplierFromVertex<I, V, Writable, I> getComponent,
+ final ConsumerWithVertex<I, V, Writable, I> setComponent) {
+ return calculateConnectedComponents(
+ maxIterations,
+ idTypeOps,
+ getComponent,
+ setComponent,
+ VertexSuppliers.vertexEdgesSupplier());
+ }
+
+ public static <V extends Writable>
+ Block calculateConnectedComponents(
+ int maxIterations,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> setComponent
+ ) {
+ return calculateConnectedComponents(
+ maxIterations,
+ LongTypeOps.INSTANCE,
+ getComponent,
+ setComponent,
+ VertexSuppliers.vertexEdgesSupplier());
+ }
+
+ /**
+ * Calculates sizes of all components by aggregating on master, and allows
+ * each vertex to consume its size. Differs from CalculateComponentSizesPiece
+ * in that aggregation happens on master, instead of message sends to the
+ * component_id.
+ */
+ public static <V extends Writable>
+ Block calculateConnectedComponentSizes(
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> sizeConsumer
+ ) {
+ Pair<LongWritable, LongWritable> componentToReducePair = Pair.of(
+ new LongWritable(), new LongWritable(1));
+ LongWritable reusableLong = new LongWritable();
+ return Pieces.reduceAndBroadcast(
+ "CalcConnectedComponentSizes",
+ new BasicMapReduce<>(
+ LongTypeOps.INSTANCE, LongTypeOps.INSTANCE, SumReduce.LONG),
+ (Vertex<LongWritable, V, Writable> vertex) -> {
+ componentToReducePair.getLeft().set(getComponent.get(vertex).get());
+ return componentToReducePair;
+ },
+ (vertex, componentSizes) -> {
+ long compSize = componentSizes.get(getComponent.get(vertex)).get();
+ reusableLong.set(compSize);
+ sizeConsumer.apply(vertex, reusableLong);
+ });
+ }
+
+ /**
+ * Given a graph with already calculated connected components - calculates
+ * ID of the largest one.
+ */
+ public static <V extends Writable>
+ Block calculateLargestConnectedComponentStats(
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ Consumer<PairWritable<LongWritable, LongWritable>>
+ largestComponentConsumer) {
+ LongWritable one = new LongWritable(1);
+ LongLongWritable pair = new LongLongWritable();
+ return SendMessageChain.<LongWritable, V, Writable, LongWritable>startSend(
+ "CalcComponentSizesPiece",
+ SumMessageCombiner.LONG,
+ (vertex) -> one,
+ (vertex) -> Iterators.singletonIterator(getComponent.get(vertex))
+ ).endReduce(
+ "CalcLargestComponent",
+ new MaxPairReducer<>(LongTypeOps.INSTANCE, LongTypeOps.INSTANCE),
+ (vertex, message) -> {
+ long curSum = message != null ? message.get() : 0;
+ pair.getLeft().set(getComponent.get(vertex).get());
+ pair.getRight().set(curSum);
+ return pair;
+ },
+ largestComponentConsumer);
+ }
+
+ /**
+ * Given a graph with already calculated connected components - calculates
+ * ID of the largest one.
+ */
+ public static <V extends Writable> Block calculateLargestConnectedComponent(
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ LongWritable largestComponent) {
+ return calculateLargestConnectedComponentStats(
+ getComponent,
+ (t) -> largestComponent.set(t.getLeft().get())
+ );
+ }
+
+ /**
+ * Given a graph with already calculated connected components - calculates
+ * size of the largest one.
+ */
+ public static <V extends Writable>
+ Block calculateLargestConnectedComponentSize(
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ LongWritable largestComponentSize) {
+ return calculateLargestConnectedComponentStats(
+ getComponent,
+ (t) -> largestComponentSize.set(t.getRight().get()));
+ }
+
+ /**
+ * Takes symmetric graph, and removes all edges/vertices that
+ * are not in largest connected component.
+ */
+ public static <V extends Writable> Block calculateAndKeepLargestComponent(
+ int maxIterations,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> setComponent
+ ) {
+ final LongWritable largestComponent = new LongWritable();
+ return new SequenceBlock(
+ calculateConnectedComponents(
+ maxIterations, LongTypeOps.INSTANCE, getComponent, setComponent,
+ VertexSuppliers.vertexEdgesSupplier()),
+ calculateLargestConnectedComponent(getComponent, largestComponent),
+ Pieces.<LongWritable, V, Writable>removeVertices(
+ "KeepOnlyLargestComponent",
+ (vertex) -> !largestComponent.equals(getComponent.get(vertex))));
+ }
+
+
+ /**
+ * Takes symmetric graph, and removes all edges/vertices that
+ * are belong to connected components smaller than specified
+ * threshold
+ */
+ public static <V extends Writable>
+ Block calculateAndKeepComponentAboveThreshold(
+ int maxIterations,
+ int threshold,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> setComponent
+ ) {
+ final ObjectTransfer<Boolean> belowThreshold = new ObjectTransfer<>();
+ return new SequenceBlock(
+ UndirectedConnectedComponents.calculateConnectedComponents(
+ maxIterations,
+ LongTypeOps.INSTANCE,
+ getComponent,
+ setComponent
+ ),
+ UndirectedConnectedComponents.calculateConnectedComponentSizes(
+ getComponent,
+ (vertex, value) -> {
+ belowThreshold.apply(value.get() < threshold);
+ }),
+ Pieces.removeVertices(
+ "KeepAboveTresholdComponents",
+ belowThreshold.castToSupplier()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/WeaklyConnectedComponents.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/WeaklyConnectedComponents.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/WeaklyConnectedComponents.java
new file mode 100644
index 0000000..9070f75
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/WeaklyConnectedComponents.java
@@ -0,0 +1,163 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph;
+
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
+
+import java.util.Iterator;
+
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.edge.EdgeFactory;
+import org.apache.giraph.edge.MutableEdge;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.types.ops.LongTypeOps;
+import org.apache.hadoop.io.FloatWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Class for computing the weakly connected components of a directed graph.
+ *
+ * If the graph is undirected, this has the exact same behavior as
+ * UndirectedConnectedComponents.
+ *
+ * This currently does not support weighted directed graphs.
+ */
+public class WeaklyConnectedComponents {
+ private WeaklyConnectedComponents() { }
+
+ /** Save existing edges */
+ private static <V extends Writable> Block saveDirectedEdges(
+ ConsumerWithVertex<LongWritable, V, Writable, LongOpenHashSet> setEdges) {
+ return Pieces.<LongWritable, V, Writable>forAllVertices(
+ "SaveDirectedEdgesPiece",
+ (vertex) -> {
+ LongOpenHashSet friendids = new LongOpenHashSet();
+ for (Edge<LongWritable, Writable> edge : vertex.getEdges()) {
+ friendids.add(edge.getTargetVertexId().get());
+ }
+ setEdges.apply(vertex, friendids);
+ });
+ }
+
+ /** Restore existing edges */
+ private static <V extends Writable> Block restoreDirectedEdges(
+ SupplierFromVertex<LongWritable, V, Writable, LongOpenHashSet> getEdges) {
+ return Pieces.<LongWritable, V, Writable>forAllVertices(
+ "RestoreDirectedEdgesPiece",
+ (vertex) -> {
+ LongOpenHashSet currentList = getEdges.get(vertex);
+ Iterator<MutableEdge<LongWritable, Writable>> iter =
+ vertex.getMutableEdges().iterator();
+ while (iter.hasNext()) {
+ MutableEdge<LongWritable, Writable> edge = iter.next();
+ if (!currentList.contains(edge.getTargetVertexId().get())) {
+ iter.remove();
+ }
+ }
+ });
+ }
+
+ /**
+ * Calculate connected components, doing as many iterations as needed,
+ * but no more than maxIterations.
+ *
+ * Graph is expected to be symmetric.
+ */
+ public static <V extends Writable> Block calculateConnectedComponents(
+ int maxIterations,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> setComponent,
+ SupplierFromVertex<LongWritable, V, Writable, LongOpenHashSet> getEdges,
+ ConsumerWithVertex<LongWritable, V, Writable, LongOpenHashSet> setEdges,
+ boolean useFloatWeights) {
+ return new SequenceBlock(
+ saveDirectedEdges(setEdges),
+ makeSymmetric(useFloatWeights),
+ UndirectedConnectedComponents.calculateConnectedComponents(
+ maxIterations, LongTypeOps.INSTANCE, getComponent, setComponent),
+ restoreDirectedEdges(getEdges));
+ }
+
+ /**
+ * Takes unweighted directed graph, and removes all edges/vertices that
+ * are not in largest weakly connected component.
+ */
+ public static <V extends Writable> Block calculateAndKeepLargestComponent(
+ int maxIterations,
+ SupplierFromVertex<LongWritable, V, Writable, LongWritable> getComponent,
+ ConsumerWithVertex<LongWritable, V, Writable, LongWritable> setComponent,
+ SupplierFromVertex<LongWritable, V, Writable, LongOpenHashSet> getEdges,
+ ConsumerWithVertex<LongWritable, V, Writable, LongOpenHashSet> setEdges,
+ boolean useFloatWeights) {
+ return new SequenceBlock(
+ saveDirectedEdges(setEdges),
+ makeSymmetric(useFloatWeights),
+ UndirectedConnectedComponents.calculateAndKeepLargestComponent(
+ maxIterations, getComponent, setComponent),
+ restoreDirectedEdges(getEdges));
+ }
+
+ private static Block makeSymmetric(boolean useFloatWeights) {
+ if (!useFloatWeights) {
+ return PrepareGraphPieces.makeSymmetricUnweighted(LongTypeOps.INSTANCE);
+ } else {
+ return makeSymmetricFloatWeighted();
+ }
+ }
+
+ /**
+ * This is just used internally in weakly connected components if the graph
+ * has float weights. This works fine here because we don't actually ever
+ * use the weights, but should not be used outside of the
+ * WeaklyConnectedComponents class and hence is private.
+ */
+ private static <V extends Writable>
+ Piece<LongWritable, V, FloatWritable, LongWritable, Object>
+ makeSymmetricFloatWeighted() {
+ LongSet set = new LongOpenHashSet();
+ FloatWritable floatWritable = new FloatWritable(1.0f);
+ ConsumerWithVertex<LongWritable, V, FloatWritable, Iterable<LongWritable>>
+ addEdges = (vertex, neighbors) -> {
+ set.clear();
+ for (Edge<LongWritable, FloatWritable> edge : vertex.getEdges()) {
+ set.add(edge.getTargetVertexId().get());
+ }
+ for (LongWritable message : neighbors) {
+ if (!set.contains(message.get())) {
+ Edge<LongWritable, FloatWritable> edge = EdgeFactory.create(
+ new LongWritable(message.get()), floatWritable);
+ vertex.addEdge(edge);
+ set.add(message.get());
+ }
+ }
+ };
+ return Pieces.sendMessageToNeighbors(
+ "MakeSymmetricFloatWeighted",
+ LongWritable.class,
+ VertexSuppliers.<LongWritable, V, FloatWritable>vertexIdSupplier(),
+ addEdges);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/package-info.java
new file mode 100644
index 0000000..d3c263a
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Vertex Value interfaces and implementations
+ * for preparing and preprocessing the graph.
+ */
+package org.apache.giraph.block_app.library.prepare_graph;
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/ConnectedComponentVertexValue.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/ConnectedComponentVertexValue.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/ConnectedComponentVertexValue.java
new file mode 100644
index 0000000..85d3a2c
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/ConnectedComponentVertexValue.java
@@ -0,0 +1,56 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph.vertex;
+
+import org.apache.giraph.block_app.library.ReusableSuppliers;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Vertex value interface for connected component extraction
+ */
+public interface ConnectedComponentVertexValue extends Writable {
+ long getComponent();
+ void setComponent(long component);
+
+ /**
+ * If your vertex value class implements ConnectedComponentVertexValue,
+ * you can use this consumer for the UndirectedConnectedComponent Blocks.
+ */
+ static
+ <I extends WritableComparable, V extends ConnectedComponentVertexValue,
+ E extends Writable>
+ ConsumerWithVertex<I, V, E, LongWritable> setComponentConsumer() {
+ return (vertex, value) -> vertex.getValue().setComponent(value.get());
+ }
+
+ /**
+ * If your vertex value class implements ConnectedComponentVertexValue, you
+ * can use this supplier for the UndirectedConnectedComponent Blocks.
+ */
+ static
+ <I extends WritableComparable, V extends ConnectedComponentVertexValue,
+ E extends Writable>
+ SupplierFromVertex<I, V, E, LongWritable> getComponentSupplier() {
+ return ReusableSuppliers.fromLong(
+ (vertex) -> vertex.getValue().getComponent());
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValue.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValue.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValue.java
new file mode 100644
index 0000000..94f6836
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValue.java
@@ -0,0 +1,57 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph.vertex;
+
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+
+/**
+ * Vertex value interface for weakly connected component extraction
+ */
+public interface WeaklyConnectedComponentVertexValue
+ extends ConnectedComponentVertexValue {
+ void setEdgeIds(LongOpenHashSet edgeIds);
+ LongOpenHashSet takeEdgeIds();
+
+ /**
+ * If your vertex value class implements WeaklyConnectedComponentVertexValue,
+ * you can use these 4 suppliers and consumers to interact with this Piece.
+ */
+ static
+ <I extends WritableComparable, V extends WeaklyConnectedComponentVertexValue,
+ E extends Writable>
+ ConsumerWithVertex<I, V, E, LongOpenHashSet> setEdgeIdsConsumer() {
+ return (vertex, edgeIds) -> vertex.getValue().setEdgeIds(edgeIds);
+ }
+
+ /**
+ * If your vertex value class implements WeaklyConnectedComponentVertexValue,
+ * you can use these 4 suppliers and consumers to interact with this Piece.
+ */
+ static
+ <I extends WritableComparable, V extends WeaklyConnectedComponentVertexValue,
+ E extends Writable>
+ SupplierFromVertex<I, V, E, LongOpenHashSet> getEdgeIdsSupplier() {
+ return (vertex) -> vertex.getValue().takeEdgeIds();
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValueImpl.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValueImpl.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValueImpl.java
new file mode 100644
index 0000000..8b1da12
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/WeaklyConnectedComponentVertexValueImpl.java
@@ -0,0 +1,84 @@
+/*
+ * 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.giraph.block_app.library.prepare_graph.vertex;
+
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Implementation of WeaklyConnectedComponentVertexValue
+ */
+public class WeaklyConnectedComponentVertexValueImpl
+ implements WeaklyConnectedComponentVertexValue {
+ private long component;
+ private LongOpenHashSet edgeIds = null;
+
+ @Override
+ public long getComponent() {
+ return component;
+ }
+
+ @Override
+ public void setComponent(long component) {
+ this.component = component;
+ }
+
+ @Override
+ public void setEdgeIds(LongOpenHashSet edgeIds) {
+ this.edgeIds = edgeIds;
+ }
+
+ @Override
+ public LongOpenHashSet takeEdgeIds() {
+ LongOpenHashSet res = edgeIds;
+ edgeIds = null;
+ return res;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeLong(component);
+ if (edgeIds == null) {
+ out.writeBoolean(false);
+ } else {
+ out.writeBoolean(true);
+ out.writeInt(edgeIds.size());
+ for (long longId: edgeIds) {
+ out.writeLong(longId);
+ }
+ }
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ component = in.readLong();
+ boolean isEdgeIds = in.readBoolean();
+ if (isEdgeIds) {
+ int friendsSize = in.readInt();
+ edgeIds = new LongOpenHashSet(friendsSize);
+ for (int i = 0; i < friendsSize; i++) {
+ edgeIds.add(in.readLong());
+ }
+ } else {
+ edgeIds = null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/d7e4bde1/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/package-info.java
new file mode 100644
index 0000000..6987fa4
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/prepare_graph/vertex/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Utilities for preparing and preprocessing the graph.
+ */
+package org.apache.giraph.block_app.library.prepare_graph.vertex;