docs: @Override
  	public DataSet<Result<K>> runInternal(Graph<K, VV, EV> input)
  			throws Exception {
  		// u, v, w, bitmask
  		DataSet<TriangleListing.Result<K>> triangles = input
  			.run(new TriangleListing<K, VV, EV>()
  				.setParallelism(parallelism));
  
  		// u, edge count
  		DataSet<Tuple2<K, LongValue>> triangleVertices = triangles
  			.flatMap(new SplitTriangles<>())
  				.name("Split triangle vertices");
  
  		// u, triangle count
  		DataSet<Tuple2<K, LongValue>> vertexTriangleCount = triangleVertices
  			.groupBy(0)
  			.reduce(new CountTriangles<>())
  			.setCombineHint(CombineHint.HASH)
  				.name("Count triangles")
  				.setParallelism(parallelism);
  
  		// u, deg(u)
  		DataSet<Vertex<K, Degrees>> vertexDegree = input
  			.run(new VertexDegrees<K, VV, EV>()
  				.setIncludeZeroDegreeVertices(includeZeroDegreeVertices.get())
  				.setParallelism(parallelism));
  
  		// u, deg(u), triangle count
  		return vertexDegree
  			.leftOuterJoin(vertexTriangleCount)
  			.where(0)
  			.equalTo(0)
  			.with(new JoinVertexDegreeWithTriangleCount<>())
  				.setParallelism(parallelism)
  				.name("Clustering coefficient");
  	}