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

kafkaParams: ju.Map[String, Object], offsets: ju.Map[TopicPartition, jl.Long]

) extends ConsumerStrategy[K, V] with Logging { /**

*继承至ConsumerStrategy的方法，获取配置参数 *

* @return */

def executorKafkaParams: ju.Map[String, Object] = kafkaParams

/** *

* 继承至ConsumerStrategy的方法,用于创建Consumer * @return */

def onStart(currentOffsets: ju.Map[TopicPartition, jl.Long]): Consumer[K, V] = { val consumer = new KafkaConsumer[K, V](kafkaParams) consumer.subscribe(topics)

val toSeek = if (currentOffsets.isEmpty) { offsets } else {

currentOffsets }

if (!toSeek.isEmpty) {

// work around KAFKA-3370 when reset is none

// poll will throw if no position, i.e. auto offset reset none and no explicit position

// but cant seek to a position before poll, because poll is what gets subscription partitions // So, poll, suppress the first exception, then seek

val aor = kafkaParams.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)

val shouldSuppress = aor != null && aor.asInstanceOf[String].toUpperCase == \ try {

consumer.poll(0) } catch {

case x: NoOffsetForPartitionException if shouldSuppress =>

logWarning(\

ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + \is none. See KAFKA-3370\ }

toSeek.asScala.foreach { case (topicPartition, offset) => consumer.seek(topicPartition, offset) }

// we've called poll, we must pause or next poll may consume messages and set position consumer.pause(consumer.assignment()) }

consumer } } /**

* Subscribe to all topics matching specified pattern to get dynamically assigned partitions. * The pattern matching will be done periodically against topics existing at the time of check. * @param pattern pattern to subscribe to

* @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. *

* SubscribePattern是订阅所有匹配正则表达式模式的topic的数据

* topic模式匹配将会周期性的进行匹配，而不是在检查已存在的topic时候进行模式匹配 *

* kafkaParams参数：

*配置参数将会在driver上使用，并且相同参数也会发送到executor上使用，在executor上可能略微的配置修改

*通知需要设置bootstrap.servers指定broker *

*offsets参数：

* offsets伴随着startup开始,如果对于TopicPartition没有指定offset， * 提交的offset或auto.offset.reset将会被使用 * * */

private case class SubscribePattern[K, V](

pattern: ju.regex.Pattern,

kafkaParams: ju.Map[String, Object], offsets: ju.Map[TopicPartition, jl.Long] ) extends ConsumerStrategy[K, V] with Logging {

def executorKafkaParams: ju.Map[String, Object] = kafkaParams

def onStart(currentOffsets: ju.Map[TopicPartition, jl.Long]): Consumer[K, V] = { val consumer = new KafkaConsumer[K, V](kafkaParams)

consumer.subscribe(pattern, new NoOpConsumerRebalanceListener()) val toSeek = if (currentOffsets.isEmpty) { offsets } else {

currentOffsets }

if (!toSeek.isEmpty) {

// work around KAFKA-3370 when reset is none, see explanation in Subscribe above val aor = kafkaParams.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)

val shouldSuppress = aor != null && aor.asInstanceOf[String].toUpperCase == \ try {

consumer.poll(0) } catch {

case x: NoOffsetForPartitionException if shouldSuppress =>

logWarning(\

ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + \is none. See KAFKA-3370\ }

toSeek.asScala.foreach { case (topicPartition, offset) => consumer.seek(topicPartition, offset) }

// we've called poll, we must pause or next poll may consume messages and set position consumer.pause(consumer.assignment()) }

consumer } } /**

* Assign a fixed collection of TopicPartitions

* @param topicPartitions collection of TopicPartitions to assign * @param kafkaParams Kafka

* configuration parameters to be used on driver. The same params will be used on executors, * with minor autowww.sm136.commatic 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. *

* 订阅一个固定数目的TopicPartitions * kafkaParams参数：

*配置参数将会在driver上使用，并且相同参数也会发送到executor上使用，在executor上可能略微的配置修改

*通知需要设置bootstrap.servers指定broker *

*offsets参数：

* offsets伴随着startup开始,如果对于TopicPartition没有指定offset， * 提交的offset或auto.offset.reset将会被使用 …… 此处隐藏：3355字，全部文档内容请下载后查看。喜欢就下载吧 ……