@Test public void testWriteToHadoop() throws Exception { GraphDatabaseService db = setUpDb(); ConfigurationLoader.testPropertyAccess = true; createSampleGraph(db); Writer.exportSubgraphToHDFS(db); }
public static Path exportSubgraphToHDFS(GraphDatabaseService db) throws IOException, URISyntaxException { FileSystem fs = FileUtil.getHadoopFileSystem(); Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("/{job_id}", "")); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt))); Transaction tx = db.beginTx(); Iterable<Node> nodes = GlobalGraphOperations.at(db) .getAllNodes(); br.write("# Adacency list" + "\n"); int nodeTotal = IteratorUtil.count(nodes); final int[] nodeCount = {0}; final int[] pathCount = {0}; int pathCountBlocks = 10000; int size = IteratorUtil.count(nodes.iterator()); nodes.iterator().forEachRemaining(n -> { Iterable<org.neo4j.graphdb.Path> nPaths = db.traversalDescription() .depthFirst() .relationships(withName(ConfigurationLoader.getInstance().getMazerunnerRelationshipType()), Direction.OUTGOING) .evaluator(Evaluators.fromDepth(1)) .evaluator(Evaluators.toDepth(1)) .traverse(n); for (org.neo4j.graphdb.Path path : nPaths) { try { String line = path.startNode().getId() + " " + path.endNode().getId(); br.write(line + "\n"); pathCount[0]++; if (pathCount[0] > pathCountBlocks) { pathCount[0] = 0; System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#%}", ((double) nodeCount[0] / (double) nodeTotal))); } } catch (Exception ex) { System.out.println(ex.getMessage()); } } nodeCount[0]++; }); System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); tx.success(); tx.close(); return pt; }
Writer { public static Path exportSubgraphToHDFS(GraphDatabaseService db) throws IOException, URISyntaxException { FileSystem fs = FileUtil.getHadoopFileSystem(); Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("/{job_id}", "")); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt))); Transaction tx = db.beginTx(); Iterable<Node> nodes = GlobalGraphOperations.at(db) .getAllNodes(); br.write("# Adacency list" + "\n"); int nodeTotal = IteratorUtil.count(nodes); final int[] nodeCount = {0}; final int[] pathCount = {0}; int pathCountBlocks = 10000; int size = IteratorUtil.count(nodes.iterator()); nodes.iterator().forEachRemaining(n -> { Iterable<org.neo4j.graphdb.Path> nPaths = db.traversalDescription() .depthFirst() .relationships(withName(ConfigurationLoader.getInstance().getMazerunnerRelationshipType()), Direction.OUTGOING) .evaluator(Evaluators.fromDepth(1)) .evaluator(Evaluators.toDepth(1)) .traverse(n); for (org.neo4j.graphdb.Path path : nPaths) { try { String line = path.startNode().getId() + " " + path.endNode().getId(); br.write(line + "\n"); pathCount[0]++; if (pathCount[0] > pathCountBlocks) { pathCount[0] = 0; System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#%}", ((double) nodeCount[0] / (double) nodeTotal))); } } catch (Exception ex) { System.out.println(ex.getMessage()); } } nodeCount[0]++; }); System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); tx.success(); tx.close(); return pt; } }
Writer { public static Path exportSubgraphToHDFS(GraphDatabaseService db) throws IOException, URISyntaxException { FileSystem fs = FileUtil.getHadoopFileSystem(); Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("/{job_id}", "")); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt))); Transaction tx = db.beginTx(); Iterable<Node> nodes = GlobalGraphOperations.at(db) .getAllNodes(); br.write("# Adacency list" + "\n"); int nodeTotal = IteratorUtil.count(nodes); final int[] nodeCount = {0}; final int[] pathCount = {0}; int pathCountBlocks = 10000; int size = IteratorUtil.count(nodes.iterator()); nodes.iterator().forEachRemaining(n -> { Iterable<org.neo4j.graphdb.Path> nPaths = db.traversalDescription() .depthFirst() .relationships(withName(ConfigurationLoader.getInstance().getMazerunnerRelationshipType()), Direction.OUTGOING) .evaluator(Evaluators.fromDepth(1)) .evaluator(Evaluators.toDepth(1)) .traverse(n); for (org.neo4j.graphdb.Path path : nPaths) { try { String line = path.startNode().getId() + " " + path.endNode().getId(); br.write(line + "\n"); pathCount[0]++; if (pathCount[0] > pathCountBlocks) { pathCount[0] = 0; System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#%}", ((double) nodeCount[0] / (double) nodeTotal))); } } catch (Exception ex) { System.out.println(ex.getMessage()); } } nodeCount[0]++; }); System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); tx.success(); tx.close(); return pt; }  }
Writer { public static Path exportSubgraphToHDFS(GraphDatabaseService db) throws IOException, URISyntaxException { FileSystem fs = FileUtil.getHadoopFileSystem(); Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("/{job_id}", "")); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt))); Transaction tx = db.beginTx(); Iterable<Node> nodes = GlobalGraphOperations.at(db) .getAllNodes(); br.write("# Adacency list" + "\n"); int nodeTotal = IteratorUtil.count(nodes); final int[] nodeCount = {0}; final int[] pathCount = {0}; int pathCountBlocks = 10000; int size = IteratorUtil.count(nodes.iterator()); nodes.iterator().forEachRemaining(n -> { Iterable<org.neo4j.graphdb.Path> nPaths = db.traversalDescription() .depthFirst() .relationships(withName(ConfigurationLoader.getInstance().getMazerunnerRelationshipType()), Direction.OUTGOING) .evaluator(Evaluators.fromDepth(1)) .evaluator(Evaluators.toDepth(1)) .traverse(n); for (org.neo4j.graphdb.Path path : nPaths) { try { String line = path.startNode().getId() + " " + path.endNode().getId(); br.write(line + "\n"); pathCount[0]++; if (pathCount[0] > pathCountBlocks) { pathCount[0] = 0; System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#%}", ((double) nodeCount[0] / (double) nodeTotal))); } } catch (Exception ex) { System.out.println(ex.getMessage()); } } nodeCount[0]++; }); System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); tx.success(); tx.close(); return pt; }  static void dispatchJob(GraphDatabaseService db, String type); static void dispatchPartitionedJob(GraphDatabaseService db, String type, PartitionDescription partitionDescription, Path pt); static Path exportPartitionToHDFSParallel(GraphDatabaseService db, Node partitionNode, PartitionDescription partitionDescription); static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType); static Path exportSubgraphToHDFSParallel(GraphDatabaseService db); static void writeBlockForNode(Node n, GraphDatabaseService db, BufferedWriter bufferedWriter, int reportBlockSize, String relationshipType); static void writeBlockForQueryResult(Map<String, Object> row, BufferedWriter bufferedWriter, int reportBlockSize, List<String> columns); static void updatePartitionBlockForRow(String line, GraphDatabaseService db, int reportBlockSize, ProcessorMessage processorMessage, Node partitionNode); static void updateBlockForRow(String line, GraphDatabaseService db, int reportBlockSize, String analysis); static void asyncPartitionedUpdate(BufferedReader bufferedReader, GraphDatabaseService graphDb, ProcessorMessage processorMessage); static void asyncImportCollaborativeFiltering(BufferedReader bufferedReader, GraphDatabaseService graphDb); static void updateCollaborativeFilteringForRow(String line, GraphDatabaseService db, int reportBlockSize); static void asyncUpdate(ProcessorMessage analysis, BufferedReader bufferedReader, GraphDatabaseService graphDb); static Path exportSubgraphToHDFS(GraphDatabaseService db); }
Writer { public static Path exportSubgraphToHDFS(GraphDatabaseService db) throws IOException, URISyntaxException { FileSystem fs = FileUtil.getHadoopFileSystem(); Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("/{job_id}", "")); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt))); Transaction tx = db.beginTx(); Iterable<Node> nodes = GlobalGraphOperations.at(db) .getAllNodes(); br.write("# Adacency list" + "\n"); int nodeTotal = IteratorUtil.count(nodes); final int[] nodeCount = {0}; final int[] pathCount = {0}; int pathCountBlocks = 10000; int size = IteratorUtil.count(nodes.iterator()); nodes.iterator().forEachRemaining(n -> { Iterable<org.neo4j.graphdb.Path> nPaths = db.traversalDescription() .depthFirst() .relationships(withName(ConfigurationLoader.getInstance().getMazerunnerRelationshipType()), Direction.OUTGOING) .evaluator(Evaluators.fromDepth(1)) .evaluator(Evaluators.toDepth(1)) .traverse(n); for (org.neo4j.graphdb.Path path : nPaths) { try { String line = path.startNode().getId() + " " + path.endNode().getId(); br.write(line + "\n"); pathCount[0]++; if (pathCount[0] > pathCountBlocks) { pathCount[0] = 0; System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#%}", ((double) nodeCount[0] / (double) nodeTotal))); } } catch (Exception ex) { System.out.println(ex.getMessage()); } } nodeCount[0]++; }); System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); tx.success(); tx.close(); return pt; }  static void dispatchJob(GraphDatabaseService db, String type); static void dispatchPartitionedJob(GraphDatabaseService db, String type, PartitionDescription partitionDescription, Path pt); static Path exportPartitionToHDFSParallel(GraphDatabaseService db, Node partitionNode, PartitionDescription partitionDescription); static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType); static Path exportSubgraphToHDFSParallel(GraphDatabaseService db); static void writeBlockForNode(Node n, GraphDatabaseService db, BufferedWriter bufferedWriter, int reportBlockSize, String relationshipType); static void writeBlockForQueryResult(Map<String, Object> row, BufferedWriter bufferedWriter, int reportBlockSize, List<String> columns); static void updatePartitionBlockForRow(String line, GraphDatabaseService db, int reportBlockSize, ProcessorMessage processorMessage, Node partitionNode); static void updateBlockForRow(String line, GraphDatabaseService db, int reportBlockSize, String analysis); static void asyncPartitionedUpdate(BufferedReader bufferedReader, GraphDatabaseService graphDb, ProcessorMessage processorMessage); static void asyncImportCollaborativeFiltering(BufferedReader bufferedReader, GraphDatabaseService graphDb); static void updateCollaborativeFilteringForRow(String line, GraphDatabaseService db, int reportBlockSize); static void asyncUpdate(ProcessorMessage analysis, BufferedReader bufferedReader, GraphDatabaseService graphDb); static Path exportSubgraphToHDFS(GraphDatabaseService db); static final String EDGE_LIST_RELATIVE_FILE_PATH; static Integer updateCounter; static Integer counter; }