Skip to content
This repository was archived by the owner on Apr 22, 2020. It is now read-only.

Commit 334307a

Browse files
mneedhammneedham
committed
remove duplication
1 parent 97d3354 commit 334307a

File tree

3 files changed

+63
-75
lines changed

3 files changed

+63
-75
lines changed

‎algo/src/main/java/org/neo4j/graphalgo/similarity/ParallelSimilarityExporter.java‎

Lines changed: 17 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -45,21 +45,15 @@
4545
import java.util.stream.Collectors;
4646
import java.util.stream.Stream;
4747

48-
public class ParallelSimilarityExporter extends StatementApi implements SimilarityExporter {
48+
public class ParallelSimilarityExporter extends SimilarityExporter {
4949

50-
private final Log log;
51-
private final int propertyId;
52-
private final int relationshipTypeId;
5350
private final int nodeCount;
5451

5552
public ParallelSimilarityExporter(GraphDatabaseAPI api,
5653
Log log,
5754
String relationshipType,
5855
String propertyName, int nodeCount) {
59-
super(api);
60-
this.log = log;
61-
propertyId = getOrCreatePropertyId(propertyName);
62-
relationshipTypeId = getOrCreateRelationshipId(relationshipType);
56+
super(api, log, propertyName, relationshipType);
6357
this.nodeCount = nodeCount;
6458
}
6559

@@ -83,24 +77,18 @@ public int export(Stream<SimilarityResult> similarityPairs, long batchSize) {
8377

8478
DisjointSetStruct dssResult = computePartitions(graph);
8579

86-
Stream<List<DisjointSetStruct.InternalResult>> stream = dssResult.internalResultStream(graph)
87-
.collect(Collectors.groupingBy(item -> item.setId))
88-
.values()
89-
.stream();
80+
Stream<List<DisjointSetStruct.InternalResult>> partitions = groupPartitions(graph, dssResult);
9081

91-
int queueSize = dssResult.getSetCount();
92-
93-
if(queueSize == 0) {
82+
int numberOfPartitions = dssResult.getSetCount();
83+
if(numberOfPartitions == 0) {
9484
return 0;
9585
}
9686

97-
log.info("ParallelSimilarityExporter: Relationships to be created: %d, Partitions found: %d", numberOfRelationships[0], queueSize);
98-
99-
ArrayBlockingQueue<List<SimilarityResult>> outQueue = new ArrayBlockingQueue<>(queueSize);
100-
87+
log.info("ParallelSimilarityExporter: Relationships to be created: %d, Partitions found: %d", numberOfRelationships[0], numberOfPartitions);
88+
ArrayBlockingQueue<List<SimilarityResult>> outQueue = new ArrayBlockingQueue<>(numberOfPartitions);
10189

10290
AtomicInteger inQueueBatchCount = new AtomicInteger(0);
103-
stream.parallel().forEach(partition -> {
91+
partitions.parallel().forEach(partition -> {
10492
IntSet nodesInPartition = new IntHashSet();
10593
for (DisjointSetStruct.InternalResult internalResult : partition) {
10694
nodesInPartition.add(internalResult.internalNodeId);
@@ -137,13 +125,20 @@ public int export(Stream<SimilarityResult> similarityPairs, long batchSize) {
137125

138126

139127
int inQueueBatches = inQueueBatchCount.get();
140-
141-
142128
int outQueueBatches = writeSequential(outQueue.stream().flatMap(Collection::stream), batchSize);
129+
143130
log.info("ParallelSimilarityExporter: Batch Size: %d, Batches written - in parallel: %d, sequentially: %d", batchSize, inQueueBatches, outQueueBatches);
131+
144132
return inQueueBatches + outQueueBatches;
145133
}
146134

135+
private Stream<List<DisjointSetStruct.InternalResult>> groupPartitions(HeavyGraph graph, DisjointSetStruct dssResult) {
136+
return dssResult.internalResultStream(graph)
137+
.collect(Collectors.groupingBy(item -> item.setId))
138+
.values()
139+
.stream();
140+
}
141+
147142
private static <T> void put(BlockingQueue<T> queue, T items) {
148143
try {
149144
queue.put(items);
@@ -185,27 +180,6 @@ private void export(List<SimilarityResult> similarityResults) {
185180

186181
}
187182

188-
private void createRelationship(SimilarityResult similarityResult, KernelTransaction statement) throws EntityNotFoundException, InvalidTransactionTypeKernelException, AutoIndexingKernelException {
189-
long node1 = similarityResult.item1;
190-
long node2 = similarityResult.item2;
191-
long relationshipId = statement.dataWrite().relationshipCreate(node1, relationshipTypeId, node2);
192-
193-
statement.dataWrite().relationshipSetProperty(
194-
relationshipId, propertyId, Values.doubleValue(similarityResult.similarity));
195-
}
196-
197-
private int getOrCreateRelationshipId(String relationshipType) {
198-
return applyInTransaction(stmt -> stmt
199-
.tokenWrite()
200-
.relationshipTypeGetOrCreateForName(relationshipType));
201-
}
202-
203-
private int getOrCreatePropertyId(String propertyName) {
204-
return applyInTransaction(stmt -> stmt
205-
.tokenWrite()
206-
.propertyKeyGetOrCreateForName(propertyName));
207-
}
208-
209183
private int writeSequential(Stream<SimilarityResult> similarityPairs, long batchSize) {
210184
int[] counter = {0};
211185
if (batchSize == 1) {

‎algo/src/main/java/org/neo4j/graphalgo/similarity/SequentialSimilarityExporter.java‎

Lines changed: 3 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -34,19 +34,13 @@
3434
import java.util.List;
3535
import java.util.stream.Stream;
3636

37-
public class SequentialSimilarityExporter extends StatementApi implements SimilarityExporter {
38-
39-
private final Log log;
40-
private final int propertyId;
41-
private final int relationshipTypeId;
37+
public class SequentialSimilarityExporter extends SimilarityExporter {
4238

4339
public SequentialSimilarityExporter(GraphDatabaseAPI api,
4440
Log log, String relationshipType,
4541
String propertyName, int nodeCount) {
46-
super(api);
47-
this.log = log;
48-
propertyId = getOrCreatePropertyId(propertyName);
49-
relationshipTypeId = getOrCreateRelationshipId(relationshipType);
42+
super(api, log, propertyName, relationshipType);
43+
5044
}
5145

5246
public int export(Stream<SimilarityResult> similarityPairs, long batchSize) {
@@ -81,27 +75,6 @@ private void export(List<SimilarityResult> similarityResults) {
8175

8276
}
8377

84-
private void createRelationship(SimilarityResult similarityResult, KernelTransaction statement) throws EntityNotFoundException, InvalidTransactionTypeKernelException, AutoIndexingKernelException {
85-
long node1 = similarityResult.item1;
86-
long node2 = similarityResult.item2;
87-
long relationshipId = statement.dataWrite().relationshipCreate(node1, relationshipTypeId, node2);
88-
89-
statement.dataWrite().relationshipSetProperty(
90-
relationshipId, propertyId, Values.doubleValue(similarityResult.similarity));
91-
}
92-
93-
private int getOrCreateRelationshipId(String relationshipType) {
94-
return applyInTransaction(stmt -> stmt
95-
.tokenWrite()
96-
.relationshipTypeGetOrCreateForName(relationshipType));
97-
}
98-
99-
private int getOrCreatePropertyId(String propertyName) {
100-
return applyInTransaction(stmt -> stmt
101-
.tokenWrite()
102-
.propertyKeyGetOrCreateForName(propertyName));
103-
}
104-
10578
private int writeSequential(Stream<SimilarityResult> similarityPairs, long batchSize) {
10679
log.info("SequentialSimilarityExporter: Writing relationships...");
10780
int[] counter = {0};

‎algo/src/main/java/org/neo4j/graphalgo/similarity/SimilarityExporter.java‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,48 @@
11
package org.neo4j.graphalgo.similarity;
22

3+
import org.neo4j.graphalgo.core.utils.StatementApi;
4+
import org.neo4j.internal.kernel.api.exceptions.EntityNotFoundException;
5+
import org.neo4j.internal.kernel.api.exceptions.InvalidTransactionTypeKernelException;
6+
import org.neo4j.internal.kernel.api.exceptions.explicitindex.AutoIndexingKernelException;
7+
import org.neo4j.kernel.api.KernelTransaction;
8+
import org.neo4j.kernel.internal.GraphDatabaseAPI;
9+
import org.neo4j.logging.Log;
10+
import org.neo4j.values.storable.Values;
11+
312
import java.util.stream.Stream;
413

5-
public interface SimilarityExporter {
6-
int export(Stream<SimilarityResult> similarityPairs, long batchSize);
14+
public abstract class SimilarityExporter extends StatementApi {
15+
final Log log;
16+
final int propertyId;
17+
final int relationshipTypeId;
18+
19+
protected SimilarityExporter(GraphDatabaseAPI api, Log log, String propertyName, String relationshipType) {
20+
super(api);
21+
this.log = log;
22+
propertyId = getOrCreatePropertyId(propertyName);
23+
relationshipTypeId = getOrCreateRelationshipTypeId(relationshipType);
24+
}
25+
26+
private int getOrCreateRelationshipTypeId(String relationshipType) {
27+
return applyInTransaction(stmt -> stmt
28+
.tokenWrite()
29+
.relationshipTypeGetOrCreateForName(relationshipType));
30+
}
31+
32+
private int getOrCreatePropertyId(String propertyName) {
33+
return applyInTransaction(stmt -> stmt
34+
.tokenWrite()
35+
.propertyKeyGetOrCreateForName(propertyName));
36+
}
37+
38+
protected void createRelationship(SimilarityResult similarityResult, KernelTransaction statement) throws EntityNotFoundException, InvalidTransactionTypeKernelException, AutoIndexingKernelException {
39+
long node1 = similarityResult.item1;
40+
long node2 = similarityResult.item2;
41+
long relationshipId = statement.dataWrite().relationshipCreate(node1, relationshipTypeId, node2);
42+
43+
statement.dataWrite().relationshipSetProperty(
44+
relationshipId, propertyId, Values.doubleValue(similarityResult.similarity));
45+
}
46+
47+
abstract int export(Stream<SimilarityResult> similarityPairs, long batchSize);
748
}

0 commit comments

Comments
 (0)