如何使用spark结构流在elasticsearch接收器中设置动态文档id

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.elasticsearch.spark.sql._

val spark = SparkSession.builder()
  .appName("Spark Elasticsearch Example")
  .master("local[*]")
  .config("spark.es.nodes", "localhost")
  .config("spark.es.port", "9200")
  .getOrCreate()

// 从数据源读取数据，假设数据源为Kafka
val data: DataFrame = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic")
  .load()

// 将数据写入Elasticsearch
data.writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    // 设置动态文档ID
    batchDF.write
      .format("org.elasticsearch.spark.sql")
      .option("es.resource", "index/type")
      .option("es.mapping.id", "id") // 设置动态文档ID字段
      .mode("append")
      .save()
  }
  .start()
  .awaitTermination()

<dependency>
  <groupId>org.elasticsearch</groupId>
  <artifactId>elasticsearch-spark-20_2.11</artifactId>
  <version>7.15.0</version>
</dependency>