public class StreamingJob {

	public static void main(String[] args) throws Exception {
		// Set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.enableCheckpointing(4000);

		// Set up a Cloud Pub/Sub topic as input

		// deserializer controls how Apache Flink deserializes the message
		// returns from Cloud Pub/Sub (a byte string) to a PubSubEvent
		// (custom Java class). See Deserialization.java and
		// PubSubEvent.java for definition.
		PlayCountEventDeserializationSchema deserializer = new PlayCountEventDeserializationSchema();

		Map<String, String> envVars = System.getenv();
		String projectId = envVars.get("GCP_PROJECT");
		String subscription = envVars.get("PUBSUB_SUBSCRIPTION");

		SourceFunction<PubSubEvent> pubSubSource = PubSubSource.newBuilder()
															   .withDeserializationSchema(deserializer)
															   .withProjectName(projectId)
															   .withSubscriptionName(subscription)
															   .build();

		DataStream<PubSubEvent> dataStream = env.addSource(pubSubSource);

    ...
  }
}