使用ElasticSearch6.0.0接收器NoSuchMethoderRror BulkProcessor.Builder闪烁

Question

我尝试用elasticsearch（版本为6.0.0)Sink构建一个flink流式单词计数演示。不幸的是出现了以下错误。这似乎是依赖关系。

Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: java.lang.NoSuchMethodError: org.elasticsearch.action.bulk.BulkProcessor.builder(Ljava/util/function/BiConsumer;Lorg/elasticsearch/action/bulk/BulkProcessor$Listener;)Lorg/elasticsearch/action/bulk/BulkProcessor$Builder;
    at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:623)
    at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:123)
    at com.quvideo.xiaoying.flink.elasticsearch.WordCountSinkElasticsearch.main(WordCountSinkElasticsearch.java:68)
Caused by: java.lang.NoSuchMethodError: org.elasticsearch.action.bulk.BulkProcessor.builder(Ljava/util/function/BiConsumer;Lorg/elasticsearch/action/bulk/BulkProcessor$Listener;)Lorg/elasticsearch/action/bulk/BulkProcessor$Builder;
    at org.apache.flink.streaming.connectors.elasticsearch6.Elasticsearch6ApiCallBridge.createBulkProcessorBuilder(Elasticsearch6ApiCallBridge.java:92)
    at org.apache.flink.streaming.connectors.elasticsearch6.Elasticsearch6ApiCallBridge.createBulkProcessorBuilder(Elasticsearch6ApiCallBridge.java:45)
    at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.buildBulkProcessor(ElasticsearchSinkBase.java:353)
    at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.open(ElasticsearchSinkBase.java:297)
    at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
    at org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:424)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:290)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
    at java.lang.Thread.run(Thread.java:748)

Process finished with exit code 1

我的elasticsearch集群是6.0.0,flink依赖项如下

<properties>
    <flink.version>1.6.0</flink.version>
    <elastic>6.0.0</elastic>
</properties>
<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>

<dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>transport</artifactId>
        <version>${elastic}</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-client</artifactId>
        <version>${elastic}</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>${elastic}</version>
    </dependency>

    public static ElasticsearchSink<WordWithCount> getEsSink(){
        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("10.0.35.148", 9200, "http"));
        ElasticsearchSink.Builder<WordWithCount> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<WordWithCount>() {
                    public IndexRequest createIndexRequest(WordWithCount element) {
                        Map<String, Object> json = new HashMap<>();
                        json.put("word", element.word);
                        json.put("count", element.count);

                        return Requests.indexRequest()
                                .index("wordcount_idx")
                                .type("test_type")
                                .source(json);
                    }

                    @Override
                    public void process(WordWithCount element, RuntimeContext ctx, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(element));
                    }
                }
        );
        esSinkBuilder.setBulkFlushMaxActions(1);
        return esSinkBuilder.build();
    }

有关详细信息，此错误在ElasticSearch6ApicallBridge.java方法中触发

@Override
     public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient client, BulkProcessor.Listener listener) {
        return BulkProcessor.builder(client::bulkAsync, listener);// error message "client is not a functional interface."
     }

谢谢你

Answer 1

我提交了https://issues.apache.org/jira/browse/flink-10173。作为一种变通方法，您可以简单地添加：

<dependency>
  <groupId>org.elasticsearch.client</groupId>
  <artifactId>elasticsearch-rest-high-level-client</artifactId>
  <version>6.3.1</version>
</dependency>