[ https://issues.apache.org/jira/browse/SPARK-19680?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Schakmann Rene updated SPARK-19680: ----------------------------------- Description: I'm using spark streaming with kafka to acutally create a toplist. I want to read all the messages in kafka. So I set "auto.offset.reset" -> "earliest" Nevertheless when I start the job on our spark cluster it is not working I get: Error: Job aborted due to stage failure: Task 2 in stage 111.0 failed 4 times, most recent failure: Lost task 2.3 in stage 111.0 (TID 1270, 194.232.55.23, executor 2): org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {SearchEvents-2=161803385} This is somehow wrong because i did set the auto.offset.reset property Setup: Kafka Parameter: def getDefaultKafkaReceiverParameter(properties: Properties):Map[String, Object] = { Map( "bootstrap.servers" -> properties.getProperty("kafka.bootstrap.servers"), "group.id" -> properties.getProperty("kafka.consumer.group"), "auto.offset.reset" -> "earliest", "spark.streaming.kafka.consumer.cache.enabled" -> "false", "enable.auto.commit" -> "false", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> "at.willhaben.sid.DTOByteDeserializer") } Job: {code:title=Bar.java|borderStyle=solid} def processSearchKeyWords(stream: InputDStream[ConsumerRecord[String, Array[Byte]]], windowDuration: Int, slideDuration: Int, kafkaSink: Broadcast[KafkaSink[TopList]]): Unit = { getFilteredStream(stream.map(_.value()), windowDuration, slideDuration).foreachRDD(rdd => { val topList = new TopList topList.setCreated(new Date()) topList.setTopListEntryList(rdd.take(TopListLength).toList) CurrentLogger.info("TopList length: " + topList.getTopListEntryList.size().toString) kafkaSink.value.send(SendToTopicName, topList) CurrentLogger.info("Last Run: " + System.currentTimeMillis()) }) } {code} def getFilteredStream(result: DStream[Array[Byte]], windowDuration: Int, slideDuration: Int): DStream[TopListEntry] = { val Mapper = MapperObject.readerFor[SearchEventDTO] result.repartition(100).map(s => Mapper.readValue[SearchEventDTO](s)) .filter(s => s != null && s.getSearchRequest != null && s.getSearchRequest.getSearchParameters != null && s.getVertical == Vertical.BAP && s.getSearchRequest.getSearchParameters.containsKey(EspParameterEnum.KEYWORD.getName)) .map(row => { val name = row.getSearchRequest.getSearchParameters.get(EspParameterEnum.KEYWORD.getName).getEspSearchParameterDTO.getValue.toLowerCase() (name, new TopListEntry(name, 1, row.getResultCount)) }) .reduceByKeyAndWindow( (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, a.getSearchCount + b.getSearchCount, a.getMeanSearchHits + b.getMeanSearchHits), (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, a.getSearchCount - b.getSearchCount, a.getMeanSearchHits - b.getMeanSearchHits), Minutes(windowDuration), Seconds(slideDuration)) .filter((x: (String, TopListEntry)) => x._2.getSearchCount > 200L) .map(row => (row._2.getSearchCount, row._2)) .transform(rdd => rdd.sortByKey(ascending = false)) .map(row => new TopListEntry(row._2.getKeyword, row._2.getSearchCount, row._2.getMeanSearchHits / row._2.getSearchCount)) } def main(properties: Properties): Unit = { val sparkSession = SparkUtil.getDefaultSparkSession(properties, TaskName) val kafkaSink = sparkSession.sparkContext.broadcast(KafkaSinkUtil.apply[TopList](SparkUtil.getDefaultSparkProperties(properties))) val kafkaParams: Map[String, Object] = SparkUtil.getDefaultKafkaReceiverParameter(properties) val ssc = new StreamingContext(sparkSession.sparkContext, Seconds(30)) ssc.checkpoint("/home/spark/checkpoints") val adEventStream = KafkaUtils.createDirectStream[String, Array[Byte]](ssc, PreferConsistent, Subscribe[String, Array[Byte]](Array(ReadFromTopicName), kafkaParams)) processSearchKeyWords(adEventStream, SparkUtil.getWindowDuration(properties), SparkUtil.getSlideDuration(properties), kafkaSink) ssc.start() ssc.awaitTermination() } was: I'm using spark streaming with kafka to acutally create a toplist. I want to read all the messages in kafka. So I set "auto.offset.reset" -> "earliest" Nevertheless when I start the job on our spark cluster it is not working I get: Error: Job aborted due to stage failure: Task 2 in stage 111.0 failed 4 times, most recent failure: Lost task 2.3 in stage 111.0 (TID 1270, 194.232.55.23, executor 2): org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {SearchEvents-2=161803385} This is somehow wrong because i did set the auto.offset.reset property Setup: Kafka Parameter: def getDefaultKafkaReceiverParameter(properties: Properties):Map[String, Object] = { Map( "bootstrap.servers" -> properties.getProperty("kafka.bootstrap.servers"), "group.id" -> properties.getProperty("kafka.consumer.group"), "auto.offset.reset" -> "earliest", "spark.streaming.kafka.consumer.cache.enabled" -> "false", "enable.auto.commit" -> "false", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> "at.willhaben.sid.DTOByteDeserializer") } Job: def processSearchKeyWords(stream: InputDStream[ConsumerRecord[String, Array[Byte]]], windowDuration: Int, slideDuration: Int, kafkaSink: Broadcast[KafkaSink[TopList]]): Unit = { getFilteredStream(stream.map(_.value()), windowDuration, slideDuration).foreachRDD(rdd => { val topList = new TopList topList.setCreated(new Date()) topList.setTopListEntryList(rdd.take(TopListLength).toList) CurrentLogger.info("TopList length: " + topList.getTopListEntryList.size().toString) kafkaSink.value.send(SendToTopicName, topList) CurrentLogger.info("Last Run: " + System.currentTimeMillis()) }) } def getFilteredStream(result: DStream[Array[Byte]], windowDuration: Int, slideDuration: Int): DStream[TopListEntry] = { val Mapper = MapperObject.readerFor[SearchEventDTO] result.repartition(100).map(s => Mapper.readValue[SearchEventDTO](s)) .filter(s => s != null && s.getSearchRequest != null && s.getSearchRequest.getSearchParameters != null && s.getVertical == Vertical.BAP && s.getSearchRequest.getSearchParameters.containsKey(EspParameterEnum.KEYWORD.getName)) .map(row => { val name = row.getSearchRequest.getSearchParameters.get(EspParameterEnum.KEYWORD.getName).getEspSearchParameterDTO.getValue.toLowerCase() (name, new TopListEntry(name, 1, row.getResultCount)) }) .reduceByKeyAndWindow( (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, a.getSearchCount + b.getSearchCount, a.getMeanSearchHits + b.getMeanSearchHits), (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, a.getSearchCount - b.getSearchCount, a.getMeanSearchHits - b.getMeanSearchHits), Minutes(windowDuration), Seconds(slideDuration)) .filter((x: (String, TopListEntry)) => x._2.getSearchCount > 200L) .map(row => (row._2.getSearchCount, row._2)) .transform(rdd => rdd.sortByKey(ascending = false)) .map(row => new TopListEntry(row._2.getKeyword, row._2.getSearchCount, row._2.getMeanSearchHits / row._2.getSearchCount)) } def main(properties: Properties): Unit = { val sparkSession = SparkUtil.getDefaultSparkSession(properties, TaskName) val kafkaSink = sparkSession.sparkContext.broadcast(KafkaSinkUtil.apply[TopList](SparkUtil.getDefaultSparkProperties(properties))) val kafkaParams: Map[String, Object] = SparkUtil.getDefaultKafkaReceiverParameter(properties) val ssc = new StreamingContext(sparkSession.sparkContext, Seconds(30)) ssc.checkpoint("/home/spark/checkpoints") val adEventStream = KafkaUtils.createDirectStream[String, Array[Byte]](ssc, PreferConsistent, Subscribe[String, Array[Byte]](Array(ReadFromTopicName), kafkaParams)) processSearchKeyWords(adEventStream, SparkUtil.getWindowDuration(properties), SparkUtil.getSlideDuration(properties), kafkaSink) ssc.start() ssc.awaitTermination() } > Offsets out of range with no configured reset policy for partitions > ------------------------------------------------------------------- > > Key: SPARK-19680 > URL: https://issues.apache.org/jira/browse/SPARK-19680 > Project: Spark > Issue Type: Bug > Components: Structured Streaming > Affects Versions: 2.1.0 > Reporter: Schakmann Rene > > I'm using spark streaming with kafka to acutally create a toplist. I want to > read all the messages in kafka. So I set > "auto.offset.reset" -> "earliest" > Nevertheless when I start the job on our spark cluster it is not working I > get: > Error: > Job aborted due to stage failure: Task 2 in stage 111.0 failed 4 times, > most recent failure: Lost task 2.3 in stage 111.0 (TID 1270, 194.232.55.23, > executor 2): org.apache.kafka.clients.consumer.OffsetOutOfRangeException: > Offsets out of range with no configured reset policy for partitions: > {SearchEvents-2=161803385} > This is somehow wrong because i did set the auto.offset.reset property > Setup: > Kafka Parameter: > def getDefaultKafkaReceiverParameter(properties: Properties):Map[String, > Object] = { > Map( > "bootstrap.servers" -> > properties.getProperty("kafka.bootstrap.servers"), > "group.id" -> properties.getProperty("kafka.consumer.group"), > "auto.offset.reset" -> "earliest", > "spark.streaming.kafka.consumer.cache.enabled" -> "false", > "enable.auto.commit" -> "false", > "key.deserializer" -> classOf[StringDeserializer], > "value.deserializer" -> "at.willhaben.sid.DTOByteDeserializer") > } > Job: > {code:title=Bar.java|borderStyle=solid} > def processSearchKeyWords(stream: InputDStream[ConsumerRecord[String, > Array[Byte]]], windowDuration: Int, slideDuration: Int, kafkaSink: > Broadcast[KafkaSink[TopList]]): Unit = { > getFilteredStream(stream.map(_.value()), windowDuration, > slideDuration).foreachRDD(rdd => { > val topList = new TopList > topList.setCreated(new Date()) > topList.setTopListEntryList(rdd.take(TopListLength).toList) > CurrentLogger.info("TopList length: " + > topList.getTopListEntryList.size().toString) > kafkaSink.value.send(SendToTopicName, topList) > CurrentLogger.info("Last Run: " + System.currentTimeMillis()) > }) > } > {code} > def getFilteredStream(result: DStream[Array[Byte]], windowDuration: Int, > slideDuration: Int): DStream[TopListEntry] = { > val Mapper = MapperObject.readerFor[SearchEventDTO] > result.repartition(100).map(s => Mapper.readValue[SearchEventDTO](s)) > .filter(s => s != null && s.getSearchRequest != null && > s.getSearchRequest.getSearchParameters != null && s.getVertical == > Vertical.BAP && > s.getSearchRequest.getSearchParameters.containsKey(EspParameterEnum.KEYWORD.getName)) > .map(row => { > val name = > row.getSearchRequest.getSearchParameters.get(EspParameterEnum.KEYWORD.getName).getEspSearchParameterDTO.getValue.toLowerCase() > (name, new TopListEntry(name, 1, row.getResultCount)) > }) > .reduceByKeyAndWindow( > (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, > a.getSearchCount + b.getSearchCount, a.getMeanSearchHits + > b.getMeanSearchHits), > (a: TopListEntry, b: TopListEntry) => new TopListEntry(a.getKeyword, > a.getSearchCount - b.getSearchCount, a.getMeanSearchHits - > b.getMeanSearchHits), > Minutes(windowDuration), > Seconds(slideDuration)) > .filter((x: (String, TopListEntry)) => x._2.getSearchCount > 200L) > .map(row => (row._2.getSearchCount, row._2)) > .transform(rdd => rdd.sortByKey(ascending = false)) > .map(row => new TopListEntry(row._2.getKeyword, row._2.getSearchCount, > row._2.getMeanSearchHits / row._2.getSearchCount)) > } > def main(properties: Properties): Unit = { > val sparkSession = SparkUtil.getDefaultSparkSession(properties, TaskName) > val kafkaSink = > sparkSession.sparkContext.broadcast(KafkaSinkUtil.apply[TopList](SparkUtil.getDefaultSparkProperties(properties))) > val kafkaParams: Map[String, Object] = > SparkUtil.getDefaultKafkaReceiverParameter(properties) > val ssc = new StreamingContext(sparkSession.sparkContext, Seconds(30)) > ssc.checkpoint("/home/spark/checkpoints") > val adEventStream = > KafkaUtils.createDirectStream[String, Array[Byte]](ssc, > PreferConsistent, Subscribe[String, Array[Byte]](Array(ReadFromTopicName), > kafkaParams)) > processSearchKeyWords(adEventStream, > SparkUtil.getWindowDuration(properties), > SparkUtil.getSlideDuration(properties), kafkaSink) > ssc.start() > ssc.awaitTermination() > } -- This message was sent by Atlassian JIRA (v6.3.15#6346) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org