Java 类org.neo4j.graphdb.traversal.Evaluators 实例源码

public static List<String> getHypernyms(String cui){
  List<String> hypers = new ArrayList<>();

  try ( Transaction tx = graphDb.beginTx() ){
    TraversalDescription td = graphDb.traversalDescription()
        .breadthFirst()
        .relationships(RelReader.RelTypes.ISA, Direction.OUTGOING)
        .evaluator(Evaluators.excludeStartPosition());

    Node cuiNode = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui);
    if(cuiNode == null) return hypers;

    Traverser traverser = td.traverse(cuiNode);
    for(Path path : traverser){
      hypers.add(path.endNode().getProperty(RelReader.CUI_PROPERTY).toString());
    }
    tx.success();
  }
  return hypers;
}

public static List<String> getHyponyms(String cui){
  List<String> hypos = new ArrayList<>();

  try ( Transaction tx = graphDb.beginTx() ){
    TraversalDescription td = graphDb.traversalDescription()
        .breadthFirst()
        .relationships(RelReader.RelTypes.ISA, Direction.INCOMING)
        .evaluator(Evaluators.excludeStartPosition());

    Node cuiNode = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui);
    if(cuiNode == null) return hypos;

    Traverser traverser = td.traverse(cuiNode);
    for(Path path : traverser){
      hypos.add(path.endNode().getProperty(RelReader.CUI_PROPERTY).toString());
    }
    tx.success();
  }
  return hypos;    
}

public static boolean isa(String cui1, String cui2){
  boolean match=false;
  try ( Transaction tx = graphDb.beginTx() ){
    Node cui1Node = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui1);
    Node cui2Node = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui2);
    if(cui1Node == null || cui2Node == null) return match;

    TraversalDescription td = graphDb.traversalDescription()
        .breadthFirst()
        .relationships(RelReader.RelTypes.ISA, Direction.OUTGOING)
        .evaluator(Evaluators.excludeStartPosition())
        .evaluator(Evaluators.includeWhereEndNodeIs(cui2Node));

    Traverser traverser = td.traverse(cui1Node);
    if(traverser.iterator().hasNext()){
      match = true;
    }
    tx.success();
  }
  return match;
}

public static int minDistance(String cui1, String cui2){
  int distance = -1;
  try ( Transaction tx = graphDb.beginTx() ){
    Node cui1Node = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui1);
    Node cui2Node = graphDb.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, cui2);
    if(cui1Node == null || cui2Node == null) return distance;

    TraversalDescription td = graphDb.traversalDescription()
        .breadthFirst()
        .relationships(RelReader.RelTypes.ISA, Direction.OUTGOING)
        .evaluator(Evaluators.excludeStartPosition())
        .evaluator(Evaluators.includeWhereEndNodeIs(cui2Node));

    Traverser traverser = td.traverse(cui1Node);
    for(Path path : traverser){
      int len = path.length();
      if(distance == -1 || len < distance){
        distance = len;
      }
    }
    tx.success();
  }
  return distance;
}

@Test
public void testRelReader() throws IOException{

  RelReader reader = new RelReader(neo4jLocation);
  reader.batchBuildGraph(new File("my_test_umls/"), "CtakesAllTuis.txt", "SNOMEDCT_US");
  GraphDatabaseService db = new GraphDatabaseFactory().newEmbeddedDatabase(new File(neo4jLocation));

  try ( Transaction tx = db.beginTx() ){
    TraversalDescription td = db.traversalDescription()
        .breadthFirst()
        .relationships(RelReader.RelTypes.ISA, Direction.INCOMING)
        .evaluator(Evaluators.excludeStartPosition());

    Node cuiNode = db.findNode(RelReader.DictLabels.Concept, RelReader.CUI_PROPERTY, "C0007102");
    Assert.assertNotNull(cuiNode);
    Traverser traverser = td.traverse(cuiNode);
    for(Path path : traverser){
      System.out.println("At depth " + path.length() + " => " + path.endNode().getProperty("cui"));
    }
  }
  db.shutdown();
}

public String knowsLikesTraverser( Node node )
{
    String output = "";
    // START SNIPPET: knowslikestraverser
    for ( Path position : db.traversalDescription()
            .depthFirst()
            .relationships( Rels.KNOWS )
            .relationships( Rels.LIKES, Direction.INCOMING )
            .evaluator( Evaluators.toDepth( 5 ) )
            .traverse( node ) )
    {
        output += position + "\n";
    }
    // END SNIPPET: knowslikestraverser
    return output;
}

private Evaluator addBlaEvaluator( Evaluator evaluator )
{
    if ( this.evaluator instanceof MultiEvaluator )
    {
        return ((MultiEvaluator) this.evaluator).add( evaluator );
    }
    else
    {
        if ( this.evaluator == Evaluators.all() )
        {
            return evaluator;
        }
        else
        {
            return new MultiEvaluator( new Evaluator[] { this.evaluator, evaluator } );
        }
    }
}

