@Test public void testParallelWriteCypherResultToHDFS() throws Exception { GraphDatabaseService db = setUpDb(); ConfigurationLoader.testPropertyAccess = true; createSampleGraph(db); Writer.exportCypherQueryToHDFSParallel(db, "MATCH (n1)-[r]->(n2) RETURN id(n1) as v1, id(n2) as v2, rand() as v3", JobRequestType.COLLABORATIVE_FILTERING); }
public static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType) 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))); Integer reportBlockSize = 20000; try (Transaction ignored = db.beginTx(); Result result = db.execute(cypherQuery)) { if(result.columns().size() != JobRequestType.parameterSize(jobRequestType)) { ignored.close(); throw new RuntimeException(format("The Cypher query must return %s columns per row for %s jobs", JobRequestType.parameterSize(jobRequestType), jobRequestType.toString())); } List<Spliterator<Map<String, Object>>> spliteratorList = new ArrayList<>(); boolean hasSpliterator = true; Spliterator<Map<String, Object>> nodeSpliterator; nodeSpliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.SIZED); while (hasSpliterator) { Spliterator<Map<String, Object>> localSpliterator = nodeSpliterator.trySplit(); hasSpliterator = localSpliterator != null; if (hasSpliterator) spliteratorList.add(localSpliterator); } counter = 0; if (spliteratorList.size() > 4) { ParallelWriter parallelWriter = new ParallelWriter<Map<String, Object>>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]), new CypherWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, result.columns())); ForkJoinPool pool = new ForkJoinPool(); pool.invoke(parallelWriter); } else { spliteratorList.forEach(sl -> sl.forEachRemaining(n -> { try { writeBlockForQueryResult(n, br, reportBlockSize, result.columns()); } catch (IOException e) { e.printStackTrace(); } })); } System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); return pt; } }
Writer { public static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType) 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))); Integer reportBlockSize = 20000; try (Transaction ignored = db.beginTx(); Result result = db.execute(cypherQuery)) { if(result.columns().size() != JobRequestType.parameterSize(jobRequestType)) { ignored.close(); throw new RuntimeException(format("The Cypher query must return %s columns per row for %s jobs", JobRequestType.parameterSize(jobRequestType), jobRequestType.toString())); } List<Spliterator<Map<String, Object>>> spliteratorList = new ArrayList<>(); boolean hasSpliterator = true; Spliterator<Map<String, Object>> nodeSpliterator; nodeSpliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.SIZED); while (hasSpliterator) { Spliterator<Map<String, Object>> localSpliterator = nodeSpliterator.trySplit(); hasSpliterator = localSpliterator != null; if (hasSpliterator) spliteratorList.add(localSpliterator); } counter = 0; if (spliteratorList.size() > 4) { ParallelWriter parallelWriter = new ParallelWriter<Map<String, Object>>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]), new CypherWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, result.columns())); ForkJoinPool pool = new ForkJoinPool(); pool.invoke(parallelWriter); } else { spliteratorList.forEach(sl -> sl.forEachRemaining(n -> { try { writeBlockForQueryResult(n, br, reportBlockSize, result.columns()); } catch (IOException e) { e.printStackTrace(); } })); } System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); return pt; } } }
Writer { public static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType) 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))); Integer reportBlockSize = 20000; try (Transaction ignored = db.beginTx(); Result result = db.execute(cypherQuery)) { if(result.columns().size() != JobRequestType.parameterSize(jobRequestType)) { ignored.close(); throw new RuntimeException(format("The Cypher query must return %s columns per row for %s jobs", JobRequestType.parameterSize(jobRequestType), jobRequestType.toString())); } List<Spliterator<Map<String, Object>>> spliteratorList = new ArrayList<>(); boolean hasSpliterator = true; Spliterator<Map<String, Object>> nodeSpliterator; nodeSpliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.SIZED); while (hasSpliterator) { Spliterator<Map<String, Object>> localSpliterator = nodeSpliterator.trySplit(); hasSpliterator = localSpliterator != null; if (hasSpliterator) spliteratorList.add(localSpliterator); } counter = 0; if (spliteratorList.size() > 4) { ParallelWriter parallelWriter = new ParallelWriter<Map<String, Object>>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]), new CypherWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, result.columns())); ForkJoinPool pool = new ForkJoinPool(); pool.invoke(parallelWriter); } else { spliteratorList.forEach(sl -> sl.forEachRemaining(n -> { try { writeBlockForQueryResult(n, br, reportBlockSize, result.columns()); } catch (IOException e) { e.printStackTrace(); } })); } System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.close(); return pt; } }  }
Writer { public static Path exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType) 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))); Integer reportBlockSize = 20000; try (Transaction ignored = db.beginTx(); Result result = db.execute(cypherQuery)) { if(result.columns().size() != JobRequestType.parameterSize(jobRequestType)) { ignored.close(); throw new RuntimeException(format("The Cypher query must return %s columns per row for %s jobs", JobRequestType.parameterSize(jobRequestType), jobRequestType.toString())); } List<Spliterator<Map<String, Object>>> spliteratorList = new ArrayList<>(); boolean hasSpliterator = true; Spliterator<Map<String, Object>> nodeSpliterator; nodeSpliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.SIZED); while (hasSpliterator) { Spliterator<Map<String, Object>> localSpliterator = nodeSpliterator.trySplit(); hasSpliterator = localSpliterator != null; if (hasSpliterator) spliteratorList.add(localSpliterator); } counter = 0; if (spliteratorList.size() > 4) { ParallelWriter parallelWriter = new ParallelWriter<Map<String, Object>>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]), new CypherWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, result.columns())); ForkJoinPool pool = new ForkJoinPool(); pool.invoke(parallelWriter); } else { spliteratorList.forEach(sl -> sl.forEachRemaining(n -> { try { writeBlockForQueryResult(n, br, reportBlockSize, result.columns()); } catch (IOException e) { e.printStackTrace(); } })); } System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.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 exportCypherQueryToHDFSParallel(GraphDatabaseService db, String cypherQuery, JobRequestType jobRequestType) 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))); Integer reportBlockSize = 20000; try (Transaction ignored = db.beginTx(); Result result = db.execute(cypherQuery)) { if(result.columns().size() != JobRequestType.parameterSize(jobRequestType)) { ignored.close(); throw new RuntimeException(format("The Cypher query must return %s columns per row for %s jobs", JobRequestType.parameterSize(jobRequestType), jobRequestType.toString())); } List<Spliterator<Map<String, Object>>> spliteratorList = new ArrayList<>(); boolean hasSpliterator = true; Spliterator<Map<String, Object>> nodeSpliterator; nodeSpliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.SIZED); while (hasSpliterator) { Spliterator<Map<String, Object>> localSpliterator = nodeSpliterator.trySplit(); hasSpliterator = localSpliterator != null; if (hasSpliterator) spliteratorList.add(localSpliterator); } counter = 0; if (spliteratorList.size() > 4) { ParallelWriter parallelWriter = new ParallelWriter<Map<String, Object>>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]), new CypherWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, result.columns())); ForkJoinPool pool = new ForkJoinPool(); pool.invoke(parallelWriter); } else { spliteratorList.forEach(sl -> sl.forEachRemaining(n -> { try { writeBlockForQueryResult(n, br, reportBlockSize, result.columns()); } catch (IOException e) { e.printStackTrace(); } })); } System.out.println("Mazerunner Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0)); br.flush(); br.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; }