如何在Spark structured streaming中读取特定的Kafka分区

val spark = SparkSession.builder
  .appName("KafkaStructuredStreaming")
  .master("local[*]")
  .getOrCreate()

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

val kafkaParams = Map(
  "kafka.bootstrap.servers" -> "kafka-server:9092",
  "subscribe" -> "topic-name",
  "startingOffsets" -> "earliest",
  "failOnDataLoss" -> "false"
)

val specificPartition = 0

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaParams("kafka.bootstrap.servers"))
  .option("subscribe", kafkaParams("subscribe"))
  .option("startingOffsets", kafkaParams("startingOffsets"))
  .option("failOnDataLoss", kafkaParams("failOnDataLoss"))
  .load()

val filteredStream = kafkaStream
  .filter(col("partition") === specificPartition)

val processedStream = filteredStream
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  // 进行其他操作，如转换、聚合等

val query = processedStream
  .writeStream
  .outputMode("append")
  .format("console")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()

query.awaitTermination()