Apache Flink (v1.6.0) 验证 Elasticsearch Sink (v6.4)

Question

我正在使用 Apache Flink v1.6.0 并且我正在尝试写入 Elasticsearch v6.4.0，它托管在Elastic Cloud 中。我在对 Elastic Cloud 集群进行身份验证时遇到问题。

我已经能够使用以下代码让 Flink 写入本地 Elasticsearch v6.4.0 节点，该节点没有加密：

/*

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

// finally, build and add the sink to the job's pipeline

stream.addSink(esSinkBuilder.build());

然而，当我尝试添加验证到代码库中，作为记录在这里的弗林克文档和这里对应Elasticsearch Java文档上。看起来像这样：

// provide a RestClientFactory for custom configuration on the internally created REST client

Header[] defaultHeaders = new Header[]{new BasicHeader("username", "password")};

esSinkBuilder.setRestClientFactory(

        restClientBuilder -> {

            restClientBuilder.setDefaultHeaders(defaultHeaders);

        }

);

谁能帮忙指出我哪里出错了？

Answer 1

override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {&nbsp; &nbsp; &nbsp; &nbsp; // TODO Additional rest client args go here - authentication headers for secure connections etc...&nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; })我希望这可以帮助你。