@Test public void testEmptyEdgeRead() throws Exception { String csvPath = getFilePath("/data/csv/input_indexed_no_edges"); String gdlPath = getFilePath("/data/csv/expected/expected_no_edges.gdl"); DataSource dataSource = new IndexedCSVDataSource(csvPath, getConfig()); GraphCollection input = dataSource.getGraphCollection(); GraphCollection expected = getLoaderFromFile(gdlPath) .getGraphCollectionByVariables("expected1", "expected2"); collectAndAssertTrue(input.equalsByGraphData(expected)); }
@Override public GraphCollection getGraphCollection() throws IOException { CSVMetaDataSource source = new CSVMetaDataSource(); CSVMetaData metaData = source.readLocal(getMetaDataPath(), hdfsConfig); DataSet<Tuple3<String, String, String>> metaDataBroadcast = source.readDistributed(getMetaDataPath(), getConfig()); ExecutionEnvironment env = getConfig().getExecutionEnvironment(); GraphCollectionFactory factory = getConfig().getGraphCollectionFactory(); Map<String, DataSet<EPGMGraphHead>> graphHeads = metaData.getGraphLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getGraphHeadCSVPath(label)) .map(new CSVLineToGraphHead(factory.getGraphHeadFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(graphHead -> graphHead.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMVertex>> vertices = metaData.getVertexLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getVertexCSVPath(label)) .map(new CSVLineToVertex(factory.getVertexFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(vertex -> vertex.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMEdge>> edges = metaData.getEdgeLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getEdgeCSVPath(label)) .map(new CSVLineToEdge(factory.getEdgeFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(edge -> edge.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); return factory.fromIndexedDataSets(graphHeads, vertices, edges); }
IndexedCSVDataSource extends CSVBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { CSVMetaDataSource source = new CSVMetaDataSource(); CSVMetaData metaData = source.readLocal(getMetaDataPath(), hdfsConfig); DataSet<Tuple3<String, String, String>> metaDataBroadcast = source.readDistributed(getMetaDataPath(), getConfig()); ExecutionEnvironment env = getConfig().getExecutionEnvironment(); GraphCollectionFactory factory = getConfig().getGraphCollectionFactory(); Map<String, DataSet<EPGMGraphHead>> graphHeads = metaData.getGraphLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getGraphHeadCSVPath(label)) .map(new CSVLineToGraphHead(factory.getGraphHeadFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(graphHead -> graphHead.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMVertex>> vertices = metaData.getVertexLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getVertexCSVPath(label)) .map(new CSVLineToVertex(factory.getVertexFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(vertex -> vertex.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMEdge>> edges = metaData.getEdgeLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getEdgeCSVPath(label)) .map(new CSVLineToEdge(factory.getEdgeFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(edge -> edge.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); return factory.fromIndexedDataSets(graphHeads, vertices, edges); } }
IndexedCSVDataSource extends CSVBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { CSVMetaDataSource source = new CSVMetaDataSource(); CSVMetaData metaData = source.readLocal(getMetaDataPath(), hdfsConfig); DataSet<Tuple3<String, String, String>> metaDataBroadcast = source.readDistributed(getMetaDataPath(), getConfig()); ExecutionEnvironment env = getConfig().getExecutionEnvironment(); GraphCollectionFactory factory = getConfig().getGraphCollectionFactory(); Map<String, DataSet<EPGMGraphHead>> graphHeads = metaData.getGraphLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getGraphHeadCSVPath(label)) .map(new CSVLineToGraphHead(factory.getGraphHeadFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(graphHead -> graphHead.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMVertex>> vertices = metaData.getVertexLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getVertexCSVPath(label)) .map(new CSVLineToVertex(factory.getVertexFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(vertex -> vertex.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMEdge>> edges = metaData.getEdgeLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getEdgeCSVPath(label)) .map(new CSVLineToEdge(factory.getEdgeFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(edge -> edge.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); return factory.fromIndexedDataSets(graphHeads, vertices, edges); } IndexedCSVDataSource(String csvPath, GradoopFlinkConfig config);  IndexedCSVDataSource(String csvPath, GradoopFlinkConfig conf, Configuration hdfsConf); }
IndexedCSVDataSource extends CSVBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { CSVMetaDataSource source = new CSVMetaDataSource(); CSVMetaData metaData = source.readLocal(getMetaDataPath(), hdfsConfig); DataSet<Tuple3<String, String, String>> metaDataBroadcast = source.readDistributed(getMetaDataPath(), getConfig()); ExecutionEnvironment env = getConfig().getExecutionEnvironment(); GraphCollectionFactory factory = getConfig().getGraphCollectionFactory(); Map<String, DataSet<EPGMGraphHead>> graphHeads = metaData.getGraphLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getGraphHeadCSVPath(label)) .map(new CSVLineToGraphHead(factory.getGraphHeadFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(graphHead -> graphHead.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMVertex>> vertices = metaData.getVertexLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getVertexCSVPath(label)) .map(new CSVLineToVertex(factory.getVertexFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(vertex -> vertex.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMEdge>> edges = metaData.getEdgeLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getEdgeCSVPath(label)) .map(new CSVLineToEdge(factory.getEdgeFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(edge -> edge.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); return factory.fromIndexedDataSets(graphHeads, vertices, edges); } IndexedCSVDataSource(String csvPath, GradoopFlinkConfig config);  IndexedCSVDataSource(String csvPath, GradoopFlinkConfig conf, Configuration hdfsConf); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection(); }
IndexedCSVDataSource extends CSVBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { CSVMetaDataSource source = new CSVMetaDataSource(); CSVMetaData metaData = source.readLocal(getMetaDataPath(), hdfsConfig); DataSet<Tuple3<String, String, String>> metaDataBroadcast = source.readDistributed(getMetaDataPath(), getConfig()); ExecutionEnvironment env = getConfig().getExecutionEnvironment(); GraphCollectionFactory factory = getConfig().getGraphCollectionFactory(); Map<String, DataSet<EPGMGraphHead>> graphHeads = metaData.getGraphLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getGraphHeadCSVPath(label)) .map(new CSVLineToGraphHead(factory.getGraphHeadFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(graphHead -> graphHead.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMVertex>> vertices = metaData.getVertexLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getVertexCSVPath(label)) .map(new CSVLineToVertex(factory.getVertexFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(vertex -> vertex.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); Map<String, DataSet<EPGMEdge>> edges = metaData.getEdgeLabels().stream() .map(label -> Tuple2.of(label, env.readTextFile(getEdgeCSVPath(label)) .map(new CSVLineToEdge(factory.getEdgeFactory())) .withBroadcastSet(metaDataBroadcast, BC_METADATA) .filter(edge -> edge.getLabel().equals(label)))) .collect(Collectors.toMap(t -> t.f0, t -> t.f1)); return factory.fromIndexedDataSets(graphHeads, vertices, edges); } IndexedCSVDataSource(String csvPath, GradoopFlinkConfig config);  IndexedCSVDataSource(String csvPath, GradoopFlinkConfig conf, Configuration hdfsConf); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection();  }