@Test public void testProcessEdgeList() throws Exception { ConfigurationLoader.testPropertyAccess = true; String path = ConfigurationLoader.getInstance().getHadoopHdfsUri() + "/test/edgeList.txt"; FileUtil.writeListFile(path, new ArrayList<>(Arrays.asList( "0 1\n", "1 3\n", "3 0" )).iterator()); GraphProcessor.processEdgeList(new ProcessorMessage(path, GraphProcessor.TRIANGLE_COUNT, ProcessorMode.Unpartitioned)); }
public static void processEdgeList(ProcessorMessage processorMessage) throws IOException, URISyntaxException { if(javaSparkContext == null) { initializeSparkContext(); } Iterable<String> results = new ArrayList<>(); switch (processorMessage.getAnalysis()) { case PAGERANK: results = algorithms.pageRank(javaSparkContext.sc(), processorMessage.getPath()); break; case CONNECTED_COMPONENTS: results = algorithms.connectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case TRIANGLE_COUNT: results = algorithms.triangleCount(javaSparkContext.sc(), processorMessage.getPath()); break; case STRONGLY_CONNECTED_COMPONENTS: results = algorithms.stronglyConnectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case CLOSENESS_CENTRALITY: results = algorithms.closenessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case BETWEENNESS_CENTRALITY: results = algorithms.betweennessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case EDGE_BETWEENNESS: results = algorithms.edgeBetweenness(javaSparkContext.sc(), processorMessage.getPath()); break; case COLLABORATIVE_FILTERING: results = algorithms.collaborativeFiltering(javaSparkContext.sc(), processorMessage.getPath()); break; default: System.out.println("Did not recognize analysis key: " + processorMessage.getAnalysis()); } if(processorMessage.getMode() == ProcessorMode.Partitioned) { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH.replace("{job_name}", processorMessage.getPartitionDescription().getPartitionId().toString())); } else { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PROPERTY_GRAPH_UPDATE_PATH); } org.mazerunner.core.hdfs.FileUtil.writePropertyGraphUpdate(processorMessage, results); }
GraphProcessor { public static void processEdgeList(ProcessorMessage processorMessage) throws IOException, URISyntaxException { if(javaSparkContext == null) { initializeSparkContext(); } Iterable<String> results = new ArrayList<>(); switch (processorMessage.getAnalysis()) { case PAGERANK: results = algorithms.pageRank(javaSparkContext.sc(), processorMessage.getPath()); break; case CONNECTED_COMPONENTS: results = algorithms.connectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case TRIANGLE_COUNT: results = algorithms.triangleCount(javaSparkContext.sc(), processorMessage.getPath()); break; case STRONGLY_CONNECTED_COMPONENTS: results = algorithms.stronglyConnectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case CLOSENESS_CENTRALITY: results = algorithms.closenessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case BETWEENNESS_CENTRALITY: results = algorithms.betweennessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case EDGE_BETWEENNESS: results = algorithms.edgeBetweenness(javaSparkContext.sc(), processorMessage.getPath()); break; case COLLABORATIVE_FILTERING: results = algorithms.collaborativeFiltering(javaSparkContext.sc(), processorMessage.getPath()); break; default: System.out.println("Did not recognize analysis key: " + processorMessage.getAnalysis()); } if(processorMessage.getMode() == ProcessorMode.Partitioned) { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH.replace("{job_name}", processorMessage.getPartitionDescription().getPartitionId().toString())); } else { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PROPERTY_GRAPH_UPDATE_PATH); } org.mazerunner.core.hdfs.FileUtil.writePropertyGraphUpdate(processorMessage, results); } }
GraphProcessor { public static void processEdgeList(ProcessorMessage processorMessage) throws IOException, URISyntaxException { if(javaSparkContext == null) { initializeSparkContext(); } Iterable<String> results = new ArrayList<>(); switch (processorMessage.getAnalysis()) { case PAGERANK: results = algorithms.pageRank(javaSparkContext.sc(), processorMessage.getPath()); break; case CONNECTED_COMPONENTS: results = algorithms.connectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case TRIANGLE_COUNT: results = algorithms.triangleCount(javaSparkContext.sc(), processorMessage.getPath()); break; case STRONGLY_CONNECTED_COMPONENTS: results = algorithms.stronglyConnectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case CLOSENESS_CENTRALITY: results = algorithms.closenessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case BETWEENNESS_CENTRALITY: results = algorithms.betweennessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case EDGE_BETWEENNESS: results = algorithms.edgeBetweenness(javaSparkContext.sc(), processorMessage.getPath()); break; case COLLABORATIVE_FILTERING: results = algorithms.collaborativeFiltering(javaSparkContext.sc(), processorMessage.getPath()); break; default: System.out.println("Did not recognize analysis key: " + processorMessage.getAnalysis()); } if(processorMessage.getMode() == ProcessorMode.Partitioned) { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH.replace("{job_name}", processorMessage.getPartitionDescription().getPartitionId().toString())); } else { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PROPERTY_GRAPH_UPDATE_PATH); } org.mazerunner.core.hdfs.FileUtil.writePropertyGraphUpdate(processorMessage, results); }  }
GraphProcessor { public static void processEdgeList(ProcessorMessage processorMessage) throws IOException, URISyntaxException { if(javaSparkContext == null) { initializeSparkContext(); } Iterable<String> results = new ArrayList<>(); switch (processorMessage.getAnalysis()) { case PAGERANK: results = algorithms.pageRank(javaSparkContext.sc(), processorMessage.getPath()); break; case CONNECTED_COMPONENTS: results = algorithms.connectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case TRIANGLE_COUNT: results = algorithms.triangleCount(javaSparkContext.sc(), processorMessage.getPath()); break; case STRONGLY_CONNECTED_COMPONENTS: results = algorithms.stronglyConnectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case CLOSENESS_CENTRALITY: results = algorithms.closenessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case BETWEENNESS_CENTRALITY: results = algorithms.betweennessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case EDGE_BETWEENNESS: results = algorithms.edgeBetweenness(javaSparkContext.sc(), processorMessage.getPath()); break; case COLLABORATIVE_FILTERING: results = algorithms.collaborativeFiltering(javaSparkContext.sc(), processorMessage.getPath()); break; default: System.out.println("Did not recognize analysis key: " + processorMessage.getAnalysis()); } if(processorMessage.getMode() == ProcessorMode.Partitioned) { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH.replace("{job_name}", processorMessage.getPartitionDescription().getPartitionId().toString())); } else { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PROPERTY_GRAPH_UPDATE_PATH); } org.mazerunner.core.hdfs.FileUtil.writePropertyGraphUpdate(processorMessage, results); }  static void processEdgeList(ProcessorMessage processorMessage); static JavaSparkContext initializeSparkContext(); }
GraphProcessor { public static void processEdgeList(ProcessorMessage processorMessage) throws IOException, URISyntaxException { if(javaSparkContext == null) { initializeSparkContext(); } Iterable<String> results = new ArrayList<>(); switch (processorMessage.getAnalysis()) { case PAGERANK: results = algorithms.pageRank(javaSparkContext.sc(), processorMessage.getPath()); break; case CONNECTED_COMPONENTS: results = algorithms.connectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case TRIANGLE_COUNT: results = algorithms.triangleCount(javaSparkContext.sc(), processorMessage.getPath()); break; case STRONGLY_CONNECTED_COMPONENTS: results = algorithms.stronglyConnectedComponents(javaSparkContext.sc(), processorMessage.getPath()); break; case CLOSENESS_CENTRALITY: results = algorithms.closenessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case BETWEENNESS_CENTRALITY: results = algorithms.betweennessCentrality(javaSparkContext.sc(), processorMessage.getPath()); break; case EDGE_BETWEENNESS: results = algorithms.edgeBetweenness(javaSparkContext.sc(), processorMessage.getPath()); break; case COLLABORATIVE_FILTERING: results = algorithms.collaborativeFiltering(javaSparkContext.sc(), processorMessage.getPath()); break; default: System.out.println("Did not recognize analysis key: " + processorMessage.getAnalysis()); } if(processorMessage.getMode() == ProcessorMode.Partitioned) { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH.replace("{job_name}", processorMessage.getPartitionDescription().getPartitionId().toString())); } else { processorMessage.setPath(ConfigurationLoader.getInstance().getHadoopHdfsUri() + PROPERTY_GRAPH_UPDATE_PATH); } org.mazerunner.core.hdfs.FileUtil.writePropertyGraphUpdate(processorMessage, results); }  static void processEdgeList(ProcessorMessage processorMessage); static JavaSparkContext initializeSparkContext(); static final String PROPERTY_GRAPH_UPDATE_PATH; static final String PARTITIONED_PROPERTY_GRAPH_UPDATE_PATH; static final String TRIANGLE_COUNT; static final String CONNECTED_COMPONENTS; static final String PAGERANK; static final String STRONGLY_CONNECTED_COMPONENTS; static final String CLOSENESS_CENTRALITY; static final String BETWEENNESS_CENTRALITY; static final String EDGE_BETWEENNESS; static final String COLLABORATIVE_FILTERING; static JavaSparkContext javaSparkContext; }