Flink教程（30）- Flink VS Spark（下）

//    改行代码会计算这个job，要消费的每个kafka分区的最大偏移
override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
   //    构建KafkaRDD，用指定的分区数和要消费的offset范围
   val untilOffsets = clamp(latestLeaderOffsets(maxRetries))
   val rdd = KafkaRDD[K, V, U, T, R](
     // Report the record number and metadata of this batch interval to InputInfoTracker.
     context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)    
   val offsetRanges = currentOffsets.map { case (tp, fo) =>
     val uo = untilOffsets(tp)      
     OffsetRange(tp.topic, tp.partition, fo, uo.offset)
   }    
   val description = offsetRanges.filter { offsetRange =>
     // Don't display empty ranges.
     offsetRange.fromOffset != offsetRange.untilOffset
   }.map { offsetRange =>
     // Copy offsetRanges to immutable.List to prevent from being modified by the user
     s"topic: ${offsetRange.topic}\tpartition: ${offsetRange.partition}\t" +
       s"offsets: ${offsetRange.fromOffset} to ${offsetRange.untilOffset}"
   }.mkString("\n")    
   val metadata = Map("offsets" -> offsetRanges.toList,      
        StreamInputInfo.METADATA_KEY_DESCRIPTION -> description)    
   val inputInfo = StreamInputInfo(id, rdd.count, metadata)
   ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
   currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)    
   Some(rdd)
}

// 可以看到的是用来指定获取最大偏移分区的列表还是只有currentOffsets，没有发现关于新增的分区的内容。
@tailrec  protected final def latestLeaderOffsets(retries: Int): Map[TopicAndPartition, LeaderOffset] = {
   val o = kc.getLatestLeaderOffsets(currentOffsets.keySet)    
   // Either.fold would confuse @tailrec, do it manually
   if (o.isLeft) {      
      val err = o.left.get.toString      
      if (retries <= 0) {        
        throw new SparkException(err)
      } else {
       logError(err)        
       Thread.sleep(kc.config.refreshLeaderBackoffMs)
       latestLeaderOffsets(retries - 1)
     }
   } else {
     o.right.get
   }
 }

currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)

//    获取当前生成job，要用到的KafkaRDD每个分区最大消费偏移值
   val untilOffsets = clamp(latestOffsets())

/**   
* Returns the latest (highest) available offsets, taking new partitions into account.   
*/
protected def latestOffsets(): Map[TopicPartition, Long] = {    
   val c = consumer
   paranoidPoll(c)    // 获取所有的分区信息
   // make sure new partitions are reflected in currentOffsets
   val parts = c.assignment().asScala    
   // 做差获取新增的分区信息
   val newPartitions = parts.diff(currentOffsets.keySet)    
   // position for new partitions determined by auto.offset.reset if no commit
   // 新分区消费位置，没有记录的化是由auto.offset.reset决定
   currentOffsets = currentOffsets ++ newPartitions.map(tp => tp -> c.position(tp)).toMap    
   // don't want to consume messages, so pause
   c.pause(newPartitions.asJava)    // find latest available offsets
   c.seekToEnd(currentOffsets.keySet.asJava)
   parts.map(tp => tp -> c.position(tp)).toMap
}

this.kafkaFetcher = createFetcher(
       sourceContext,
       subscribedPartitionsToStartOffsets,
       periodicWatermarkAssigner,
       punctuatedWatermarkAssigner,
       (StreamingRuntimeContext) getRuntimeContext(),
       offsetCommitMode,
       getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
       useMetrics);

if (discoveryIntervalMillis != PARTITION_DISCOVERY_DISABLED) {      
  final AtomicReference<Exception> discoveryLoopErrorRef = new AtomicReference<>();      
  this.discoveryLoopThread = new Thread(new Runnable() {        
      @Override
      public void run() {          
          try {            
          // --------------------- partition discovery loop ---------------------
           List<KafkaTopicPartition> discoveredPartitions;            
           // throughout the loop, we always eagerly check if we are still running before
           // performing the next operation, so that we can escape the loop as soon as possible
           while (running) {              
             if (LOG.isDebugEnabled()) {                
               LOG.debug("Consumer subtask {} is trying to discover new partitions ...", getRuntimeContext().getIndexOfThisSubtask());
             }              
             try {
               discoveredPartitions = partitionDiscoverer.discoverPartitions();
             } catch (AbstractPartitionDiscoverer.WakeupException | AbstractPartitionDiscoverer.ClosedException e) { 
               // the partition discoverer may have been closed or woken up before or during the discovery;
               // this would only happen if the consumer was canceled; simply escape the loop
               break;
             } 
             // no need to add the discovered partitions if we were closed during the meantime
             if (running && !discoveredPartitions.isEmpty()) {
               kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
             } 
             // do not waste any time sleeping if we're not running anymore
             if (running && discoveryIntervalMillis != 0) {                
               try {                  
                 Thread.sleep(discoveryIntervalMillis);
               } catch (InterruptedException iex) {
                 // may be interrupted if the consumer was canceled midway; simply escape the loop
                 break;
               }
             }
           }
         } catch (Exception e) {
           discoveryLoopErrorRef.set(e);
         } finally {            
           // calling cancel will also let the fetcher loop escape
           // (if not running, cancel() was already called)
           if (running) {
             cancel();
           }
         }
       }
     }, "Kafka Partition Discovery for " + getRuntimeContext().getTaskNameWithSubtasks());
}
discoveryLoopThread.start();
kafkaFetcher.runFetchLoop();

rdd.repartition(1).foreachPartition(partition=>{    
        //    开启事务
       partition.foreach(each=>{//提交数据
       })    //  提交事务
   })
 })

def compute(time: Long, // in milliseconds
     numElements: Long,      
     processingDelay: Long, // in milliseconds
     schedulingDelay: Long // in milliseconds
   ): Option[Double] = {
   logTrace(s"\ntime = $time, # records = $numElements, " +
     s"processing time = $processingDelay, scheduling delay = $schedulingDelay")    
     this.synchronized {      
       if (time > latestTime && numElements > 0 && processingDelay > 0) {        
       val delaySinceUpdate = (time - latestTime).toDouble / 1000
       val processingRate = numElements.toDouble / processingDelay * 1000
       val error = latestRate - processingRate        
       val historicalError = schedulingDelay.toDouble * processingRate / batchIntervalMillis        
       // in elements/(second ^ 2)
       val dError = (error - latestError) / delaySinceUpdate       
       val newRate = (latestRate - proportional * error -
                                   integral * historicalError -
                                   derivative * dError).max(minRate)
       logTrace(s"""  | latestRate = $latestRate, error = $error            
                      | latestError = $latestError, historicalError = $historicalError            
                      | delaySinceUpdate = $delaySinceUpdate, dError = $dError           
                 """.stripMargin)
       latestTime = time        
       if (firstRun) {
         latestRate = processingRate
         latestError = 0D
         firstRun = false
         logTrace("First run, rate estimation skipped")         
         None
       } else {
         latestRate = newRate
         latestError = error
         logTrace(s"New rate = $newRate")          
         Some(newRate)
       }
     } else {
       logTrace("Rate estimation skipped")        
       None
     }
   }
 }