Apache Flink(v1.6.0)验证ElasticSearch接收器(v6.4)

override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {
        // TODO Additional rest client args go here - authentication headers for secure connections etc...
      }
    })

我希望这能对你有所帮助。

我看了一个打火的例子,就知道了。 here 以及ElasticSearch文档 here .

结果我试图在上面设置错误的配置:

restClientBuilder.setDefaultHeaders(...);

不是实际需要的设置:

restClientBuilder.setHttpClientConfigCallback(...);

一旦您使用了正确的自定义配置,剩下的就非常简单了。所以我错过的部分是:

// provide a RestClientFactory for custom configuration on the internally created REST client
esSinkBuilder.setRestClientFactory(
    restClientBuilder -> {
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {

                // elasticsearch username and password
                CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("$USERNAME", "$PASSWORD"));

                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        });
    }
);

最后是一个完整的ElasticSearch接收器的片段:

/*
    Elasticsearch Configuration
*/
List<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

// use a ElasticsearchSink.Builder to create an ElasticsearchSink
ElasticsearchSink.Builder<ObjectNode> esSinkBuilder = new ElasticsearchSink.Builder<>(
        httpHosts,
        new ElasticsearchSinkFunction<ObjectNode>() {
            private IndexRequest createIndexRequest(ObjectNode payload) {

                // remove the value node so the fields are at the base of the json payload
                JsonNode jsonOutput = payload.get("value");

                return Requests.indexRequest()
                        .index("raw-payload")
                        .type("payload")
                        .source(jsonOutput.toString(), XContentType.JSON);
            }

            @Override
            public void process(ObjectNode payload, RuntimeContext ctx, RequestIndexer indexer) {
                indexer.add(createIndexRequest(payload));
            }
        }
);

// set number of events to be seen before writing to Elasticsearch
esSinkBuilder.setBulkFlushMaxActions(1);

// provide a RestClientFactory for custom configuration on the internally created REST client
esSinkBuilder.setRestClientFactory(
    restClientBuilder -> {
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {

                // elasticsearch username and password
                CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("$USERNAME", "$PASSWORD"));

                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        });
    }
);

// finally, build and add the sink to the job's pipeline
stream.addSink(esSinkBuilder.build());

我希望这能帮助其他被困在同一个地方的人!