Flink与ElasticSearch集成

在生产中，经常会使用Flink和ES集成，将Flink处理后的数据存放在ES之中，我们了解Flink是个实时处理的框架，其输出Sink可以由用户自定义输出，所以就可以将Flink处理后的数据通过自定义的Sink输出到ES中。

1. 首先要添加相关的依赖

   
       org.apache.flink
       flink-scala_2.12
       1.12.0
   
   
   
       org.apache.flink
       flink-streaming-scala_2.12
       1.12.0
   
   
   
       org.apache.flink
       flink-clients_2.12
       1.12.0
   
   
   
       org.apache.flink
       flink-connector-elasticsearch7_2.11
       1.12.0
   
   

2. 创建flink的执行环境

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   

3. 使用socket当作数据源

   DataStreamSource source = env.socketTextStream("localhost", 9999);
   
   

4. 创建ES连接

   List httpHosts = new ArrayList<>();
   httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
   ElasticsearchSink.Builder esSinkBuilder = new ElasticsearchSink.Builder<>(
   	httpHosts, new ElasticsearchSinkFunction() {
   		public IndexRequest createIndexRequest(String element) {
   			Map json = new HashMap<>();
                json.put("data", element);
   			return Requests.indexRequest()
   			.index("my-index")
   			.source(json);
               }
               @Override
               public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
               indexer.add(createIndexRequest(element));
   		}
   	}
   );
   

5. 将ES作为Sink

   source.addSink(esSinkBuilder.build());
   

6. 执行

   env.execute("flink-es");