title: Logstash+Elasticsearch throughput
body: <p>we're trying to process 5K msgs/sec with 2 identical machines, but seems like we max out logstash or elasticsearch.</p>
  
  <p>Each has: 
  64Gb RAM, ≈3Ghz Xeon CPU
  Logstash 1.5 installed
  Elasticsearch 1.7.8 installed in cluster mode with second machine.</p>
  
  <p>Logstash is configured to receive messages from 16-node kafka cluster and send it to Elasticsearch.</p>
  
  <p>Data is CSV, contains 22 fields. Is that a normal throughput?</p>
  
  <p>Here's the config:</p>
  
  <pre><code>input{
      kafka {
          type =&gt; "api"
          zk_connect =&gt; "node1:2181,node2:2181,node3:2181"
          codec =&gt; "plain"
          topic_id =&gt; "api_events"
          consumer_threads =&gt; 8
          queue_size =&gt; 10000
          rebalance_backoff_ms =&gt; 10000
          rebalance_max_retries =&gt; 10
      }
  }
  filters{
      csv {
          separator =&gt; "::"
          columns =&gt; [
              "hostname",
              "status",
              "body_bytes_sent",
              "request_time",
              "http_x_forwarded_for",
              "uri",
              "arg_key",
              "http_user_agent",
              "http_deviceid",
              "http_country_code",
              "http_language_code",
              "http_platform",
              "http_versioncode",
              "request_method",
              "http_x_forwarded_proto",
              "upstream_cache_status",
              "upstream_response_time",
              "upstream_header_time",
              "upstream_status",
              "bytes_sent",
              "time_local",
              "upstream_addr"
          ]
          remove_field =&gt; [
              "message"
          ]
      }
      mutate {
          convert =&gt; {
              "body_bytes_sent" =&gt; "integer"
              "request_time" =&gt; "float"
              "upstream_response_time" =&gt; "float"
              "upstream_header_time" =&gt; "float"
              "bytes_sent" =&gt; "integer"
              }
          }
      }
  }
  output{
      elasticsearch {
          cluster =&gt; "MyCluster"
          protocol =&gt; "node"
          index =&gt; "api-%{+YYYY.MM.dd}"
          host =&gt; "elasticnode1"
          flush_size =&gt; 50000
          workers =&gt; 4
      }
  }
  </code></pre>