Flink写入es6.8

要将Flink数据写入Elasticsearch 6.8版本，需要使用Elasticsearch Hadoop库。以下是步骤：

1. 在Flink项目中添加Elasticsearch Hadoop依赖：

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>6.8.0</version>
</dependency>

2. 创建一个ElasticsearchOutputFormat实例并设置连接信息和索引名称：

Map<String, String> config = new HashMap<>();
config.put("cluster.name", "elasticsearch");
config.put("bulk.flush.max.actions", "1");
config.put("es.nodes", "localhost:9200");
config.put("es.index.auto.create", "true");

ElasticsearchOutputFormat.Builder builder = new ElasticsearchOutputFormat.Builder(new SimpleElasticsearchSerializer(), config);
builder.setIndex("myindex");

ElasticsearchOutputFormat outputFormat = builder.build();

3. 将Flink的DataStream转换成ElasticsearchOutputFormat需要的JavaRDD格式：

DataStream<String> stream = ...;
JavaDStream<String> javaDStream = JavaDStream.fromDStream(stream.getWrappedStream(), scala.reflect.ClassManifestFactory$.MODULE$.fromClass(String.class));

JavaRDD<String> javaRDD = JavaDStream.toRDD(javaDStream, scala.reflect.ClassManifestFactory$.MODULE$.fromClass(String.class));

4. 将JavaRDD写入Elasticsearch：

javaRDD.foreachPartition(new VoidFunction<Iterator<String>>() {
    @Override
    public void call(Iterator<String> iterator) throws Exception {
        ElasticsearchOutputFormat.writeRecordsToElasticsearch(iterator, outputFormat);
    }
});

以上是Flink写入Elasticsearch 6.8版本的基本步骤，具体实现可以根据自己的需求进行调整和优化。