Make gradoop demo work with gradoop-compatible datasets from HDFS ...

pom.xml

@@ -19,7 +19,7 @@
     <properties>
         <dep.flink.version>1.3.1</dep.flink.version>
         <dep.gradoop.version>0.3.2</dep.gradoop.version>
-        <dep.jersey.version>1.19.3</dep.jersey.version>
+        <dep.jersey.version>1.19</dep.jersey.version>
         <dep.gson.version>2.8.0</dep.gson.version>
         <dep.mongodb.version>3.3.0</dep.mongodb.version>
     </properties>
@@ -34,6 +34,33 @@
                     <target>1.8</target>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>2.3</version>
+                <executions>
+                    <!-- Run shade goal on package phase -->
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <finalName>gradoop-demo-shaded</finalName>
+                            <transformers>
+                                <!-- add Main-Class to manifest file -->
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass>org.gradoop.demo.server.Server</mainClass>
+                                </transformer>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>reference.conf</resource>
+                                </transformer>
+                            </transformers>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 
@@ -43,19 +70,46 @@
             <groupId>org.gradoop</groupId>
             <artifactId>gradoop-flink</artifactId>
             <version>${dep.gradoop.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId> <!-- because this brings 2.5.1 -->
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-hadoop-compatibility_2.11</artifactId> <!-- because this brings 2.5.1 via flink-shaded-hadoop2-->
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <!-- Flink -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-java</artifactId>
             <version>${dep.flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId> <!-- because this brings 2.5.1 via flink-shaded-hadoop2-->
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop2</artifactId> <!-- because this brings 2.5.1 via flink-shaded-hadoop2-->
+                </exclusion>
+            </exclusions>
+
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-clients_2.11</artifactId>
             <version>${dep.flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop2</artifactId> <!-- because this brings 2.5.1 via flink-shaded-hadoop2-->
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <!-- Jersey -->
@@ -77,5 +131,39 @@
             <version>${dep.jersey.version}</version>
         </dependency>
 
+        <!-- Explicit dependencies -->
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>2.8.3</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>2.8.3</version>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>2.8.3</version>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>2.8.3</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs-client</artifactId>
+            <version>2.8.3</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
-</project>
+</project>

src/main/java/org/gradoop/demo/server/Constants.java

@@ -0,0 +1,18 @@
+package org.gradoop.demo.server;
+
+import java.util.Set;
+
+import static com.google.common.collect.Sets.newHashSet;
+import static java.util.Collections.unmodifiableSet;
+
+public final class Constants {
+    static final Set<String> BUNDLED_DATABASE_NAMES =
+        unmodifiableSet(newHashSet("Pokec_Sample", "Graphalytics_SF1_Sample", "Example"));
+
+    static final Set<String> GRADOOP_FILE_NAMES =
+        unmodifiableSet(newHashSet("metadata.csv", "vertices.csv", "edges.csv"));
+
+    static final String DATASOURCES_NAME_KEY = "datasourceNames";
+    private Constants() {
+    }
+}

src/main/java/org/gradoop/demo/server/FetchStatus.java

@@ -0,0 +1,32 @@
+package org.gradoop.demo.server;
+
+import java.util.Objects;
+
+import static java.util.Objects.requireNonNull;
+
+final class FetchStatus {
+    enum Status {
+        FETCH_ERROR,
+        FETCHED_FROM_HDFS,
+        PRESENT_LOCALLY
+    }
+    private final String name;
+    private final Status status;
+
+    FetchStatus(String name, Status status) {
+        requireNonNull(name);
+        requireNonNull(status);
+        this.name = name;
+        this.status = status;
+    }
+    String getName() {
+        return name;
+    }
+    Status getStatus() {
+        return status;
+    }
+    @Override
+    public String toString() {
+        return "[name: " + name + ", status: " + status + "]";
+    }
+}

src/main/java/org/gradoop/demo/server/GradoopGraphsetStore.java

@@ -0,0 +1,49 @@
+package org.gradoop.demo.server;
+
+import org.gradoop.flink.io.api.DataSource;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * <p>Represents a Store for Graphsets.</p>
+ */
+interface GradoopGraphsetStore {
+    /**
+     * Returns a {@linkplain List} of names of {@linkplain DataSource}s that are available for exploration. A DataSource
+     * could be a set of files, for instance, if the DataSource is a {@linkplain org.gradoop.flink.io.impl.csv.CSVDataSource},
+     * then it is a folder with three files: <code>vertices.csv, edges.csv, and metadata.csv</code>.
+     */
+    Set<String> getDataSourceNames();
+
+    /**
+     * Returns the absolute path of the files that represent this data source. Remember that every data source
+     * is a collection of three files: vertices.csv, edges.csv, and metadata.csv.
+     * @param dataSourceName
+     * @return a String representing the absolute path of the datasource, null otherwise
+     */
+    String getPath(String dataSourceName);
+
+    /**
+     * <p> Updates the Gradoop Graphsets from remote (HDFS, e.g.) location. It returns a list of strings
+     * indicating the status of the refresh operation. A graphset that is available locally is fetched again
+     * when this operation is done.
+     * </p>
+     * <p>
+     *     Details:
+     *     <ul>
+     *        <li>Get the set of locally available graphsets (A)</li>
+     *        <li>Get the set of graphsets available in HDFS (B)</li>
+     *        <li>For each graphset b in B, remove it from A, (re) fetch it from HDFS, set its status as
+     *        {@linkplain FetchStatus.Status#FETCHED_FROM_HDFS}, add the status to the return set (C)</li>
+     *        <li>For each graphset a in A, set a status {@linkplain FetchStatus.Status#PRESENT_LOCALLY}</li>
+     *        <li>Return C</li>
+     *     </ul>
+     * </p>
+     */
+    Set<FetchStatus> refresh() throws IOException;
+
+    /** Decides whether this store is "local-only" */
+    boolean isLocal();
+}

src/main/java/org/gradoop/demo/server/HdfsGradoopGraphsetStore.java

@@ -0,0 +1,123 @@
+package org.gradoop.demo.server;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.Set;
+
+import static java.util.Objects.requireNonNull;
+import static org.gradoop.demo.server.Constants.GRADOOP_FILE_NAMES;
+
+/**
+ * <p>Represents a store of graphsets that are available to explore using Gradoop and that are stored in HDFS.</p>
+ */
+public class HdfsGradoopGraphsetStore extends Configured {
+
+    public static final String DEFAULT_BASE_PATH_IN_HDFS = "/app/ugraph/gradoop-graphsets/";
+    private static final Logger log = LoggerFactory.getLogger(HdfsGradoopGraphsetStore.class);
+
+    private final String basePath;
+    private final Configuration config;
+
+    public HdfsGradoopGraphsetStore() {
+        this(new Configuration(), DEFAULT_BASE_PATH_IN_HDFS);
+    }
+
+    public HdfsGradoopGraphsetStore(Configuration config, String basePath) {
+        requireNonNull(config);
+        requireNonNull(basePath);
+        this.basePath = basePath;
+        this.config = config;
+        this.config.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
+        this.config.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
+    }
+
+    public Set<String> getDataSourceNames() throws IOException {
+        Path basePath = new Path(this.basePath);
+        FileSystem fs = basePath.getFileSystem(config);
+//        if (!fs.getFileStatus(basePath).isDirectory()) {
+//            throw new RuntimeException("path is not a directory: " + basePath);
+//        }
+        Set<String> dataSources = new HashSet<>(16); // expected to be fewer than this
+        try {
+            RemoteIterator<LocatedFileStatus> iter = fs.listFiles(basePath, true);
+            while (iter.hasNext()) {
+                LocatedFileStatus fstat = iter.next();
+                String name = getGradoopGraphsetName(fstat.getPath().toUri().getPath());
+                if (name != null && ! dataSources.contains(name)) {
+                    boolean added = dataSources.add(name);
+                    assert added;
+                    log.debug("adding graphset: {}", name);
+                }
+            }
+        } catch (Exception e) {
+            throw new IOException(e);
+        } finally {
+            if (fs != null) {
+                fs.close();
+            }
+        }
+        return dataSources;
+    }
+
+    @Override
+    public String toString() {
+        return "[fs.default.name:" + config.get("fs.default.name") + ", fs.hdfs.impl: " +
+            config.get("fs.hdfs.impl") + ", basePath: " + basePath + "]";
+    }
+
+    private String getGradoopGraphsetName(String path) {
+        //TODO improve this logic
+        String prefix = basePath;
+        String rem = path.substring(prefix.length());
+        int fsi = rem.indexOf('/'); // index of first slash in rem e.g. "one/foo"
+        if (fsi < 0) {
+            log.debug("Returning null, this is not a gradoop graphset: {}", path);
+            return null;
+        }
+        String folder = rem.substring(0, fsi);
+        if (rem.length() - folder.length() > 1) {
+            return folder;
+        }
+        log.debug("Returning null, this is not a gradoop graphset: {}", path);
+        return null;
+    }
+
+    /* package-private */void copyGradoopFiles(String graphsetName, File localBase) throws Exception {
+        // copying all the files is critical, so make it an all-or-nothing operations, cleaning up
+        // however, transactional semantics are not intended here
+        Path basePath = new Path(this.basePath);
+        FileSystem fs = basePath.getFileSystem(config);
+        File localGraphsetFolder = new File(localBase, graphsetName);
+        boolean a = localGraphsetFolder.mkdirs();
+        if (! a) {
+            throw new RuntimeException("Unlikely error in making the local folder: " + localGraphsetFolder);
+        }
+        Path localPath = new Path(localGraphsetFolder.getAbsolutePath());
+        try {
+            for (String f : GRADOOP_FILE_NAMES) {
+                Path remotePath = new Path(this.basePath + "/" + graphsetName + "/" + f);
+                fs.copyToLocalFile(false, remotePath, localPath);
+                log.info("Copied {} to {}", remotePath.toString(), localPath);
+                File gf = new File(localPath.toString(), f);
+                log.debug("File {} -- is it copied? : {}", gf.getAbsolutePath(), gf.exists());
+            }
+        } catch (Exception e) {
+            log.warn("Error copying graphset: {}, deleting the whole folder: " + e.getMessage());
+            FileUtil.fullyDelete(localGraphsetFolder);
+            throw e; //rethrow
+        }
+    }
+}