val streams = (1 to inputTopic.partitions) map { _ =>
          KafkaUtils.createStream[Array[Byte], Array[Byte], DefaultDecoder, DefaultDecoder](
            ssc,
            kafkaParams,
            Map(inputTopic.name -> 1),
            storageLevel = StorageLevel.MEMORY_ONLY_SER // or: StorageLevel.MEMORY_AND_DISK_SER
          ).map(_._2)
        }