[ https://issues.apache.org/jira/browse/SPARK-40188?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Madhav Madhu updated SPARK-40188: --------------------------------- Description: Spark Kafka consumer is unable to read messages, of a certain size or count in batches. I have tried few approaches as mentioned in Kafka docs but with no success. Here is a link to Stack Overflow where I asked the same question with no response and think this is a possible bug here. Same configuration works fine when the consumer is a java code. https://stackoverflow.com/questions/73398533/spark-streaming-context-kafka-consumer-read-messages-of-a-certain-byte-size-in Here is the consumer code which fetches data from Kafka, {code:scala} val streamingContext = new StreamingContext(sparkSession.sparkContext, Seconds(10)) val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "localhost:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "test", "fetch.max.bytes" -> "65536", "max.partition.fetch.bytes" -> "8192", "max.poll.records" -> "100", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean), "sasl.jaas.config"-> "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";", "sasl.mechanism" -> "PLAIN", "security.protocol" -> "SASL_PLAINTEXT", ) val topics = Array("test.topic") val stream = KafkaUtils.createDirectStream[String, String]( streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) stream.foreachRDD { rdd => val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges println(offsetRanges.foreach(a => println(a.topic + ":" + a.partition + ":" + a.fromOffset + ":" + a.untilOffset + ":" + a.count()))) val df = rdd.map(a => a.value().split(",")).toDF() val selectCols = columns.indices.map(i => $"value"(i)) var newDF = df.select(selectCols: _*).toDF(columns: _*) // Some business operations here and then write to back to kafka. newDF.write .format("kafka") .option("kafka.bootstrap.servers", "localhost:9092") .option("topic", "topic.ouput") .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";") .option("kafka.sasl.mechanism", "PLAIN") .option("kafka.security.protocol", "SASL_PLAINTEXT") .save() stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) sparkSession.catalog.clearCache() } streamingContext.start() streamingContext.awaitTermination() {code} Output: {code:java} test.topic:6:1345075:4163058:2817983 test.topic:0:1339456:4144190:2804734 test.topic:3:1354266:4189336:2835070 test.topic:7:1353542:4186148:2832606 test.topic:5:1355140:4189071:2833931 test.topic:2:1351162:4173375:2822213 test.topic:1:1352801:4184073:2831272 test.topic:4:1348558:4166749:2818191 () test.topic:6:4163058:4163058:0 test.topic:0:4144190:4144190:0 test.topic:3:4189336:4189336:0 test.topic:7:4186148:4186148:0 test.topic:5:4189071:4189071:0 test.topic:2:4173375:4173375:0 test.topic:1:4184073:4184073:0 test.topic:4:4166749:4166749:0 {code} I tried different options as followed, Option 1: Topic Partition 8 Streaming Context 1 sec: "fetch.max.bytes" -> "65536", // 64 Kb "max.partition.fetch.bytes" -> "8192" // 8Kb "max.poll.records" -> "100" DataFrame count which it read from Kafka in the very first batch: 1200000 Option 2: Partition 1 Streaming Context 1 sec "fetch.max.bytes" -> "65536", "max.partition.fetch.bytes" -> "8192" "max.poll.records" -> "100" Kafka Lag: 126360469 DataFrame count which it read from Kafka in the very first batch: 126360469. was: Spark Kafka consumer is unable to read messages, of a certain size or count in batches. I have tried few approaches as mentioned in Kafka docs but with no success. Here is a link to Stack Overflow where I asked the same question with no response and think this is a possible bug here. Same configuration works fine when the consumer is a java code. https://stackoverflow.com/questions/73398533/spark-streaming-context-kafka-consumer-read-messages-of-a-certain-byte-size-in Here is the consumer code which fetches data from Kafka, {code:scala} val streamingContext = new StreamingContext(sparkSession.sparkContext, Seconds(10)) val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "localhost:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "test", "fetch.max.bytes" -> "65536", "max.partition.fetch.bytes" -> "8192", "max.poll.records" -> "100", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean), "sasl.jaas.config"-> "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";", "sasl.mechanism" -> "PLAIN", "security.protocol" -> "SASL_PLAINTEXT", ) val topics = Array("test.topic") val stream = KafkaUtils.createDirectStream[String, String]( streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) stream.foreachRDD { rdd => val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges println(offsetRanges.foreach(a => println(a.topic + ":" + a.partition + ":" + a.fromOffset + ":" + a.untilOffset + ":" + a.count()))) val df = rdd.map(a => a.value().split(",")).toDF() val selectCols = columns.indices.map(i => $"value"(i)) var newDF = df.select(selectCols: _*).toDF(columns: _*) // Some business operations here and then write to back to kafka. newDF.write .format("kafka") .option("kafka.bootstrap.servers", "localhost:9092") .option("topic", "topic.ouput") .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";") .option("kafka.sasl.mechanism", "PLAIN") .option("kafka.security.protocol", "SASL_PLAINTEXT") .save() stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) sparkSession.catalog.clearCache() } streamingContext.start() streamingContext.awaitTermination() {code} Output: {code:java} test.topic:6:1345075:4163058:2817983 test.topic:0:1339456:4144190:2804734 test.topic:3:1354266:4189336:2835070 test.topic:7:1353542:4186148:2832606 test.topic:5:1355140:4189071:2833931 test.topic:2:1351162:4173375:2822213 test.topic:1:1352801:4184073:2831272 test.topic:4:1348558:4166749:2818191 () test.topic:6:4163058:4163058:0 test.topic:0:4144190:4144190:0 test.topic:3:4189336:4189336:0 test.topic:7:4186148:4186148:0 test.topic:5:4189071:4189071:0 test.topic:2:4173375:4173375:0 test.topic:1:4184073:4184073:0 test.topic:4:4166749:4166749:0 {code} > Spark Direct Streaming: Read messages of a certain bytes or count in batches > from Kafka is not working. > ------------------------------------------------------------------------------------------------------- > > Key: SPARK-40188 > URL: https://issues.apache.org/jira/browse/SPARK-40188 > Project: Spark > Issue Type: Bug > Components: DStreams > Affects Versions: 3.2.1 > Environment: Spark Version: 3.2.1 > Kafka version: 3.2.0 > > Reporter: Madhav Madhu > Priority: Major > > Spark Kafka consumer is unable to read messages, of a certain size or count > in batches. I have tried few approaches as mentioned in Kafka docs but with > no success. Here is a link to Stack Overflow where I asked the same question > with no response and think this is a possible bug here. Same configuration > works fine when the consumer is a java code. > https://stackoverflow.com/questions/73398533/spark-streaming-context-kafka-consumer-read-messages-of-a-certain-byte-size-in > Here is the consumer code which fetches data from Kafka, > {code:scala} > val streamingContext = new StreamingContext(sparkSession.sparkContext, > Seconds(10)) > val kafkaParams = Map[String, Object]( > "bootstrap.servers" -> "localhost:9092", > "key.deserializer" -> classOf[StringDeserializer], > "value.deserializer" -> classOf[StringDeserializer], > "group.id" -> "test", > "fetch.max.bytes" -> "65536", > "max.partition.fetch.bytes" -> "8192", > "max.poll.records" -> "100", > "auto.offset.reset" -> "latest", > "enable.auto.commit" -> (false: java.lang.Boolean), > "sasl.jaas.config"-> "org.apache.kafka.common.security.plain.PlainLoginModule > required username=\"admin\" password=\"admin\";", > "sasl.mechanism" -> "PLAIN", > "security.protocol" -> "SASL_PLAINTEXT", > ) > val topics = Array("test.topic") > val stream = KafkaUtils.createDirectStream[String, String]( > streamingContext, > PreferConsistent, > Subscribe[String, String](topics, kafkaParams) > ) > stream.foreachRDD { > rdd => > val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges > println(offsetRanges.foreach(a => println(a.topic + ":" + a.partition + ":" > + a.fromOffset + ":" + a.untilOffset + ":" + a.count()))) > val df = rdd.map(a => a.value().split(",")).toDF() > val selectCols = columns.indices.map(i => $"value"(i)) > var newDF = df.select(selectCols: _*).toDF(columns: _*) > // Some business operations here and then write to back to kafka. > > newDF.write > .format("kafka") > .option("kafka.bootstrap.servers", "localhost:9092") > .option("topic", "topic.ouput") > .option("kafka.sasl.jaas.config", > "org.apache.kafka.common.security.plain.PlainLoginModule required > username=\"admin\" password=\"admin\";") > .option("kafka.sasl.mechanism", "PLAIN") > .option("kafka.security.protocol", "SASL_PLAINTEXT") > .save() > stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) > sparkSession.catalog.clearCache() > } > streamingContext.start() > streamingContext.awaitTermination() > {code} > Output: > {code:java} > test.topic:6:1345075:4163058:2817983 > test.topic:0:1339456:4144190:2804734 > test.topic:3:1354266:4189336:2835070 > test.topic:7:1353542:4186148:2832606 > test.topic:5:1355140:4189071:2833931 > test.topic:2:1351162:4173375:2822213 > test.topic:1:1352801:4184073:2831272 > test.topic:4:1348558:4166749:2818191 > () > test.topic:6:4163058:4163058:0 > test.topic:0:4144190:4144190:0 > test.topic:3:4189336:4189336:0 > test.topic:7:4186148:4186148:0 > test.topic:5:4189071:4189071:0 > test.topic:2:4173375:4173375:0 > test.topic:1:4184073:4184073:0 > test.topic:4:4166749:4166749:0 > {code} > I tried different options as followed, > Option 1: > Topic Partition 8 > Streaming Context 1 sec: > "fetch.max.bytes" -> "65536", // 64 Kb > "max.partition.fetch.bytes" -> "8192" // 8Kb > "max.poll.records" -> "100" > DataFrame count which it read from Kafka in the very first batch: 1200000 > Option 2: > Partition 1 > Streaming Context 1 sec > "fetch.max.bytes" -> "65536", > "max.partition.fetch.bytes" -> "8192" > "max.poll.records" -> "100" > Kafka Lag: 126360469 > DataFrame count which it read from Kafka in the very first batch: 126360469. > -- This message was sent by Atlassian Jira (v8.20.10#820010) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org