阅读量:0
使用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 httpHosts = new ArrayList<>(); httpHosts.add(new HttpHost("localhost", 9200, "http")); ElasticsearchSourceFunction sourceFunction = new ElasticsearchSource<>(httpHosts, "index_name", "_doc", new ElasticsearchSourceFunction() { @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 dataStream = env.addSource(sourceFunction); dataStream.print(); env.execute("Read from Elasticsearch Example"); } } 需要注意的是，要使用ElasticsearchSinkFunction和ElasticsearchSourceFunction需要添加相应的依赖，具体可以参考官方文档或者搜索相关资料。
