本文的基础环境可以参考flink 1.10.1 java版本wordcount演示 (nc + socket)，在此基础上增加输出结果到elasticsearch。

1. 添加依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch7_2.11</artifactId><version>1.10.1</version><!--<scope>${scope}</scope>-->
</dependency>

2. 测试代码

package com.demo.redis;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;import java.util.ArrayList;
import java.util.HashMap;/*** flink结果写入elasticsearch*/
public class FlinkESSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStream<String> dataStream = env.socketTextStream("192.168.0.181",9000);SingleOutputStreamOperator<String> flatMap = dataStream.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String value, Collector<String> out) throws Exception {String[] strings = value.split(" ");for (String s : strings) {out.collect(s);}}});SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {return Tuple2.of(value, 1);}});SingleOutputStreamOperator<Tuple2<String, Integer>> sum = map.keyBy("f0").sum(1);DataStream<String> result = sum.map(new MapFunction<Tuple2<String, Integer>, String>() {@Overridepublic String map(Tuple2<String, Integer> data) throws Exception {return data.f0 + ":" + data.f1;}});// 配置es参数ArrayList<HttpHost> httpHosts = new ArrayList<>();httpHosts.add(new HttpHost("localhost", 9200, "http"));ElasticsearchSink.Builder<String> esBuilder = new ElasticsearchSink.Builder<String>(httpHosts, new MyEsSinkFunction());esBuilder.setBulkFlushMaxActions(1);result.addSink(esBuilder.build());result.print();env.execute();}public static class MyEsSinkFunction implements ElasticsearchSinkFunction<String>{@Overridepublic void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {// 定义写入的数据HashMap<String, String> dataSouce = new HashMap<>();String timeStamp = "" + System.currentTimeMillis();dataSouce.put(timeStamp, s);// 创建es请求IndexRequest indexRequest = Requests.indexRequest().index("flink-es").source(dataSouce);try {requestIndexer.add(indexRequest);}catch (Exception e){e.printStackTrace();}}}}

设置只有有记录时，立即写入ES

esBuilder.setBulkFlushMaxActions(1);

3. 启动elasticsearch服务

启动elasticsearch服务以后，使用浏览器访问，出现如下输出，表示elasticsearch服务启动成功。

4. 输出结果

启动程序，在nc输入以下测试字符串，

 可以在idea控制台看到输出的数据。

通过浏览器，可以看到写入elasticsearch的数据: