使用Flink读取Elasticsearch（ES）数据需要使用Flink的DataStream API结合ElasticsearchSinkFunction和ElasticsearchSourceFunction来实现。

下面是一个简单的示例代码，演示了如何在Flink中读取ES数据：

import org.apache.flink.api.common.functions.RuntimeContext;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSourceFunction;import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSource;import org.apache.flink.streaming.connectors.elasticsearch6.RestClientFactory;import org.apache.flink.streaming.connectors.elasticsearch6.RestClientFactoryImpl;import org.apache.flink.streaming.connectors.elasticsearch6.index.IndexRequestBuilder;import org.apache.flink.streaming.connectors.elasticsearch6.index.IndexRequestBuilderProvider;import org.apache.flink.streaming.connectors.elasticsearch6.index.IndexRequestBuilderFactory;import org.apache.flink.streaming.connectors.elasticsearch6.index.IndexRequestParameters;import org.apache.flink.streaming.connectors.elasticsearch6.index.IndexRequestParametersProvider;import org.apache.http.HttpHost;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.Requests;import org.elasticsearch.client.RestClientBuilder;import org.elasticsearch.common.xcontent.XContentType;import java.util.ArrayList;import java.util.List;public class ReadFromESExample {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置ES连接的地址List<HttpHost> httpHosts = new ArrayList<>();httpHosts.add(new HttpHost("localhost", 9200, "http"));ElasticsearchSourceFunction<String> sourceFunction = new ElasticsearchSource<>(httpHosts, "index_name", "_doc", new ElasticsearchSourceFunction<String>() {@Overridepublic IndexRequest createIndexRequest(String element) {return Requests.indexRequest().index("index_name").type("_doc").source(element, XContentType.JSON);}@Overridepublic void processElement(String element, RuntimeContext ctx, RequestIndexer indexer) {indexer.add(createIndexRequest(element));}});DataStream<String> dataStream = env.addSource(sourceFunction);dataStream.print();env.execute("Read from Elasticsearch Example");}}

需要注意的是，要使用ElasticsearchSinkFunction和ElasticsearchSourceFunction需要添加相应的依赖，具体可以参考官方文档或者搜索相关资料。