@ Override public DataSet < Vertex < K , VV > > createResult ( ) { if ( this . initialVertices == null ) { throw new IllegalStateException ( "The input data set has not been set." ) ; } // prepare some type information TypeInformation < K > keyType = ( ( TupleTypeInfo < ? > ) initialVertices . getType ( ) ) . getTypeAt ( 0 ) ; TypeInformation < Tuple2 < K , Message > > messageTypeInfo = new TupleTypeInfo <> ( keyType , messageType ) ; // create a graph Graph < K , VV , EV > graph = Graph . fromDataSet ( initialVertices , edgesWithValue , initialVertices . getExecutionEnvironment ( ) ) ; // check whether the numVertices option is set and, if so, compute the total number of vertices // and set it within the scatter and gather functions DataSet < LongValue > numberOfVertices = null ; if ( this . configuration != null && this . configuration . isOptNumVertices ( ) ) { try { numberOfVertices = GraphUtils . count ( this . initialVertices ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; } } if ( this . configuration != null ) { scatterFunction . setDirection ( this . configuration . getDirection ( ) ) ; } else { scatterFunction . setDirection ( EdgeDirection . OUT ) ; } // retrieve the direction in which the updates are made and in which the messages are sent EdgeDirection messagingDirection = scatterFunction . getDirection ( ) ; // check whether the degrees option is set and, if so, compute the in and the out degrees and // add them to the vertex value if ( this . configuration != null && this . configuration . isOptDegrees ( ) ) { return createResultVerticesWithDegrees ( graph , messagingDirection , messageTypeInfo , numberOfVertices ) ; } else { return createResultSimpleVertex ( messagingDirection , messageTypeInfo , numberOfVertices ) ; } }
Creates the operator that represents this scatter - gather graph computation .