有 Java 编程相关的问题?

你可以在下面搜索框中键入要查询的问题!

配置Logstash以从socket接收数据,并将其插入java中的Elasticsearch

我想将数据直接注入ElasticSearch以执行一些性能测试。我的第一个想法是为每个文档创建JSON文件,并将这些文件导入ElasticSearch,但这会花费太长时间。我测试了110K个文件,创建这些文件只花了18分钟,我需要5500万个文档——这是我测试的500倍。快速计算:需要150小时,或者6.25天,太长了。 第二个选项是在stirng中包含JSON并使用Logstash将字符串注入ElasticSearch时停止。但是,我有一个例外:

2019-12-16 13:49:27,240 | Timer-0 | ERROR | search-injector | c.n.es.injector.output.SocketOutput | SocketOutput::output: 
java.net.SocketException: Software caused connection abort: socket write error
at java.net.SocketOutputStream.socketWrite0(Native Method)
at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
at java.net.SocketOutputStream.write(SocketOutputStream.java:134)
at java.io.DataOutputStream.writeBytes(DataOutputStream.java:276)
at com.beniregev.es.injector.output.SocketOutput.output(SocketOutput.java:39)
at com.beniregev.es.injector.policies.UpdateOutputHandlers.run(UpdateOutputHandlers.java:60)
at java.util.TimerThread.mainLoop(Timer.java:555)
at java.util.TimerThread.run(Timer.java:505)

ElasticSearchlocalhost端口9200上运行,Logstashlocalhost端口9600上运行。 我的袜子发出来了。爪哇

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;

public class SocketOutput implements OutputHandler {

    private static final Logger log = LoggerFactory.getLogger(SocketOutput.class);

    public static final String CLI_OPTION = "socket";

    @Value("${socket.hostname}")
    private String hostname;
    @Value("${socket.port}")
    private int port;
    Socket clientSocket;

    public boolean open() {
        try {
            clientSocket = new Socket(hostname, port);
        } catch (IOException ioe) {
            log.error("", ioe);
            return false;
        }
        return true;
    }

    @Override
    public void output(String data) {

        DataOutputStream outToServer = null;
        try {
            outToServer = new DataOutputStream(clientSocket.getOutputStream());
            outToServer.writeBytes( data );
        } catch (IOException ioe) {
            log.error("", ioe);
        }
    }

}

logstash很简单。形态

# Simple Logstash configuration for creating a simple
# Stdin -> Logstash -> Elasticsearch pipeline.
input { stdin { } }

output {
  elasticsearch { 
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }
  stdout { codec => rubydebug }
}

我正在使用以下命令运行Logstash:bin/logstash.bat -f config/logstash-simple.conf

JSON字符串已创建且有效,Socket将获得正确的参数值(hostname=“localhost”和port=9600)。 我将感谢任何帮助


共 (1) 个答案

  1. # 1 楼答案

    解决方案有几个层次,首先使用Kibana管理器在Elastic Search中创建并使用正确的索引,然后正确配置Logstash,最后使用JSON字符串而不是文件

    创建索引不是这个问题的问题,所以我不会深入讨论

    Logstash。配置文件:

    #################################################
    # Stdin -> Logstash -> Elastic Search pipeline.
    #################################################
    input {
        stdin{}
        tcp{
            host => "localhost"
            port => 9600
            codec => json
        }
    }
    
    filter 
    {
       mutate 
        {
            remove_field => ["host", "@version", "@timestamp", "port", "tags", "level", "logger_name", "themessage", "mensage", "spring.application.name", "level_value", "thread_name"]
        }
    }
    
    output {
        stdout{ codec => rubydebug }
    
         elasticsearch{
            hosts => ["localhost:9200"]
            index => ["my-index"]
         }
    }
    

    注意:index中输入您创建并将使用的索引的名称

    正在使用配置文件运行Logstashbin/logstash.bat -f config/logstash-simple.conf

    将字符串输出到Java中的Logstash:

    package com.beniregev.injector.output;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    
    import java.io.DataOutputStream;
    import java.io.IOException;
    import java.net.Socket;
    
    public class SocketOutput implements OutputHandler {
    
        private static final Logger log = LoggerFactory.getLogger(SocketOutput.class);
    
        public static final String CLI_OPTION = "socket";
        private int outputIndex = 0;
    
        @Value("${socket.hostname}")
        private String hostname;
        @Value("${socket.port}")
        private int port;
        Socket clientSocket;
    
        public boolean open() {
            try {
                clientSocket = new Socket(hostname, port);
            } catch (IOException ioe) {
                log.error("", ioe);
                return false;
            }
            return true;
        }
    
        @Override
        public void output(String data) {
            DataOutputStream outToServer = null;
            try {
                outToServer = new DataOutputStream(clientSocket.getOutputStream());
                outToServer.writeBytes( data );
                outputIndex++;
            } catch (IOException ioe) {
                log.error("", ioe);
            }
            System.out.println("Wrote segment " + outputIndex + " to socket");
    
        }
    
    }
    

    hostport是到Logstash,默认端口为9600

    这为我解决了问题