Modified to Spark version 0.8 to integrate with Kafka 0.7