public List<Category> getAll() {
    List<Category> categories = new ArrayList<Category>();
    for(Category cat : catRepo.findAllByTraversal(getCategoryRoot(), Traversal.description()
            .breadthFirst()
            .evaluator(Evaluators.fromDepth(1))
            .expand(Traversal
                .expanderForTypes(DynamicRelationshipType
                .withName(RelTypes.PARENT_CATEGORY), 
                          Direction.INCOMING).addNodeFilter(new Predicate<Node>() {
                                @Override
                                public boolean accept(Node item) {
                                    return !StringUtils.isEmpty(item.getProperty("name"));
                                }
                        }
                        )))) {
        if(cat != null && cat.getName() != null) { //TODO: Figure out why a null element appears in this traversal
            categories.add(cat);
        }
    }
    return categories;
}

private void fillNodesFromBeginning(Node currentNode, ViralShortUrl url) {
    // nodes from beginning, counted via a simple depth-first traversal
    // in the opposite direction
    Traverser t = Traversal.description().depthFirst()
        .evaluator(Evaluators.excludeStartPosition())
        .relationships(LinkRelationship.SPAWNS, Direction.INCOMING)
        .traverse(currentNode);
    int nodesFromBeginning = IteratorUtil.count(t);
    url.setNodesFromBeginning(nodesFromBeginning);
}

/**
     * demo usage of traversal API, see http://docs.neo4j.org/chunked/stable/tutorial-traversal.html
     */
    @GET
    @Produces("application/json")
    @Path("/traversalapi/{label}/{key}/{value}/{relType}/{depth}")
    public People findConnectedNodesUsingTraversalAPI(
            @PathParam("label") String label,
            @PathParam("key") String key,
            @PathParam("value") String value,
            @PathParam("relType") String relType,
            @PathParam("depth") int depth) {

        try (Transaction tx = graphDatabaseService.beginTx()) {
            Node startNode = IteratorUtil.single(graphDatabaseService.findNodesByLabelAndProperty(
                    DynamicLabel.label(label), key, value
            ));

            Iterable<org.neo4j.graphdb.Path> traverserResult =
                    graphDatabaseService
                            .traversalDescription()
                            .relationships(DynamicRelationshipType.withName(relType), Direction.OUTGOING)
                            .uniqueness(Uniqueness.NODE_LEVEL)
                            .evaluator(Evaluators.atDepth(depth))
//                            .evaluator(new TraversalPrinter(Evaluators.atDepth(depth)))
                            .traverse(startNode);
            return new People(new LastNodePropertyIterator(traverserResult, key));
        }
    }

/***
 * @param parent
 * @param relationship
 * @param traverseEquivalentEdges
 * @return the entailment
 */
public Collection<Node> getEntailment(Node parent, DirectedRelationshipType relationship,
    boolean traverseEquivalentEdges) {
  Set<Node> entailment = new HashSet<>();
  TraversalDescription description = graphDb.traversalDescription().depthFirst()
      .relationships(relationship.getType(), relationship.getDirection())
      .evaluator(Evaluators.fromDepth(0)).evaluator(Evaluators.all());
  if (traverseEquivalentEdges) {
    description = description.relationships(OwlRelationships.OWL_EQUIVALENT_CLASS);
  }
  for (Path path : description.traverse(parent)) {
    entailment.add(path.endNode());
  }
  return entailment;
}

public String depth3( Node node )
{
    String output = "";
    // START SNIPPET: depth3
    for ( Path path : friendsTraversal
            .evaluator( Evaluators.toDepth( 3 ) )
            .traverse( node ) )
    {
        output += path + "\n";
    }
    // END SNIPPET: depth3
    return output;
}

public String depth4( Node node )
{
    String output = "";
    // START SNIPPET: depth4
    for ( Path path : friendsTraversal
            .evaluator( Evaluators.fromDepth( 2 ) )
            .evaluator( Evaluators.toDepth( 4 ) )
            .traverse( node ) )
    {
        output += path + "\n";
    }
    // END SNIPPET: depth4
    return output;
}

