如何在Apache Flink中 join到elasticsearch 5.x


0

ApacheFlink版本是1.1.3,elasticsearch是5.1.1。

flink文档仅 explainelasticsearch 2.x API(flink-connector-elasticsearch2_2.1.1)

elasticsearch 5.x是否没有flink join器API?

我试图将此版本用于elasticsearch 5.x,但遇到了如下错误

Flink例外

01/03/2017 20:01:21 Job execution switched to status FAILING.
java.lang.RuntimeException: Client is not connected to any Elasticsearch nodes!
    at org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink.open(ElasticsearchSink.java:172)
    at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:376)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:256)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:585)
    at java.lang.Thread.run(Thread.java:745)

elasticsearch日志

[2017-01-03T20:01:21,642][WARN ][o.e.h.n.Netty4HttpServerTransport] [7CTs2-R] caught exception while handling client http traffic, closing connection [id: 0xbce51ef2, L:/127.0.0.1:9200 - R:/127.0.0.1:58429]
java.io.IOException: The current connection has been interrupted by a remote host
    at sun.nio.ch.SocketDispatcher.read0(Native Method) ~[?:?]
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43) ~[?:?]
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) ~[?:?]
    at sun.nio.ch.IOUtil.read(IOUtil.java:197) ~[?:?]
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380) ~[?:?]
    at io.netty.buffer.PooledHeapByteBuf.setBytes(PooledHeapByteBuf.java:261) ~[netty-buffer-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1100) ~[netty-buffer-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:366) ~[netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:118) [netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:651) [netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:536) [netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:490) [netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:450) [netty-transport-4.1.6.Final.jar:4.1.6.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:873) [netty-common-4.1.6.Final.jar:4.1.6.Final]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_91]

2 答案


0

ApacheFlink1.1.3不包括Elasticsearch 5.x的 join器。

有一些 job 正在进行中(JIRA,Github PR)这样的 join器,但它还没有被添加到Flink代码库中。

您可以尝试从拉请求作者的存储库构建 join器。


0

您应该能够使用下面的依赖项并使其 job 。

请注意,flink发行版似乎对我Mac上的elasticsearch5依赖项不起作用。所以如果你能将它降级到1.3快照,它应该可以 job 。

我能够让它 job 和张贴事件到 Elasticsearch5.4。

更改以下内容

在pom.xml文件

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-elasticsearch5_2.10</artifactId>
        <version>${flink.version}</version>
    </dependency>

在使用elasticsearch2的Java代码中,将其更改为5,如下所示

进口org.apache.flink网站.streaming.connectors.elasticsearch5.ElasticsearchSink;

在扩展ElasticsearchSinkFunction的类中,将此方法(如在PopularPlaceInserter中)与现有的进程方法一起添加。请确保更改了elasticsearch的索引和 map名称类型,并且应该能够运行该程序。

    public IndexRequest createIndexRequest(String element) {
        Map<String, String> json = new HashMap<>();
        json.put("data", element);

    return Requests.indexRequest()
            .index("nyc-idx")
            .type("popular-locations")
            .source(json);
}


我来回答

写文章

提问题

面试题