Kafka（Re-）加入小組堅持超過2個主題

Question

我正在開發一個使用Kafka作爲消息發佈/子工具的系統。

數據由斯卡拉腳本生成的：

val kafkaParams = new Properties() 
    kafkaParams.put("bootstrap.servers", "localhost:9092") 
    kafkaParams.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer") 
    kafkaParams.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer") 
    kafkaParams.put("group.id", "test_luca") 

    //kafka producer 
    val producer = new KafkaProducer[String, String](kafkaParams) 

    //Source list 
    val s1 = new java.util.Timer() 
    val tasks1 = new java.util.TimerTask { 
     def run() = { 
      val date = new java.util.Date 
      val date2 = date.getTime() 
      val send = ""+ date2 + ", 45.1234, 12.5432, 4.5, 3.0" 
      val data = new ProducerRecord[String,String]("topic_s1", send) 
      producer.send(data) 
     } 
    } 
    s1.schedule(tasks1, 1000L, 1000L) 

    val s2 = new java.util.Timer() 
    val tasks2 = new java.util.TimerTask { 
     def run() = { 
      val date = new java.util.Date 
      val date2 = date.getTime() 
      val send = ""+ date2 + ", 1.111, 9.999, 10.4, 10.0" 
      val data = new ProducerRecord[String,String]("topic_s2", send) 
      producer.send(data) 
     } 
    } 
    s2.schedule(tasks2, 2000L, 2000L) 

我需要測試在某些特定情況下的卡夫卡表演。在其中一種情況下，我有一個使用主題爲「topic_s1」和「topic_s2」的數據的其他腳本，對它們進行詳細說明，然後生成包含不同主題（topic_s1b和topic_s2b）的新數據。隨後，這些詳細的數據被Apache Spark Streaming腳本使用。

如果我省略消費者/生產者腳本（我只有一個擁有2個主題和Spark腳本的卡夫卡製作人），一切正常。

如果我使用完整配置（包含2個主題的1個kafka生產者，使用來自kafka生產者的數據的「中間件」腳本，詳細說明它們並用新主題生成新數據，1個使用新主題消費數據的spark腳本）Spark Streaming腳本卡住了INFO AbstractCoordinator: (Re-)joining group test_luca

我在本地運行所有內容，並且不對kafka和zookeeper配置進行修改。

有什麼建議嗎？

UPDATE：火花腳本：

val sparkConf = new SparkConf().setAppName("SparkScript").set("spark.driver.allowMultipleContexts", "true").setMaster("local[2]") 
val sc = new SparkContext(sparkConf) 

val ssc = new StreamingContext(sc, Seconds(4)) 

case class Thema(name: String, metadata: JObject) 
case class Tempo(unit: String, count: Int, metadata: JObject) 
case class Spatio(unit: String, metadata: JObject) 
case class Stt(spatial: Spatio, temporal: Tempo, thematic: Thema) 
case class Location(latitude: Double, longitude: Double, name: String) 

case class Data(location: Location, timestamp: Long, measurement: Int, unit: String, accuracy: Double) 
case class Sensor(sensor_name: String, start_date: String, end_date: String, data_schema: Array[String], data: Data, stt: Stt) 


case class Datas(location: Location, timestamp: Long, measurement: Int, unit: String, accuracy: Double) 
case class Sensor2(sensor_name: String, start_date: String, end_date: String, data_schema: Array[String], data: Datas, stt: Stt) 


val kafkaParams = Map[String, Object](
    "bootstrap.servers" -> "localhost:9092", 
    "key.deserializer" -> classOf[StringDeserializer].getCanonicalName, 
    "value.deserializer" -> classOf[StringDeserializer].getCanonicalName, 
    "group.id" -> "test_luca", 
    "auto.offset.reset" -> "latest", 
    "enable.auto.commit" -> (false: java.lang.Boolean) 
) 

val topics1 = Array("topics1") 
val topics2 = Array("topics2") 

val stream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics1, kafkaParams)) 
val stream2 = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics2, kafkaParams)) 

val s1 = stream.map(record => { 
    implicit val formats = DefaultFormats 
    parse(record.value).extract[Sensor] 
} 
) 
val s2 = stream2.map(record => { 
    implicit val formats = DefaultFormats 
    parse(record.value).extract[Sensor2] 
} 
) 

val f1 = s1.map { x => x.sensor_name } 
f1.print() 
val f2 = s2.map { x => x.sensor_name } 
f2.print() 

感謝 盧卡

Answer 1

也許你應該改變火花流腳本的group.id。我猜你的「中間件」腳本的使用者與你的spark流腳本的使用者有相同的group.id。那麼可怕的事情就會發生。

在kafka中，消費者羣體是話題的真正訂閱者，羣組中的消費者只是一個拆分工作者，所以對於您的情況，您應該在中間件腳本消費者和spark流腳本消費者中使用不同的group.id。

在你沒有中間腳本的第一次嘗試中，它只是因爲這個。