private static void determineKeyEntities(final GraphDatabaseService graphDB, final AttributePath commonAttributePath,
                                         final ContentSchema prefixedContentSchema, final Map<Long, CSEntity> csEntities, final Node... csEntityNodesArray) {

    for (final AttributePath keyAttributePath : prefixedContentSchema.getKeyAttributePaths()) {

        final Optional<LinkedList<Attribute>> optionalRelativeKeyAttributePath = determineRelativeAttributePath(keyAttributePath,
                commonAttributePath);

        if (optionalRelativeKeyAttributePath.isPresent()) {

            final LinkedList<Attribute> relativeKeyAttributePath = optionalRelativeKeyAttributePath.get();

            final Iterable<Path> relativeKeyPaths = graphDB.traversalDescription().depthFirst()
                    .evaluator(Evaluators.toDepth(relativeKeyAttributePath.size())).evaluator(new EntityEvaluator(relativeKeyAttributePath))
                    .traverse(csEntityNodesArray);

            for (final Path relativeKeyPath : relativeKeyPaths) {

                final Node keyNode = relativeKeyPath.endNode();
                final Node csEntityNode = relativeKeyPath.startNode();
                final String keyValue = (String) keyNode.getProperty(GraphStatics.VALUE_PROPERTY, null);
                final KeyEntity keyEntity = new KeyEntity(keyNode.getId(), keyValue);
                csEntities.get(csEntityNode.getId()).addKeyEntity(keyEntity);
            }
        } else {

            LOG.debug("couldn't determine relative key attribute path for key attribute path '{}' and common attribute path ''",
                    keyAttributePath.toString(), commonAttributePath.toString());
        }
    }
}

private static void determineValueEntities(final GraphDatabaseService graphDB, final AttributePath commonAttributePath,
                                           final ContentSchema contentSchema, final Map<Long, CSEntity> csEntities, final Node... csEntityNodesArray) {

    final AttributePath valueAttributePath = contentSchema.getValueAttributePath();

    final Optional<LinkedList<Attribute>> optionalRelativeValueAttributePath = determineRelativeAttributePath(
            valueAttributePath,
            commonAttributePath);

    if (optionalRelativeValueAttributePath.isPresent()) {

        final LinkedList<Attribute> relativeValueAttributePath = optionalRelativeValueAttributePath.get();

        final Iterable<Path> relativeValuePaths = graphDB.traversalDescription().depthFirst()
                .evaluator(Evaluators.toDepth(relativeValueAttributePath.size())).evaluator(new EntityEvaluator(relativeValueAttributePath))
                .traverse(csEntityNodesArray);

        for (final Path relativeValuePath : relativeValuePaths) {

            final Node valueNode = relativeValuePath.endNode();
            final Node csEntityNode = relativeValuePath.startNode();
            final String valueValue = (String) valueNode.getProperty(GraphStatics.VALUE_PROPERTY, null);
            final Long valueOrder = (Long) relativeValuePath.lastRelationship().getProperty(GraphStatics.ORDER_PROPERTY, null);
            final ValueEntity valueEntity = new ValueEntity(valueNode.getId(), valueValue, valueOrder);
            csEntities.get(csEntityNode.getId()).addValueEntity(valueEntity);
        }
    } else {

        LOG.debug("couldn't determine relative value attribute path for value attribute path '{}' and common attribute path ''",
                valueAttributePath.toString(), commonAttributePath.toString());
    }
}

public void getOrCreateNode(Long start, Long end, GraphDatabaseService db) {
    List<Long> relList = relationshipCache.getIfPresent(start);

    Node startNode = db.getNodeById(start);

    if (relList == null) {
        List<Long> nodeList = new ArrayList<>();
        for(Node endNodes : db.traversalDescription()
                .depthFirst()
                .relationships(withName(relationshipType), Direction.OUTGOING)
                .evaluator(Evaluators.fromDepth(1))
                .evaluator(Evaluators.toDepth(1))
                .traverse(startNode)
                .nodes())
        {
            nodeList.add(endNodes.getId());
        }

        relList = nodeList;
        relationshipCache.put(start, relList);
    }

    if (!relList.contains(end)) {
        Transaction tx = db.beginTx();
        try {
            Node endNode = db.getNodeById(end);
            startNode.createRelationshipTo(endNode, withName(relationshipType));
            tx.success();
        } catch (final Exception e) {
            tx.failure();
        } finally {
            tx.close();
            relList.add(end);
            relationshipCache.put(start, relList);
        }
    }
}

private List<Long> getLongs(Long start, GraphDatabaseService db, List<Long> relList, Node startNode) {
    if (relList == null) {
        relList = new ArrayList<>();
        try (Transaction tx = db.beginTx()) {
            ResourceIterable<Node> nodes = db.traversalDescription()
                    .depthFirst()
                    .relationships(withName(getRelationshipType()), Direction.OUTGOING)
                    .evaluator(Evaluators.fromDepth(1))
                    .evaluator(Evaluators.toDepth(1))
                    .traverse(startNode)
                    .nodes();

            final List<Long> finalRelList = relList;
            nodes.forEach(a -> finalRelList.add(a.getId()));

            tx.success();
        } catch(Exception ex) {
            if(relList.size() == 0) {
                return relList;
            }
        }

        relList = new HashSet<>(relList).stream().map(n -> n).collect(Collectors.toList());

        if (relList.size() > 0) {
            String propertyKey = getRelationshipAggregateKey();
            Integer propertyValue = relList.size();
            try (Transaction tx = db.beginTx()) {
                startNode.setProperty(propertyKey, propertyValue);
                tx.success();
            }
        }

        getRelationshipCache().put(start, relList);
    }
    return relList;
}

