Kafkas konu başlığından bir RDD oluşturabilir miyim?

Question

KafkaUtils.createRDD, offsetRanges'i bir parametre olarak alır. Okumak istediğim konuya kadar bilmiyorum. En çok konudaki ilk 30 mesajı okumak istiyorum.

Görüyorum ki bir KafkaCluster.html#getLatestLeaderOffsets var, ancak bu bir Geliştirme API'si olarak açıklanmıştır.

Bir konu için en eski ve en son sapmaları belirlemeye yönelik herhangi bir genel yol var mı?

Answer 1

Bu basit bir şey değil, çünkü yalnızca bireysel aracılar belirli bir konu/bölüm için en yeni ofset bilgisinin ne olduğunu biliyorlar.

OffsetRequest'u yapabilirsiniz. Aşağıdakiler, bir konu/bölüm için en erken ve en son uzaklıkları döndürecektir (Scala'dır, ancak Scala'yı kullanmıyorsanız bu fikri elde edebilmelisiniz).

İstenen bölümün lideri olan aracıya bağlı bir SimpleConsumer kullanmak zorunda olduğunuzu unutmayın. Genelde yaptığım şey, her bir brokerim için bir SimpleConsumer yaratıyorum. Sonra bir meta veri isteği yapmak ve lider eşleştirme bölümü olsun, bunu daha sonra foreach bölüm:

def getOffsets(consumer: SimpleConsumer, topic: String, partition: Int) : (Long,Long) = { 
    val time = kafka.api.OffsetRequest.LatestTime 
    val reqInfo = Map[TopicAndPartition,PartitionOffsetRequestInfo]((new TopicAndPartition(topic, partition)) -> (new PartitionOffsetRequestInfo(time, 1000))) 
    val req = new kafka.javaapi.OffsetRequest(reqInfo, kafka.api.OffsetRequest.CurrentVersion, "offReq") 
    val resp = consumer.getOffsetsBefore(req) 
    val offsets = resp.offsets(topic, partition) 
    if (offsets.size > 0) (offsets(offsets.size - 1), offsets(0)) 
    else (0, -1) 
} 

Umut bu yardımcı olur.