Ufdjrw

I am consuming data from kafka topic through spark structured streaming, the topic has 3 partitions. As Spark structured streaming does not allow you to explicitly provide group.id and assigns some random id to consumer, I tried to check the consumer group id's using below kafka command

./kafka-consumer-groups.sh --bootstrap-server kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092 --list

output
 spark-kafka-source-054e8dac-bea9-46e8-9374-8298daafcd23--1587684247-driver-0
 spark-kafka-source-756c08e8-6a84-447c-8326-5af1ef0412f5-209267112-driver-0
 spark-kafka-source-9528b191-4322-4334-923d-8c1500ef8194-2006218471-driver-0

Below are my questions

1) Why does it create 3 consumer groups? Is it because of 3 partitions?

2) Is there any way I can get these consumer group names in spark application?

3) Even though my spark application was still running, after some time these group names didn't show up in consumer groups list. Is this because all the data was consumed by spark application and there was no more data in that kafka topic?

4) If my assumption is right about point 3, will it create new consumer group id if new data arrives or the name of consumer group will remain same?

Below is my read stream

 val inputDf = spark.readStream
 .format("kafka")
 .option("kafka.bootstrap.servers", brokers)
 .option("subscribe", topic)
 // .option("assign"," ""+topic+"":[0]") 
 .option("startingOffsets", "earliest")
 .option("maxOffsetsPerTrigger", 60000)
 .load()

I have 3 writestreams in application as below

 val df = inputDf.selectExpr("CAST(value AS STRING)","CAST(topic AS STRING)","CAST (partition AS INT)","CAST (offset AS INT)","CAST (timestamp AS STRING)") 
 val df1 = inputDf.selectExpr("CAST (partition AS INT)","CAST (offset AS INT)","CAST (timestamp AS STRING)")

//First stream
 val checkpoint_loc1= "/warehouse/test_duplicate/download/chk1"
 df1.agg(min("offset"), max("offset"))
 .writeStream
 .foreach(writer)
 .outputMode("complete")
 .option("checkpointLocation", checkpoint_loc1).start()
val result = df.select(
df1("result").getItem("_1").as("col1"),
df1("result").getItem("_2").as("col2"),
df1("result").getItem("_5").as("eventdate"))
val distDates = result.select(result("eventdate")).distinct

//Second stream
val checkpoint_loc2= "/warehouse/test_duplicate/download/chk2" 
distDates.writeStream.foreach(writer1)
 .option("checkpointLocation", checkpoint_loc2).start() 

//Third stream
val kafkaOutput =result.writeStream
 .outputMode("append")
 .format("orc")
 .option("path",data_dir)
 .option("checkpointLocation", checkpoint_loc3)
 .start()

The streaming query is used only once in the code and there are no joins.

Execution plan

== Parsed Logical Plan ==
 StreamingRelationV2 org.apache.spark.sql.kafka010.KafkaSourceProvider@109e44ba, kafka, Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092), [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13], StreamingRelation DataSource(org.apache.spark.sql.SparkSession@593197cb,kafka,List(),None,List(),None,Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092),None), kafka, [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6]

== Analyzed Logical Plan ==
key: binary, value: binary, topic: string, partition: int, offset: bigint, timestamp: timestamp, timestampType: int
StreamingRelationV2 org.apache.spark.sql.kafka010.KafkaSourceProvider@109e44ba, kafka, Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092), [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13], StreamingRelation DataSource(org.apache.spark.sql.SparkSession@593197cb,kafka,List(),None,List(),None,Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092),None), kafka, [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6]

== Optimized Logical Plan ==
StreamingRelationV2 org.apache.spark.sql.kafka010.KafkaSourceProvider@109e44ba, kafka, Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092), [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13], StreamingRelation DataSource(org.apache.spark.sql.SparkSession@593197cb,kafka,List(),None,List(),None,Map(maxOffsetsPerTrigger -> 60000, startingOffsets -> earliest, subscribe -> downloading, kafka.bootstrap.servers -> kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092),None), kafka, [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6]

== Physical Plan ==
StreamingRelation kafka, [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]

1) Why does it create 3 consumer groups? Is it because of 3 partitions?

Certainly not. It is just a coincidence. You seem to have run the application 3 times already and the topic has 3 partitions.

Let's start over to back it up.

I deleted all consumer groups to make sure that we start afresh.