public static Map<Long, Integer> getTermFrequencyMapForDocument(GraphDatabaseService db, Long classId)
{
    Map<Long, Integer> termDocumentMatrix;

    String cacheKey = "TERM_DOCUMENT_FREQUENCY_" + classId;

    if(vectorSpaceModelCache.getIfPresent(cacheKey) == null) {
        Node classNode = db.getNodeById(classId);

        termDocumentMatrix = new HashMap<>();

        IteratorUtil.asCollection(db.traversalDescription()
                .depthFirst()
                .relationships(withName("HAS_CLASS"), Direction.INCOMING)
                .evaluator(Evaluators.fromDepth(1))
                .evaluator(Evaluators.toDepth(1))
                .traverse(classNode)).stream()
                .forEach(p ->
                {
                    int matches = (Integer) p.lastRelationship().getProperty("matches");
                    termDocumentMatrix.put(p.endNode().getId(), matches);
                });



        vectorSpaceModelCache.put(cacheKey, termDocumentMatrix);
    }
    else
    {
        termDocumentMatrix =  (Map<Long, Integer>)vectorSpaceModelCache.getIfPresent(cacheKey);
    }

    return termDocumentMatrix;
}

public static double getFeatureMatchDistribution(GraphDatabaseService db, Long patternId)
{
    Transaction tx = db.beginTx();
    Node startNode = db.getNodeById(patternId);

    // Feature match distribution
    List<Double> matches = IteratorUtil.asCollection(db.traversalDescription()
            .depthFirst()
            .relationships(withName("HAS_CLASS"), Direction.OUTGOING)
            .evaluator(Evaluators.fromDepth(1))
            .evaluator(Evaluators.toDepth(1))
            .traverse(startNode)
            .relationships())
            .stream()
            .map(p -> ((Integer)p.getProperty("matches")).doubleValue())
            .collect(Collectors.toList());

    tx.success();
    tx.close();

    double variance = 1.0;

    if(matches.size() > 1) {
        Double[] matchArr = matches.toArray(new Double[matches.size()]);
        // Get the standard deviation
        DescriptiveStatistics ds = new DescriptiveStatistics();
        matches.forEach(m -> ds.addValue(m.doubleValue() / StatUtils.sum(ArrayUtils.toPrimitive(matchArr))));
        variance = ds.getStandardDeviation();
    }

    return variance;
}

public static int getDocumentSizeForFeature(GraphDatabaseService db, Long id)
{
    int documentSize;

    String cacheKey = "DOCUMENT_SIZE_FEATURE_" + id;

    if(vectorSpaceModelCache.getIfPresent(cacheKey) == null) {
        Node startNode = db.getNodeById(id);

        Iterator<Node> classes = db.traversalDescription()
                .depthFirst()
                .relationships(withName("HAS_CLASS"), Direction.OUTGOING)
                .evaluator(Evaluators.fromDepth(1))
                .evaluator(Evaluators.toDepth(1))
                .traverse(startNode)
                .nodes().iterator();

        documentSize = IteratorUtil.count(classes);

        vectorSpaceModelCache.put(cacheKey, documentSize);
    }
    else
    {
        documentSize = (Integer)vectorSpaceModelCache.getIfPresent(cacheKey);
    }

    return documentSize;
}

private static List<LinkedHashMap<String, Object>> getFeaturesForClass(GraphDatabaseService db, Node classNode) {
    List<LinkedHashMap<String, Object>> patternIds = new ArrayList<>();

    for (Path p : db.traversalDescription()
            .depthFirst()
            .relationships(withName("HAS_CLASS"), Direction.INCOMING)
            .evaluator(Evaluators.fromDepth(1))
            .evaluator(Evaluators.toDepth(1))
            .traverse(classNode)) {

        if(getFeatureMatchDistribution(db, p.endNode().getId()) > CONFIDENCE_INTERVAL) {

            LinkedHashMap<String, Object> featureMap = new LinkedHashMap<>();

            if (p.relationships().iterator().hasNext()) {
                featureMap.put("frequency", p.relationships().iterator().next().getProperty("matches"));
            } else {
                featureMap.put("frequency", 0);
            }

            featureMap.put("feature", ((Long) p.endNode().getId()).intValue());

            patternIds.add(featureMap);
        }
    }
    return patternIds;
}

