@Test public void testReadWithDictionary() throws Exception { String tlfFile = getFilePath("/data/tlf/io_test.tlf"); String tlfVertexDictionaryFile = getFilePath("/data/tlf/io_test_vertex_dictionary.tlf"); String tlfEdgeDictionaryFile = getFilePath("/data/tlf/io_test_edge_dictionary.tlf"); DataSource dataSource = new TLFDataSource(tlfFile, tlfVertexDictionaryFile, tlfEdgeDictionaryFile, getConfig()); DataSet<GraphTransaction> transactions = dataSource.getGraphCollection().getGraphTransactions(); String asciiGraphs = "" + "g1[(v1:A)-[:a]->(v2:B)-[:b]->(v1)]" + "g2[(v1:A)-[:a]->(v2:B)<-[:b]-(v1)]"; FlinkAsciiGraphLoader loader = getLoaderFromString(asciiGraphs); collectAndAssertTrue( loader.getGraphCollectionByVariables("g1", "g2").equalsByGraphData( getConfig().getGraphCollectionFactory().fromTransactions(transactions) ) ); }
@Override public GraphCollection getGraphCollection() throws IOException { DataSet<GraphTransaction> transactions; ExecutionEnvironment env = getConfig().getExecutionEnvironment(); DataSet<Tuple2<LongWritable, Text>> input = null; input = env.createInput(HadoopInputs.readHadoopFile( new TLFInputFormat(), LongWritable.class, Text.class, getTLFPath())); assert input != null; transactions = input.map(new GraphTransactionFromText(getConfig())); if (hasVertexDictionary()) { transactions = transactions .map(new VertexLabelDecoder()) .withBroadcastSet( getVertexDictionary(), TLFConstants.VERTEX_DICTIONARY); } if (hasEdgeDictionary()) { transactions = transactions .map(new EdgeLabelDecoder()) .withBroadcastSet( getEdgeDictionary(), TLFConstants.EDGE_DICTIONARY); } return getConfig().getGraphCollectionFactory().fromTransactions(transactions); }
TLFDataSource extends TLFBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { DataSet<GraphTransaction> transactions; ExecutionEnvironment env = getConfig().getExecutionEnvironment(); DataSet<Tuple2<LongWritable, Text>> input = null; input = env.createInput(HadoopInputs.readHadoopFile( new TLFInputFormat(), LongWritable.class, Text.class, getTLFPath())); assert input != null; transactions = input.map(new GraphTransactionFromText(getConfig())); if (hasVertexDictionary()) { transactions = transactions .map(new VertexLabelDecoder()) .withBroadcastSet( getVertexDictionary(), TLFConstants.VERTEX_DICTIONARY); } if (hasEdgeDictionary()) { transactions = transactions .map(new EdgeLabelDecoder()) .withBroadcastSet( getEdgeDictionary(), TLFConstants.EDGE_DICTIONARY); } return getConfig().getGraphCollectionFactory().fromTransactions(transactions); } }
TLFDataSource extends TLFBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { DataSet<GraphTransaction> transactions; ExecutionEnvironment env = getConfig().getExecutionEnvironment(); DataSet<Tuple2<LongWritable, Text>> input = null; input = env.createInput(HadoopInputs.readHadoopFile( new TLFInputFormat(), LongWritable.class, Text.class, getTLFPath())); assert input != null; transactions = input.map(new GraphTransactionFromText(getConfig())); if (hasVertexDictionary()) { transactions = transactions .map(new VertexLabelDecoder()) .withBroadcastSet( getVertexDictionary(), TLFConstants.VERTEX_DICTIONARY); } if (hasEdgeDictionary()) { transactions = transactions .map(new EdgeLabelDecoder()) .withBroadcastSet( getEdgeDictionary(), TLFConstants.EDGE_DICTIONARY); } return getConfig().getGraphCollectionFactory().fromTransactions(transactions); } TLFDataSource(String tlfPath, GradoopFlinkConfig config);  TLFDataSource(String tlfPath, String tlfVertexDictionaryPath,
    String tlfEdgeDictionaryPath, GradoopFlinkConfig config); }
TLFDataSource extends TLFBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { DataSet<GraphTransaction> transactions; ExecutionEnvironment env = getConfig().getExecutionEnvironment(); DataSet<Tuple2<LongWritable, Text>> input = null; input = env.createInput(HadoopInputs.readHadoopFile( new TLFInputFormat(), LongWritable.class, Text.class, getTLFPath())); assert input != null; transactions = input.map(new GraphTransactionFromText(getConfig())); if (hasVertexDictionary()) { transactions = transactions .map(new VertexLabelDecoder()) .withBroadcastSet( getVertexDictionary(), TLFConstants.VERTEX_DICTIONARY); } if (hasEdgeDictionary()) { transactions = transactions .map(new EdgeLabelDecoder()) .withBroadcastSet( getEdgeDictionary(), TLFConstants.EDGE_DICTIONARY); } return getConfig().getGraphCollectionFactory().fromTransactions(transactions); } TLFDataSource(String tlfPath, GradoopFlinkConfig config);  TLFDataSource(String tlfPath, String tlfVertexDictionaryPath,
    String tlfEdgeDictionaryPath, GradoopFlinkConfig config); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection(); }
TLFDataSource extends TLFBase implements DataSource { @Override public GraphCollection getGraphCollection() throws IOException { DataSet<GraphTransaction> transactions; ExecutionEnvironment env = getConfig().getExecutionEnvironment(); DataSet<Tuple2<LongWritable, Text>> input = null; input = env.createInput(HadoopInputs.readHadoopFile( new TLFInputFormat(), LongWritable.class, Text.class, getTLFPath())); assert input != null; transactions = input.map(new GraphTransactionFromText(getConfig())); if (hasVertexDictionary()) { transactions = transactions .map(new VertexLabelDecoder()) .withBroadcastSet( getVertexDictionary(), TLFConstants.VERTEX_DICTIONARY); } if (hasEdgeDictionary()) { transactions = transactions .map(new EdgeLabelDecoder()) .withBroadcastSet( getEdgeDictionary(), TLFConstants.EDGE_DICTIONARY); } return getConfig().getGraphCollectionFactory().fromTransactions(transactions); } TLFDataSource(String tlfPath, GradoopFlinkConfig config);  TLFDataSource(String tlfPath, String tlfVertexDictionaryPath,
    String tlfEdgeDictionaryPath, GradoopFlinkConfig config); @Override LogicalGraph getLogicalGraph(); @Override GraphCollection getGraphCollection();  }