使用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>() {
            @Override
            public IndexRequest createIndexRequest(String element) {
                return Requests.indexRequest()
                        .index("index_name")
                        .type("_doc")
                        .source(element, XContentType.JSON);
            }

            @Override
            public 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需要添加相應的依賴，具體可以參考官方文檔或者搜索相關資料。