flink 写入到es_Apache Flink（v1.6.0）对Elasticsearch Sink（v6.4）进行身份验证

小编典典

我能看弗林克例子后去解决它在这里和Elasticsearch文档这里。

原来，我试图在上面设置错误的配置：

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 Sink的完整片段：

/*

Elasticsearch Configuration

*/

List httpHosts = new ArrayList<>();

httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

// use a ElasticsearchSink.Builder to create an ElasticsearchSink

ElasticsearchSink.Builder esSinkBuilder = new ElasticsearchSink.Builder<>(

httpHosts,

new ElasticsearchSinkFunction() {

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());

希望这对卡在同一地方的其他人有所帮助！

2020-06-22