@Test public void testReadWithLabelAndProperties() throws Exception { ExecutionEnvironment env = getExecutionEnvironment(); Map<String, Object> properties = Maps.newHashMap(); properties.put("foo", 42); DataSet<ImportVertex<Long>> importVertices = env.fromElements( new ImportVertex<>(0L, "A", Properties.createFromMap(properties)), new ImportVertex<>(1L, "B", Properties.createFromMap(properties))); DataSet<ImportEdge<Long>> importEdges = env.fromElements( new ImportEdge<>(0L, 0L, 1L, "a", Properties.createFromMap(properties)), new ImportEdge<>(1L, 1L, 0L, "b", Properties.createFromMap(properties))); LogicalGraph expected = getLoaderFromString("expected[" + "(a:A {foo : 42})" + "(b:B {foo : 42})" + "(a)-[:a {foo : 42}]->(b)-[:b {foo : 42}]->(a)" + "]").getLogicalGraphByVariable("expected"); GraphDataSource<Long> dataSource = new GraphDataSource<>( importVertices, importEdges, getConfig()); LogicalGraph output = dataSource.getLogicalGraph(); collectAndAssertTrue(output.equalsByElementData(expected)); }
@Override public LogicalGraph getLogicalGraph() { TypeInformation<K> externalIdType = ((TupleTypeInfo<?>) importVertices .getType()).getTypeAt(0); LogicalGraphFactory factory = config.getLogicalGraphFactory(); DataSet<Tuple3<K, GradoopId, EPGMVertex>> vertexTriples = importVertices .map(new InitVertex<K>( factory.getVertexFactory(), lineagePropertyKey, externalIdType)); DataSet<EPGMVertex> epgmVertices = vertexTriples .map(new Value2Of3<K, GradoopId, EPGMVertex>()); DataSet<Tuple2<K, GradoopId>> vertexIdPair = vertexTriples .map(new Project3To0And1<K, GradoopId, EPGMVertex>()); DataSet<EPGMEdge> epgmEdges = importEdges .join(vertexIdPair) .where(1).equalTo(0) .with(new InitEdge<K>( factory.getEdgeFactory(), lineagePropertyKey, externalIdType)) .join(vertexIdPair) .where(0).equalTo(0) .with(new UpdateEdge<EPGMEdge, K>()); return factory.fromDataSets(epgmVertices, epgmEdges); }
GraphDataSource implements DataSource { @Override public LogicalGraph getLogicalGraph() { TypeInformation<K> externalIdType = ((TupleTypeInfo<?>) importVertices .getType()).getTypeAt(0); LogicalGraphFactory factory = config.getLogicalGraphFactory(); DataSet<Tuple3<K, GradoopId, EPGMVertex>> vertexTriples = importVertices .map(new InitVertex<K>( factory.getVertexFactory(), lineagePropertyKey, externalIdType)); DataSet<EPGMVertex> epgmVertices = vertexTriples .map(new Value2Of3<K, GradoopId, EPGMVertex>()); DataSet<Tuple2<K, GradoopId>> vertexIdPair = vertexTriples .map(new Project3To0And1<K, GradoopId, EPGMVertex>()); DataSet<EPGMEdge> epgmEdges = importEdges .join(vertexIdPair) .where(1).equalTo(0) .with(new InitEdge<K>( factory.getEdgeFactory(), lineagePropertyKey, externalIdType)) .join(vertexIdPair) .where(0).equalTo(0) .with(new UpdateEdge<EPGMEdge, K>()); return factory.fromDataSets(epgmVertices, epgmEdges); } }
GraphDataSource implements DataSource { @Override public LogicalGraph getLogicalGraph() { TypeInformation<K> externalIdType = ((TupleTypeInfo<?>) importVertices .getType()).getTypeAt(0); LogicalGraphFactory factory = config.getLogicalGraphFactory(); DataSet<Tuple3<K, GradoopId, EPGMVertex>> vertexTriples = importVertices .map(new InitVertex<K>( factory.getVertexFactory(), lineagePropertyKey, externalIdType)); DataSet<EPGMVertex> epgmVertices = vertexTriples .map(new Value2Of3<K, GradoopId, EPGMVertex>()); DataSet<Tuple2<K, GradoopId>> vertexIdPair = vertexTriples .map(new Project3To0And1<K, GradoopId, EPGMVertex>()); DataSet<EPGMEdge> epgmEdges = importEdges .join(vertexIdPair) .where(1).equalTo(0) .with(new InitEdge<K>( factory.getEdgeFactory(), lineagePropertyKey, externalIdType)) .join(vertexIdPair) .where(0).equalTo(0) .with(new UpdateEdge<EPGMEdge, K>()); return factory.fromDataSets(epgmVertices, epgmEdges); } GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, GradoopFlinkConfig config);  GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, String lineagePropertyKey,
    GradoopFlinkConfig config); }
GraphDataSource implements DataSource { @Override public LogicalGraph getLogicalGraph() { TypeInformation<K> externalIdType = ((TupleTypeInfo<?>) importVertices .getType()).getTypeAt(0); LogicalGraphFactory factory = config.getLogicalGraphFactory(); DataSet<Tuple3<K, GradoopId, EPGMVertex>> vertexTriples = importVertices .map(new InitVertex<K>( factory.getVertexFactory(), lineagePropertyKey, externalIdType)); DataSet<EPGMVertex> epgmVertices = vertexTriples .map(new Value2Of3<K, GradoopId, EPGMVertex>()); DataSet<Tuple2<K, GradoopId>> vertexIdPair = vertexTriples .map(new Project3To0And1<K, GradoopId, EPGMVertex>()); DataSet<EPGMEdge> epgmEdges = importEdges .join(vertexIdPair) .where(1).equalTo(0) .with(new InitEdge<K>( factory.getEdgeFactory(), lineagePropertyKey, externalIdType)) .join(vertexIdPair) .where(0).equalTo(0) .with(new UpdateEdge<EPGMEdge, K>()); return factory.fromDataSets(epgmVertices, epgmEdges); } GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, GradoopFlinkConfig config);  GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, String lineagePropertyKey,
    GradoopFlinkConfig config); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection(); }
GraphDataSource implements DataSource { @Override public LogicalGraph getLogicalGraph() { TypeInformation<K> externalIdType = ((TupleTypeInfo<?>) importVertices .getType()).getTypeAt(0); LogicalGraphFactory factory = config.getLogicalGraphFactory(); DataSet<Tuple3<K, GradoopId, EPGMVertex>> vertexTriples = importVertices .map(new InitVertex<K>( factory.getVertexFactory(), lineagePropertyKey, externalIdType)); DataSet<EPGMVertex> epgmVertices = vertexTriples .map(new Value2Of3<K, GradoopId, EPGMVertex>()); DataSet<Tuple2<K, GradoopId>> vertexIdPair = vertexTriples .map(new Project3To0And1<K, GradoopId, EPGMVertex>()); DataSet<EPGMEdge> epgmEdges = importEdges .join(vertexIdPair) .where(1).equalTo(0) .with(new InitEdge<K>( factory.getEdgeFactory(), lineagePropertyKey, externalIdType)) .join(vertexIdPair) .where(0).equalTo(0) .with(new UpdateEdge<EPGMEdge, K>()); return factory.fromDataSets(epgmVertices, epgmEdges); } GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, GradoopFlinkConfig config);  GraphDataSource(DataSet<ImportVertex<K>> importVertices,
    DataSet<ImportEdge<K>> importEdges, String lineagePropertyKey,
    GradoopFlinkConfig config); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection();  }