大数据Flink VS Spark

val Array(brokers, topics) = args//    创建一个批处理时间是2s的context    
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")    
val ssc = new StreamingContext(sparkConf, Seconds(2))    
//    使用broker和topic创建DirectStream    
val topicsSet = topics.split(",").toSet    
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)    
val messages = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent,    ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))  
// Get the lines, split them into words, count the words and print    
val lines = messages.map(_.value)    
val words = lines.flatMap(_.split(" "))    
val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)   
wordCounts.print()     //    启动流    
ssc.start()    
ssc.awaitTermination()

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
// create a checkpoint every 5 seconds
env.enableCheckpointing(5000); 
// make parameters available in the web interface
env.getConfig().setGlobalJobParameters(parameterTool); 
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//  ExecutionConfig.GlobalJobParameters
env.getConfig().setGlobalJobParameters(null);    
DataStream<KafkaEvent> input = env
           .addSource(new FlinkKafkaConsumer010<>(
                   parameterTool.getRequired("input-topic"),  new KafkaEventSchema(),
                   parameterTool.getProperties())
           .assignTimestampsAndWatermarks(new CustomWatermarkExtractor()))
           .setParallelism(1).rebalance()
           .keyBy("word")
           .map(new RollingAdditionMapper()).setParallelism(0);
input.addSink(new FlinkKafkaProducer010<>(parameterTool.getRequired("output-topic"), new KafkaEventSchema(),
        parameterTool.getProperties()));
env.execute("Kafka 0.10 Example");

/    改行代码会计算这个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.

// 可以看到的是用来指定获取最大偏移分区的列表还是只有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();

Dstream.foreachRDD(rdd=>{
   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
     }
   }
 }