将数据插入弹性搜索时出现问题答案

【问题标题】：Issue while inserting data to elastic search将数据插入弹性搜索时出现问题
【发布时间】：2014-01-02 08:32:16
【问题描述】：

我正在尝试学习如何使用 logstash、redis、elasticsearch 和 Kibana 等工具。

我已经成功完成了工具logstash、redis、elasticsearch和Kibana的配置。读取日志文件并能够将其正确插入到弹性搜索中。

现在我的目标是使用 Java 并将字符串存储到 elasticsearch。我已经配置了在同一台机器上运行的弹性搜索（Linux Ubuntu）。

使用的代码如下：

Settings settings = ImmutableSettings.settingsBuilder()
                        .put("client.transport.sniff", false).build();

Client client = new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress("192.168.160.242", 9300));

                //Convert to JSON and index
                IndexResponse response = client.prepareIndex("twitter", "tweet", "1")
                        .setSource(jsonBuilder()
                                    .startObject()
                                        .field("user", "kimchy")
                                        .field("postDate", new Date())
                                        .field("message", "trying out Elastic     Search")
                                    .endObject()
                                  )
                        .execute()
                        .actionGet();

但是在使用弹性搜索控制台时，我观察到以下错误：

[2014-01-02 14:00:20,410][WARN ][transport.netty          ] [Kine, Benedict] exception caught on transport layer [[id: 0xaa3df382, /**192.168.160.242:38045** :> /**192.168.160.242:9300**]], closing connection
java.io.StreamCorruptedException: invalid internal transport message format
        at org.elasticsearch.transport.netty.SizeHeaderFrameDecoder.decode(SizeHeaderFrameDecoder.java:27)
        at org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:425)
        at org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303)
        at org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
        at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
        at org.elasticsearch.common.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
        at org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:74)
        at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
        at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559)
        at org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:268)
        at org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:255)
        at org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88)
        at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108)
        at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:318)
        at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89)
        at org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
        at org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
        at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:724)

如果我遗漏了什么，请告诉我？

【问题讨论】：

标签： java elasticsearch

【解决方案1】：

这是非常随机的建议，但您是否使用自定义分类名称设置了弹性搜索？如果是，也许尝试将其添加到设置中？

【讨论】：

我尝试将集群名称更改为“abcd”，但在更改名称后，我观察到logstash shipper 将集群名称设为“abcd”，而logstash indexer 仍将其设为“elasticsearch” '（集群的默认名称）。此外，由于我已将“client.transport.sniff”指定为 false，因此我知道这不会引起任何问题。请注意，我已经在同一台 Ubuntu 机器上配置了所有这些。一切都在同一台机器上。

【解决方案2】：

检查您正在使用的 java 客户端和 elasticsearch 服务器的版本。

根据this 指南，它们应该相同

【讨论】：

【解决方案3】：

您可以尝试将集群名称而不是 client.transport.sniff" 设置为 false

要查看您当前的 cluser_name：

curl -XGET 主机名:port/_cluster/nodes?pretty=true

并使用

创建连接

Settings settings = ImmutableSettings.settingsBuilder()
    .put("cluster.name", "test").build();
Client client = new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress("hostname", port));

【讨论】：