如何将JSON JavaInputDStream到ElasticSearch JAVA

import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;

// 创建JavaInputDStream对象，读取JSON数据流
JavaInputDStream<String> jsonStream = ...;

// 解析JSON数据流，将其转换为Java对象
JavaDStream<MyObject> objectStream = jsonStream.map(json -> {
    ObjectMapper objectMapper = new ObjectMapper();
    return objectMapper.readValue(json, MyObject.class);
});

// 将Java对象索引到Elasticsearch中
objectStream.foreachRDD(rdd -> {
    rdd.foreachPartition(objects -> {
        RestHighLevelClient client = createElasticsearchClient();

        while (objects.hasNext()) {
            MyObject object = objects.next();
            IndexRequest request = new IndexRequest("index_name")
                    .source(objectToJson(object), XContentType.JSON);
            client.index(request, RequestOptions.DEFAULT);
        }

        client.close();
    });
});

// 启动流处理
streamingContext.start();
streamingContext.awaitTermination();