Spark整合kafka0.10.0新特性（一）(4)

* Subscribe to a collection of topics.

* @param topics collection of topics to subscribe * @param kafkaParams Kafka

*

* configuration parameters to be used on driver. The same params will be used on executors,

* with minor automatic modifications applied. * Requires \

* with Kafka broker(s) specified in host1:port1,host2:port2 form.

* @param offsets: offsets to begin at on initial startup. If no offset is given for a * TopicPartition, the committed offset (if applicable) or kafka param * auto.offset.reset will be used. */

@Experimental

def Subscribe[K, V](

topics: Iterable[jl.String],

kafkaParams: collection.Map[String, Object],

offsets: collection.Map[TopicPartition, Long]): ConsumerStrategy[K, V] = {

new Subscribe[K, V](

new ju.ArrayList(topics.asJavaCollection),

new ju.HashMap[String, Object](kafkaParams.asJava),

new ju.HashMap[TopicPartition, jl.Long](offsets.mapValues(l => new jl.Long(l)).asJava)) } /**

* :: Experimental ::

* Subscribe to a collection of topics.

* @param topics collection of topics to subscribe * @param kafkaParams Kafka

*

* configuration parameters to be used on driver. The same params will be used on executors,

* with minor automatic modifications applied. * Requires \

* with Kafka broker(s) specified in host1:port1,host2:port2 form. */

@Experimental

def Subscribe[K, V](

topics: Iterable[jl.String],

kafkaParams: collection.Map[String, Object]): ConsumerStrategy[K, V] = {

new Subscribe[K, V](

new ju.ArrayList(topics.asJavaCollection),

new ju.HashMap[String, Object](kafkaParams.asJava), ju.Collections.emptyMap[TopicPartition, jl.Long]()) } /**

* :: Experimental ::

* Subscribe to a collection of topics.

* @param topics collection of topics to subscribe * @param kafkaParams Kafka

* configuration parameters to be used on driver. The same params will be used on executors,

* with minor automatic modifications applied. * Requires \

* with Kafka broker(s) specified in host1:port1,host2:port2 form.

* @param offsets: offsets to begin at on initial startup. If no offset is given for a * TopicPartition, the committed offset (if applicable) or kafka param * auto.offset.reset will be used. */

@Experimental