private CoGroupOperator < ? , ? , Tuple2 < K , Message > > buildScatterFunctionVerticesWithDegrees ( DeltaIteration < Vertex < K , Tuple3 < VV , LongValue , LongValue > > , Vertex < K , Tuple3 < VV , LongValue , LongValue > > > iteration , TypeInformation < Tuple2 < K , Message > > messageTypeInfo , int whereArg , int equalToArg , DataSet < LongValue > numberOfVertices ) { // build the scatter function (co group) CoGroupOperator < ? , ? , Tuple2 < K , Message > > messages ; ScatterUdfWithEdgeValues < K , Tuple3 < VV , LongValue , LongValue > , VV , Message , EV > messenger = new ScatterUdfWithEVsVVWithDegrees <> ( scatterFunction , messageTypeInfo ) ; messages = this . edgesWithValue . coGroup ( iteration . getWorkset ( ) ) . where ( whereArg ) . equalTo ( equalToArg ) . with ( messenger ) ; // configure coGroup message function with name and broadcast variables messages = messages . name ( "Messaging" ) ; if ( this . configuration != null ) { for ( Tuple2 < String , DataSet < ? > > e : this . configuration . getScatterBcastVars ( ) ) { messages = messages . withBroadcastSet ( e . f1 , e . f0 ) ; } if ( this . configuration . isOptNumVertices ( ) ) { messages = messages . withBroadcastSet ( numberOfVertices , "number of vertices" ) ; } } return messages ; }
Method that builds the scatter function using a coGroup operator for a vertex containing degree information . It afterwards configures the function with a custom name and broadcast variables .