$ ./bin/kafka-consumer-groups.sh --list --bootstrap-server :9092
spark-kafka-source-cd8c4070-cac0-4653-81bd-4819501769f9-1567209638-driver-0
spark-kafka-source-6a60f735-f05c-49e4-ae88-a6193e7d4bf8--525530617-driver-0

$ ./bin/kafka-consumer-groups.sh --bootstrap-server :9092 --delete --group spark-kafka-source-cd8c4070-cac0-4653-81bd-4819501769f9-1567209638-driver-0
Deletion of requested consumer groups ('spark-kafka-source-cd8c4070-cac0-4653-81bd-4819501769f9-1567209638-driver-0') was successful.

$ ./bin/kafka-consumer-groups.sh --bootstrap-server :9092 --delete --group spark-kafka-source-6a60f735-f05c-49e4-ae88-a6193e7d4bf8--525530617-driver-0
Deletion of requested consumer groups ('spark-kafka-source-6a60f735-f05c-49e4-ae88-a6193e7d4bf8--525530617-driver-0') was successful.

$ ./bin/kafka-consumer-groups.sh --list --bootstrap-server :9092
// nothing got printed out

I created a topic with 5 partitions.

$ ./bin/kafka-topics.sh --create --zookeeper :2181 --topic jacek-five-partitions --partitions 5 --replication-factor 1
Created topic "jacek-five-partitions".

$ ./bin/kafka-topics.sh --describe --zookeeper :2181 --topic jacek-five-partitions
Topic:jacek-five-partitions PartitionCount:5 ReplicationFactor:1 Configs:
 Topic: jacek-five-partitions Partition: 0 Leader: 0 Replicas: 0 Isr: 0
 Topic: jacek-five-partitions Partition: 1 Leader: 0 Replicas: 0 Isr: 0
 Topic: jacek-five-partitions Partition: 2 Leader: 0 Replicas: 0 Isr: 0
 Topic: jacek-five-partitions Partition: 3 Leader: 0 Replicas: 0 Isr: 0
 Topic: jacek-five-partitions Partition: 4 Leader: 0 Replicas: 0 Isr: 0

The code I use is as follows:

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger

object SparkApp extends App 

 val spark = SparkSession.builder.master("local[*]").getOrCreate()
 import spark.implicits._
 val q = spark
 .readStream
 .format("kafka")
 .option("startingoffsets", "latest")
 .option("subscribe", "jacek-five-partitions")
 .option("kafka.bootstrap.servers", ":9092")
 .load
 .select($"value" cast "string")
 .writeStream
 .format("console")
 .trigger(Trigger.ProcessingTime("30 seconds"))
 .start
 q.awaitTermination()

When I run the above Spark Structured Streaming application, I have just one consumer group created.

$ ./bin/kafka-consumer-groups.sh --list --bootstrap-server :9092
spark-kafka-source-380da653-c829-45db-859f-09aa9b37784d-338965656-driver-0

And that makes sense since all the Spark processing should use as many Kafka consumers as there are partitions, but regardless of the number of consumers, there should only be one consumer group (or the Kafka consumers would consume all records and there would be duplicates).

2) Is there any way I can get these consumer group names in spark application?

There is no public API for this so the answer is no.

You could however "hack" Spark and go below the public API up to the internal Kafka consumer that uses this line:

val uniqueGroupId = s"spark-kafka-source-$UUID.randomUUID-$metadataPath.hashCode"

Or even this line to be precise:

val kafkaOffsetReader = new KafkaOffsetReader(
 strategy(caseInsensitiveParams),
 kafkaParamsForDriver(specifiedKafkaParams),
 parameters,
 driverGroupIdPrefix = s"$uniqueGroupId-driver")

Just find the KafkaMicroBatchReader for the Kafka data source, request it for the KafkaOffsetReader that knows groupId. That seems doable.

Even though my spark application was still running, after some time these group names didn't show up in consumer groups list. Is this because all the data was consumed by spark application and there was no more data in that kafka topic?

Could that be related to KIP-211: Revise Expiration Semantics of Consumer Group Offsets which says:

The offset of a topic partition within a consumer group expires when the expiration timestamp associated with that partition is reached. This expiration timestamp is usually affected by the broker config offsets.retention.minutes, unless user overrides that default and uses a custom retention.

4) will it create new consumer group id if new data arrives or the name of consumer group will remain same?

Will remain same.

Moreover, the consumer group must not be deleted when at least one consumer from the group is active.