スパーク：バッチの終わりにカフカオフセットをコミット

Question

バージョン：

• スパーク2.2
• カフカ0.11

Iを使用する必要がカフカのオフセットをコミットするdocumentationに従って：

stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)

結果として、オフセットはsta次のバッチのrt。それは「一定の」遅れを引き起こす。

現在のバッチの最後にオフセットをコミットする（ただし、引き続き同じkafkaグループをオフセットに使用する）回避策はありますか？ラグモニタリングの

例：

Answer 1

は、現在のバッチcommitAsync API経由

ないの終わりにオフセットをコミットする任意の回避策があります。

override def compute(validTime: Time): Option[KafkaRDD[K, V]] = { 
    val untilOffsets = clamp(latestOffsets()) 

    // Create KafkaRDD and other irrelevant code 

    currentOffsets = untilOffsets 
    commitAll() 
    Some(rdd) 
} 

キューが commitAsyncによっていっぱいになっ commitAllだけポーリング

：

protected def commitAll(): Unit = { 
    val m = new ju.HashMap[TopicPartition, OffsetAndMetadata]() 
    var osr = commitQueue.poll() 
    while (null != osr) { 
    val tp = osr.topicPartition 
    val x = m.get(tp) 
    val offset = if (null == x) { osr.untilOffset } else { Math.max(x.offset, osr.untilOffset) } 
    m.put(tp, new OffsetAndMetadata(offset)) 
    osr = commitQueue.poll() 
    } 
    if (!m.isEmpty) { 
    consumer.commitAsync(m, commitCallback.get) 
    } 
} 

したがってどのようなメソッド呼び出しがないと、キューアップコミットするオフセット、その後、DirectKafkaInputDStream.compute中に非同期コミットんです残念ながら、オフセットをトランザクションとしてコミットする場合は、独自のストアに別々に格納し、カフカのオフセットコミットトラッキングを使用しないでください。