Flink写入数据到Elasticsearch示例

        <!-- elasticsearch connector -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
            <version>1.11.0</version>
        </dependency>

        <!-- jackson -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.11.1</version>
        </dependency>

        List<HttpHost> elsearchHosts = new ArrayList<>(); 
        elsearchHosts.add(new HttpHost("192.168.32.36", 9200, "http"));
        elsearchHosts.add(new HttpHost("192.168.32.37", 9200, "http"));
        elsearchHosts.add(new HttpHost("192.168.32.38", 9200, "http"));

        ObjectMapper mapper = new ObjectMapper(); // jaskson ObjectMapper

        ElasticsearchSink.Builder<ResultCollector> esSinkBuilder = new ElasticsearchSink.Builder<>( // ResultCollector 是你要保存的对象类型，替换即可
                elsearchHosts,
                new ElasticsearchSinkFunction<ResultCollector>() {

                    private static final long serialVersionUID = -6797861015704600807L;

                    public IndexRequest createIndexRequest(ResultCollector collector) throws Exception {
                        return Requests.indexRequest()
                                .index("flink-test-index") // 设置Index
                                .id(collector.getId()) // 设置ID
　　　　　　　　　　　　　　　　　　　 // 这里要特别注意需要传map
                                .source(mapper.readValue(mapper.writeValueAsString(collector), Map.class));
                    }

                    @SneakyThrows
                    @Override
                    public void process(ResultCollector collector, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                        requestIndexer.add(createIndexRequest(collector));
                    }
                }
        );

        esSinkBuilder.setBulkFlushMaxActions(1);
        esSinkBuilder.setFailureHandler(new RetryRejectedExecutionFailureHandler());

        esSinkBuilder.setRestClientFactory((RestClientFactory) restClientBuilder -> {
            Header[] headers = new BasicHeader[]{new BasicHeader("Content-Type", "application/json")};
            restClientBuilder.setDefaultHeaders(headers);
        });

        //stream.addSink(esSinkBuilder.build());