private void constructExecutionGraph ( final JobGraph jobGraph , final InstanceManager instanceManager ) throws GraphConversionException { // Clean up temporary data structures final HashMap < AbstractJobVertex , ExecutionVertex > temporaryVertexMap = new HashMap < AbstractJobVertex , ExecutionVertex > ( ) ; final HashMap < AbstractJobVertex , ExecutionGroupVertex > temporaryGroupVertexMap = new HashMap < AbstractJobVertex , ExecutionGroupVertex > ( ) ; // Initially, create only one execution stage that contains all group vertices final ExecutionStage initialExecutionStage = new ExecutionStage ( this , 0 ) ; this . stages . add ( initialExecutionStage ) ; // Convert job vertices to execution vertices and initialize them final AbstractJobVertex [ ] all = jobGraph . getAllJobVertices ( ) ; for ( int i = 0 ; i < all . length ; i ++ ) { final ExecutionVertex createdVertex = createVertex ( all [ i ] , instanceManager , initialExecutionStage , jobGraph . getJobConfiguration ( ) ) ; temporaryVertexMap . put ( all [ i ] , createdVertex ) ; temporaryGroupVertexMap . put ( all [ i ] , createdVertex . getGroupVertex ( ) ) ; } // Create initial edges between the vertices createInitialGroupEdges ( temporaryVertexMap ) ; // Now that an initial graph is built, apply the user settings applyUserDefinedSettings ( temporaryGroupVertexMap ) ; // Calculate the connection IDs calculateConnectionIDs ( ) ; // Finally, construct the execution pipelines reconstructExecutionPipelines ( ) ; }
Sets up an execution graph from a job graph .