docs: @Override
  	public Graph<K, VV, EV> runInternal(Graph<K, VV, EV> input)
  			throws Exception {
  		// u, d(u)
  		DataSet<Vertex<K, LongValue>> vertexDegree = input
  			.run(new VertexDegree<K, VV, EV>()
  				.setReduceOnTargetId(reduceOnTargetId.get())
  				.setParallelism(parallelism));
  
  		// u, d(u) if d(u) > maximumDegree
  		DataSet<Tuple1<K>> highDegreeVertices = vertexDegree
  			.flatMap(new DegreeFilter<>(maximumDegree))
  				.setParallelism(parallelism)
  				.name("Filter high-degree vertices");
  
  		JoinHint joinHint = broadcastHighDegreeVertices.get() ? JoinHint.BROADCAST_HASH_SECOND : JoinHint.REPARTITION_HASH_SECOND;
  
  		// Vertices
  		DataSet<Vertex<K, VV>> vertices = input
  			.getVertices()
  			.leftOuterJoin(highDegreeVertices, joinHint)
  			.where(0)
  			.equalTo(0)
  			.with(new ProjectVertex<>())
  				.setParallelism(parallelism)
  				.name("Project low-degree vertices");
  
  		// Edges
  		DataSet<Edge<K, EV>> edges = input
  			.getEdges()
  			.leftOuterJoin(highDegreeVertices, joinHint)
  			.where(reduceOnTargetId.get() ? 1 : 0)
  			.equalTo(0)
  				.with(new ProjectEdge<>())
  				.setParallelism(parallelism)
  				.name("Project low-degree edges by " + (reduceOnTargetId.get() ? "target" : "source"))
  			.leftOuterJoin(highDegreeVertices, joinHint)
  			.where(reduceOnTargetId.get() ? 0 : 1)
  			.equalTo(0)
  			.with(new ProjectEdge<>())
  				.setParallelism(parallelism)
  				.name("Project low-degree edges by " + (reduceOnTargetId.get() ? "source" : "target"));
  
  		// Graph
  		return Graph.fromDataSet(vertices, edges, input.getContext());
  	}