public final List<UserItemRelationship> getRelationships(final String userID,
                                                         final String depth,
                                                         final String sinceTimestamp,
                                                         final String untilTimestamp,
                                                         final boolean removeDuplicates)

        throws ObelixNodeNotFoundException {
    try (Transaction tx = neoDb.beginTx()) {
        List<UserItemRelationship> userItemRelationships = new LinkedList<>();
        Node user = getUserNode(neoDb, userID);

        for (Path path : neoDb.traversalDescription()
                .breadthFirst()
                .expand(new TimeStampExpander(sinceTimestamp, untilTimestamp, depth))
                .evaluator(Evaluators.toDepth(Integer.parseInt(depth)))
                .uniqueness(Uniqueness.RELATIONSHIP_GLOBAL)
                .traverse(user)) {

            if (path.lastRelationship() != null) {

                String itemid = path.lastRelationship()
                        .getEndNode().getProperty("node_id").toString();

                userItemRelationships.add(new UserItemRelationship(itemid, path.length()));
            }
        }
        tx.success();
        return userItemRelationships;
    }
}

public Traverser getTraverser(String task, Direction direction){
    TraversalDescription td = Traversal.description()
            .breadthFirst()
            .relationships( TaskRelations.DEPENDS_ON, direction )
            .evaluator(Evaluators.excludeStartPosition());
    return td.traverse(getNode(task));
}

@GET
@Path("/cc/{name}")
public String getConnectedComponentsCount(@PathParam("name") String name, @Context GraphDatabaseService db) throws IOException {
    int CCid = 0;
    for ( Node n : GlobalGraphOperations.at( db ).getAllNodes() ) {
        if(!n.hasProperty("CCId")) {
            Transaction tx = db.beginTx();
            try {
                Traverser traverser = Traversal.description()
                        .breadthFirst()
                        .relationships(DynamicRelationshipType.withName(name), Direction.BOTH)
                        .evaluator(Evaluators.excludeStartPosition())
                        .uniqueness(Uniqueness.NODE_GLOBAL)
                        .traverse(n);
                int currentCCid = CCid;
                CCid++;
                n.setProperty("CCId", currentCCid);
                for ( org.neo4j.graphdb.Path p : traverser )
                {
                    p.endNode().setProperty("CCId", currentCCid);
                }
                tx.success();
            }
            catch ( Exception e )
            {
                tx.failure();
            }
            finally
            {
                tx.finish();
            }
        }
    }
    return String.valueOf(CCid);
}

private Traverser getFriends(final Node person) {
    TraversalDescription td = graphDb.traversalDescription().breadthFirst().relationships(RelTypes.KNOWS, Direction.OUTGOING)
            .evaluator(Evaluators.excludeStartPosition());
    return td.traverse(person);
}

private Traverser findHackers(final Node startNode) {
    TraversalDescription td = graphDb.traversalDescription().breadthFirst().relationships(RelTypes.CODED_BY, Direction.OUTGOING)
            .relationships(RelTypes.KNOWS, Direction.OUTGOING).evaluator(Evaluators.includeWhereLastRelationshipTypeIs(RelTypes.CODED_BY));
    return td.traverse(startNode);
}

public static Path exportPartitionToHDFSParallel(GraphDatabaseService db, Node partitionNode, PartitionDescription partitionDescription) throws IOException, URISyntaxException {
    FileSystem fs = FileUtil.getHadoopFileSystem();
    Path pt = new Path(ConfigurationLoader.getInstance().getHadoopHdfsUri() + EDGE_LIST_RELATIVE_FILE_PATH.replace("{job_id}", partitionDescription.getPartitionId().toString()));
    BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt)));

    Integer reportBlockSize = 20000;

    Transaction tx = db.beginTx();

    ResourceIterable<Node> nodes = db.traversalDescription()
            .depthFirst()
            .relationships(withName(partitionDescription.getGroupRelationship()), Direction.OUTGOING)
            .evaluator(Evaluators.toDepth(1))
            .traverse(partitionNode)
            .nodes();

    if (nodes.iterator().hasNext()) {

        br.write("# Adacency list" + "\n");

        List<Spliterator<Node>> spliteratorList = new ArrayList<>();
        boolean hasSpliterator = true;
        Spliterator<Node> nodeSpliterator = nodes.spliterator();

        while (hasSpliterator) {
            Spliterator<Node> localSpliterator = nodeSpliterator.trySplit();
            hasSpliterator = localSpliterator != null;
            if (hasSpliterator)
                spliteratorList.add(localSpliterator);
        }


        counter = 0;

        if (spliteratorList.size() > 4) {
            // Fork join
            ParallelWriter parallelWriter = new ParallelWriter<Node>(spliteratorList.toArray(new Spliterator[spliteratorList.size()]),
                    new GraphWriter(0, spliteratorList.size(), br, spliteratorList.size(), reportBlockSize, db, partitionDescription.getTargetRelationship()));
            ForkJoinPool pool = new ForkJoinPool();
            pool.invoke(parallelWriter);
        } else {
            // Sequential
            spliteratorList.forEach(sl -> sl.forEachRemaining(n -> {
                try {
                    writeBlockForNode(n, db, br, reportBlockSize, partitionDescription.getTargetRelationship());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }));
        }

        System.out.println("Mazerunner Partition Export Status: " + MessageFormat.format("{0,number,#.##%}", 1.0));

        br.flush();
        br.close();

        tx.success();
        tx.close();

        return pt;
    } else {
        return null;
    }
}

