如何为Spark结构化流媒体编写ElasticsearchSink

import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.sql.{ForeachWriter, Row, SparkSession}
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.{RequestOptions, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.common.xcontent.json.JsonXContent

class ElasticsearchSink(elasticsearchHost: String, elasticsearchPort: Int, index: String, `type`: String) extends ForeachWriter[Row] {
  private var client: RestHighLevelClient = _

  override def open(partitionId: Long, version: Long): Boolean = {
    // 创建Elasticsearch连接
    client = new RestHighLevelClient(RestClient.builder(new HttpHost(elasticsearchHost, elasticsearchPort)))
    true
  }

  override def process(row: Row): Unit = {
    // 将数据写入Elasticsearch
    val json = JsonXContent.contentBuilder().startObject()
    for (i <- 0 until row.length) {
      json.field(row.schema.fieldNames(i), row.get(i))
    }
    json.endObject()
    val request = new IndexRequest(index, `type`).source(json, XContentType.JSON)
    client.index(request, RequestOptions.DEFAULT)
  }

  override def close(errorOrNull: Throwable): Unit = {
    // 关闭Elasticsearch连接
    if (client != null) {
      client.close()
    }
  }
}

object Main {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Spark Elasticsearch Sink")
      .master("local[*]")
      .getOrCreate()

    val data = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "topic")
      .load()

    val query = data.writeStream
      .foreach(new ElasticsearchSink("localhost", 9200, "index", "type"))
      .start()

    query.awaitTermination()
  }
}