Spark Streaming VS Flink

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));
    env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
    env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
    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");

override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {//    改行代码会计算这个job，要消费的每个kafka分区的最大偏移
    val untilOffsets = clamp(latestLeaderOffsets(maxRetries))//    构建KafkaRDD，用指定的分区数和要消费的offset范围
    val rdd = KafkaRDD[K, V, U, T, R](
      context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)    // Report the record number and metadata of this batch interval to InputInfoTracker.
    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 =>
      s"topic: ${offsetRange.topic}\tpartition: ${offsetRange.partition}\t" +
        s"offsets: ${offsetRange.fromOffset} to ${offsetRange.untilOffset}"
    }.mkString("\n")    // Copy offsetRanges to immutable.List to prevent from being modified by the user
    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)
  }

@tailrec  protected final def latestLeaderOffsets(retries: Int): Map[TopicAndPartition, LeaderOffset] = {//    可以看到的是用来指定获取最大偏移分区的列表还是只有currentOffsets，没有发现关于新增的分区的内容。
    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)    // 获取所有的分区信息
    val parts = c.assignment().asScala    // make sure new partitions are reflected in currentOffsets
    // 做差获取新增的分区信息
    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
      }
    }
  }