class StarGraph extends GraphGeneratorBase<LongValue, NullValue, NullValue> {

	@Override
	public Graph<LongValue, NullValue, NullValue> generate() {
		Preconditions.checkState(vertexCount >= 2);

		// Vertices
		DataSet<Vertex<LongValue, NullValue>> vertices = GraphGeneratorUtils.vertexSequence(env, parallelism, vertexCount);

		// Edges
		LongValueSequenceIterator iterator = new LongValueSequenceIterator(1, this.vertexCount - 1);

		DataSet<Edge<LongValue, NullValue>> edges = env
			.fromParallelCollection(iterator, LongValue.class)
				.setParallelism(parallelism)
				.name("Edge iterators")
			.flatMap(new LinkVertexToCenter())
				.setParallelism(parallelism)
				.name("Star graph edges");

		// Graph
		return Graph.fromDataSet(vertices, edges, env);
	}

	public  StarGraph(ExecutionEnvironment env, long vertexCount);

}