/**
 * Applies the result of the analysis as a partitioned value connecting the partition node to the target node.
 *
 * @param line             The line from the HDFS text file containing the analysis results.
 * @param db               The Neo4j graph database context.
 * @param reportBlockSize  The report block size for progress status.
 * @param processorMessage The processor message containing the description of the analysis.
 * @param partitionNode    The partition node that will be the source node for creating partitioned relationships to the target node.
 */
public static void updatePartitionBlockForRow(String line, GraphDatabaseService db, int reportBlockSize, ProcessorMessage processorMessage, Node partitionNode) {
    if (line != null && !line.startsWith("#")) {
        String[] rowVal = line.split("\\s");
        Long nodeId = Long.parseLong(rowVal[0]);
        Double weight = Double.parseDouble(rowVal[1]);
        Node targetNode = db.getNodeById(nodeId);

        Iterator<Relationship> rels = db.traversalDescription()
                .depthFirst()
                .relationships(withName(processorMessage.getAnalysis()), Direction.INCOMING)
                .evaluator(Evaluators.fromDepth(1))
                .evaluator(Evaluators.toDepth(1))
                .traverse(targetNode)
                .relationships()
                .iterator();

        // Get the relationship to update
        Relationship updateRel = null;

        // Scan the relationships
        while (rels.hasNext() && updateRel == null) {
            Relationship currentRel = rels.next();
            if (currentRel.getStartNode().getId() == partitionNode.getId())
                updateRel = currentRel;
        }

        // Create or update the relationship for the analysis on the partition
        if (updateRel != null) {
            updateRel.setProperty("value", weight);
        } else {
            Relationship newRel = partitionNode.createRelationshipTo(targetNode, withName(processorMessage.getAnalysis()));
            newRel.setProperty("value", weight);
        }

        Writer.updateCounter++;
        if (Writer.updateCounter % reportBlockSize == 0) {
            System.out.println("Nodes updated: " + Writer.updateCounter);
        }
    }
}

