使用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需要添加相应的依赖,具体可以参考官方文档或者搜索相关资料。
辰迅云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
推荐阅读: java中flink的应用场景有哪些