Apache Flink 1.4中的Elasticsearch 5连接器

Question

我是一个新手，我正在尝试使用Elasticsearch Connector玩，但没有成功。我在Docker上安装了Elasticsearch 5.5，Kibana和Flink 1.4。

这是我的代码：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.11</artifactId>
        <version>1.4.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.4.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>1.4.0</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-rabbitmq_2.11</artifactId>
        <version>1.4.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-elasticsearch5_2.11</artifactId>
        <version>1.4.0</version>
    </dependency>
</dependencies>

Java代码是：

DataStream<String> consumer = .....
    consumer.print();
    Map<String, String> config = new HashMap<>();
    config.put("cluster.name", "docker-cluster");
    config.put("bulk.flush.max.actions", "1");
    List<InetSocketAddress> transports = new ArrayList<>();
    transports.add(new InetSocketAddress("127.0.0.1", 9300));
    consumer.addSink(new ElasticsearchSink<>(config, transports, new ElasticsearchSinkFunction<String>() {
        public IndexRequest createIndexRequest(String element) {
            Map<String, String> json = new HashMap<>();
            json.put("id", "2");
            return Requests.indexRequest()
                    .index("signinindex").type("signin").source(json);
        }
        public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
            indexer.add(createIndexRequest(element));
        }
    }));
    env.execute();

但是，当我尝试在IDE中提交新工作时，它似乎一直在等待，所以我想知道Elasticsearch 5.5是否有一些与Flink冲突的问题。非常感谢。

[更新]：我对ES 5.1.2进行了一些测试，现在在提交罐子时，我会收到以下例外(我也尝试使用所有连接器升级到1.4.1(：

    java.util.concurrent.CompletionException: org.apache.flink.util.FlinkException: Could not run the jar.
    at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJsonRequest$0(JarRunHandler.java:90)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not run the jar.
    ... 9 more
Caused by: java.lang.RuntimeException: Could not look up the main(String[]) method from the class ProtoSignInMap: org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction
    at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:495)
    at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:216)
    at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:147)
    at org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:83)
    at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJsonRequest$0(JarRunHandler.java:69)
    ... 8 more
Caused by: java.lang.NoClassDefFoundError: org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction
    at java.lang.Class.getDeclaredMethods0(Native Method)
    at java.lang.Class.privateGetDeclaredMethods(Class.java:2701)
    at java.lang.Class.privateGetMethodRecursive(Class.java:3048)
    at java.lang.Class.getMethod0(Class.java:3018)
    at java.lang.Class.getMethod(Class.java:1784)
    at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:489)
    ... 12 more
Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 18 more

显然存在主要方法。

Answer 1

flink当前支持Elasticsearch至版本5.1.x。有关更多信息，请参见：https：//issues.apache.org/jira/browse/flink-7386