public static void updateCollaborativeFilteringForRow(String line, GraphDatabaseService db, int reportBlockSize) {
    if (line != null && !line.startsWith("#")) {
        String[] rowVal = line.split(",");
        Long from = Long.parseLong(rowVal[0]);
        Long to = Long.parseLong(rowVal[1]);
        Integer rank = Integer.parseInt(rowVal[2]);
        Node fromNode = db.getNodeById(from);

        final String recommendation = "RECOMMENDATION";

        Iterator<Relationship> rels = db.traversalDescription()
                .depthFirst()
                .relationships(withName(recommendation), Direction.INCOMING)
                .evaluator(Evaluators.fromDepth(1))
                .evaluator(Evaluators.toDepth(1))
                .traverse(fromNode)
                .relationships()
                .iterator();

        Relationship updateRel = null;

        // Scan the relationships
        while (rels.hasNext()) {
            Relationship currentRel = rels.next();
            if(currentRel.hasProperty("rank") && Objects.equals(currentRel.getProperty("rank"), rank)) {
                if(currentRel.getEndNode().getId() != to) {
                    currentRel.delete();
                } else updateRel = currentRel;

                break;
            }
        }

        // Create or update the relationship for the analysis on the partition
        if (updateRel == null) {
            Relationship newRel = fromNode.createRelationshipTo(db.getNodeById(to), withName(recommendation));
            newRel.setProperty("rank", rank);
        }

        Writer.updateCounter++;
        if (Writer.updateCounter % reportBlockSize == 0) {
            System.out.println("Nodes updated: " + Writer.updateCounter);
        }

    }
}

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();

    // Get all nodes in the graph
    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());

    //System.out.println(nodes.spliterator().trySplit().estimateSize());

    // Fork join

    nodes.iterator().forEachRemaining(n -> {
        // Filter nodes by all paths connected by the relationship type described in the configuration properties
        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;
}

public TraversalDescriptionImpl()
{
    this( StandardExpander.DEFAULT, Uniqueness.NODE_GLOBAL, null,
            Evaluators.all(), Traversal.preorderDepthFirst() );
}

public TraversalDescription prune( PruneEvaluator pruning )
{
    return evaluator( pruning == PruneEvaluator.NONE ? Evaluators.all() :
            new WrappedPruneEvaluator( pruning ) );
}

public static List<CostFunction> getCostFunctionsForNode(Long id, EmbeddedGraphDatabase database) {

        List<CostFunction> costFunctions = new ArrayList<CostFunction>();
        boolean transactionAllreadyRunning = false;
        try {
            transactionAllreadyRunning = (database.getTxManager().getStatus() == Status.STATUS_ACTIVE);
        } catch (SystemException ex) {
            log.error(ex.getMessage(), ex);
        }
        Transaction tx = (transactionAllreadyRunning) ? null : database.beginTx();

        try {

            Node parentNode = database.getNodeById(id);

            if (parentNode == null) {
                return costFunctions;
            }

            //search from this node with ID=id the target nodes for which it has a HAS_COST_FUNCTION relationship
            TraversalDescription description = Traversal.traversal()
                    .evaluator(Evaluators.excludeStartPosition())
                    .relationships(ServiceUnitRelationship.hasCostFunction, Direction.OUTGOING)
                    .uniqueness(Uniqueness.NODE_PATH);
            Traverser traverser = description.traverse(parentNode);
            for (Path path : traverser) {

                Node node = path.endNode();
                CostFunction costFunction = new CostFunction();
                costFunction.setId(node.getId());

                if (node.hasProperty(KEY)) {
                    costFunction.setName(node.getProperty(KEY).toString());
                } else {
                    log.warn("Retrieved CostFunction " + node + " has no " + KEY);
                }

                if (node.hasProperty(UUID)) {
                    costFunction.setUuid(java.util.UUID.fromString(node.getProperty(UUID).toString()));
                } else {
                    log.warn("Retrieved CloudProvider " + costFunction + " has no " + UUID);
                }

                //carefull. this can lead to infinite recursion (is still a graph. maybe improve later)
                costFunction.getAppliedIfServiceInstanceUses().addAll(getAppliedInConjunctionWithEntities(node.getId(), database));
                //need to also retrieve Resurce and Quality

                costFunction.getCostElements().addAll(CostElementDAO.getCostElementPropertiesForNode(node.getId(), database));

                if (costFunction != null) {
                    //hack. if the costFunction has allready been added (equals is done on the DB Node),
                    //this means ServiceUnit has elasticity capability on it, and the old is also removed
                    if (costFunctions.contains(costFunction)) {
                        costFunctions.remove(costFunction);
                    } else {
                        costFunctions.add(costFunction);
                    }
                }
            }
            if (!transactionAllreadyRunning) {
                tx.success();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            e.printStackTrace();
        } finally {
            if (!transactionAllreadyRunning) {
                tx.finish();
            }
        }

        return costFunctions;
    }

/**
 * Counts how many elasticity characteristic nodes point to it with a
 * "elasticityCapabilityFor" relationship
 *
 * @param id
 * @param database
 * @return sum of incoming MANDATORY_ASSOCIATION and OPTIONAL_ASSOCIATION
 * elasticity capabilities if returns -1, means error encountered. otherwise
 * the result is always >= 0
 */
public static int getElasticityDependency(long id, EmbeddedGraphDatabase database) {

    CloudOfferedService elTarget = null;
    int incomingPaths = 0;
    boolean transactionAllreadyRunning = false;
    try {
        transactionAllreadyRunning = (database.getTxManager().getStatus() == Status.STATUS_ACTIVE);
    } catch (SystemException ex) {
        log.error(ex.getMessage(), ex);
    }
    Transaction tx = (transactionAllreadyRunning) ? null : database.beginTx();

    try {
        Node parentNode = database.getNodeById(id);

        if (parentNode == null) {
            log.error("Node with id " + id + " was not found");
            return 0;
        }

        TraversalDescription description = Traversal.traversal()
                .evaluator(Evaluators.excludeStartPosition())
                .relationships(ServiceUnitRelationship.hasElasticityCapability, Direction.OUTGOING)
                .uniqueness(Uniqueness.NODE_PATH);
        Traverser traverser = description.traverse(parentNode);

        //for each incoming path, if is MANDATORY_ASSOCIATION decrease the in
        for (Path path : traverser) {
            incomingPaths++;
        }
        if (!transactionAllreadyRunning) {
            if (!transactionAllreadyRunning) {
                tx.success();
            }
        }
    } catch (Exception e) {
        log.error(e.getMessage(), e);

        e.printStackTrace();
    } finally {
        if (!transactionAllreadyRunning) {
            tx.finish();
        }
    }
    return incomingPaths;
}

public static CloudOfferedService getByID(Long nodeID, EmbeddedGraphDatabase database) {

        CloudOfferedService serviceUnit = null;
        boolean transactionAllreadyRunning = false;
        try {
            transactionAllreadyRunning = (database.getTxManager().getStatus() == Status.STATUS_ACTIVE);
        } catch (SystemException ex) {
            log.error(ex.getMessage(), ex);
        }
        Transaction tx = (transactionAllreadyRunning) ? null : database.beginTx();

        try {
            Node parentNode = database.getNodeById(nodeID);

            if (parentNode == null) {
                return serviceUnit;
            }

            TraversalDescription description = Traversal.traversal()
                    .evaluator(Evaluators.excludeStartPosition())
                    .relationships(ServiceUnitRelationship.elasticityCapabilityFor, Direction.OUTGOING)
                    .uniqueness(Uniqueness.NODE_PATH);
            Traverser traverser = description.traverse(parentNode);
            for (Path path : traverser) {

                Node node = path.endNode();
                if (!node.hasLabel(LABEL)) {
                    continue;
                }
                serviceUnit = new CloudOfferedService();
                serviceUnit.setId(node.getId());

                if (node.hasProperty(KEY)) {
                    serviceUnit.setName(node.getProperty(KEY).toString());
                } else {
                    log.warn("Retrieved serviceUnit " + nodeID + " has no " + KEY);
                }

                if (node.hasProperty(CATEGORY)) {
                    serviceUnit.setCategory(node.getProperty(CATEGORY).toString());
                } else {
                    log.warn("Retrieved serviceUnit " + nodeID + " has no " + CATEGORY);
                }

                if (node.hasProperty(SUBCATEGORY)) {
                    serviceUnit.setSubcategory(node.getProperty(SUBCATEGORY).toString());
                } else {
                    log.warn("Retrieved serviceUnit " + nodeID + " has no " + SUBCATEGORY);
                }
                if (node.hasProperty(UUID)) {
                    serviceUnit.setUuid(java.util.UUID.fromString(node.getProperty(UUID).toString()));
                } else {
                    log.warn("Retrieved CloudProvider " + serviceUnit + " has no " + UUID);
                }

                serviceUnit.getResourceProperties().addAll(ResourceDAO.getResourcePropertiesForNode(node.getId(), database));
                serviceUnit.getQualityProperties().addAll(QualityDAO.getQualityPropertiesForNode(node.getId(), database));
                serviceUnit.getCostFunctions().addAll(CostFunctionDAO.getCostFunctionsForNode(node.getId(), database));
                serviceUnit.getElasticityCapabilities().addAll(ElasticityCapabilityDAO.getELasticityCapabilitiesForNode(node.getId(), database));
                //serviceUnit.setElasticityQuantification(getElasticityDependency(node.getId(), database));

            }
            if (!transactionAllreadyRunning) {
                if (!transactionAllreadyRunning) {
                    tx.success();
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);

            e.printStackTrace();
        } finally {
            if (!transactionAllreadyRunning) {
                tx.finish();
            }
        }

        return serviceUnit;
    }

public static Collection<CSEntity> getCSEntities(final GraphDatabaseService graphDB, final String prefixedResourceURI,
                                                 final AttributePath commonPrefixedAttributePath, final ContentSchema prefixedContentSchema)
        throws DMPGraphException {

    final Map<Long, CSEntity> csEntities = new LinkedHashMap<>();

    try (final Transaction tx = graphDB.beginTx()) {

        final Node resourceNode = getResourceNode(graphDB, prefixedResourceURI);

        // determine CS entity nodes
        final ResourceIterable<Node> csEntityNodes = graphDB.traversalDescription().breadthFirst()
                .evaluator(Evaluators.toDepth(commonPrefixedAttributePath.getAttributes().size()))
                .evaluator(new EntityEvaluator(commonPrefixedAttributePath.getAttributes()))
                .traverse(resourceNode).nodes();

        if (csEntityNodes == null) {

            tx.success();

            return null;
        }

        for (final Node node : csEntityNodes) {

            final CSEntity csEntity = new CSEntity(node.getId());
            csEntities.put(csEntity.getNodeId(), csEntity);
        }

        final ArrayList<Node> csEntityNodesList = Lists.newArrayList(csEntityNodes);
        final Node[] csEntityNodesArray = new Node[csEntityNodesList.size()];
        csEntityNodesList.toArray(csEntityNodesArray);

        if (prefixedContentSchema.getKeyAttributePaths() != null) {

            // determine key entities
            determineKeyEntities(graphDB, commonPrefixedAttributePath, prefixedContentSchema, csEntities, csEntityNodesArray);
        }

        if (prefixedContentSchema.getValueAttributePath() != null) {

            // determine value entities
            determineValueEntities(graphDB, commonPrefixedAttributePath, prefixedContentSchema, csEntities, csEntityNodesArray);
        }

        tx.success();
    } catch (final Exception e) {

        final String message = "couldn't determine cs entities successfully";

        GraphDBUtil.LOG.error(message, e);

        throw new DMPGraphException(message);
    }

    final Collection<CSEntity> csEntitiesCollection = csEntities.values();

    // determine cs entity order
    determineCSEntityOrder(csEntitiesCollection);

    return csEntitiesCollection;
}