1、使用Elasticsearch作为主数据存储实践
http://t.cn/EyS6wcL
2、Elasticsearch 6.x启动过程
http://t.cn/EyJgQ7U
3、Elasticsearch你知道多少？
http://t.cn/EyS6I4P

编辑：铭毅天下
归档: https://elasticsearch.cn/article/6177
订阅：https://tinyletter.com/elastic-daily

Logstash 在 6.0 推出了 multiple pipeline 的解决方案，即在一个 logstash 实例中可以同时进行多个独立数据流程的处理工作，如下图所示。


而在这之前用户只能通过在单机运行多个 logstash 实例或者在配置文件中增加大量 if-else 条件判断语句来解决。要使用 multiple pipeline 也很简单，只需要将不同的 pipeline 在 config/pipeline.yml中定义好即可，如下所示：

```yaml

• pipeline.id: apache
  pipeline.batch.size: 125
  queue.type: persisted
  path.config: "/path/to/config/apache.cfg"
  
• pipeline.id: nginx
  path.config: "/path/to/config/nginx.cfg"
  ``<br /> <br /> 其中apache和nginx作为独立的 pipeline 执行，而且配置也可以独立设置，互不干扰。pipeline.yml的引入极大地简化了 logstash 的配置管理工作，使得新手也可以很快完成复杂的 ETL 配置。<br /> <br /> 在 6.3 版本中，Logstash 又增加了Pipeline-to-Pipeline的管道机制（beta），即管道和管道之间可以连接在一起组成一个完成的数据处理流。熟悉 linux 的管道命令|`的同学应该可以很快明白这种模式的好处。这无疑使得 Logstash 的配置会更加灵活，今天我们就来了解下这种灵活自由的配置方式。
  
  
  
  

  1. 上手
  

  
  废话少说，快速上手。修改 config/pipeline.yml文件如下：
  
  ```yaml
  

  • pipeline.id: upstream
    config.string: input { stdin {} } output { pipeline { send_to => [test_output] } }
    
  • pipeline.id: downstream
    config.string: input { pipeline { address => test_output } } output{ stdout{}}
    ``<br /> <br /> <br /> <br /> 然后运行 logstash，其中-r` 表示配置文件有改动时自动重新加载，方便我们调试。
    
    

    bin/logstash -r
    
    在终端随意输入字符(比如aaa)后回车，会看到屏幕输出了类似下面的内容，代表运行成功了。
    
    json<br /> {<br /> "@timestamp" => 2018-12-06T14:43:50.310Z,<br /> "@version" => "1",<br /> "message" => "aaa",<br /> "host" => "rockybean-MacBook-Pro.local"<br /> }<br />
    
    我们再回头看下这个配置，upstreamoutput 使用了名为 pipeline 的 plugin，然后 send_to的输出对象test_output是在 downstream的 input pipeline plugin 中定义的。通过这个唯一的address(虚拟地址)就能够把不同的 pipeline 连接在一起组成一个更长的pipeline来处理数据。类似下图所示：
    
    
    
    
    
    当数据由 upstream传递给 downstream时会进行一个复制操作，这也意味着在这两个 pipeline 中的数据是完全独立的，互不影响。有一点要注意的是：数据的复制会增加额外的性能开销，比如会加大 JVM Heap 的使用。
    
    

    2. 使用场景
    

    
    使用方法是不是很简单，接下来我们来看下官方为我们开的几个脑洞。
    
    

    2.1 Distributor Pattern 分发者模式
    

    
    该模式执行效果类似下图所示：
    
    
    
    在一个 pipeline 处理输入，然后根据不同的数据类型再分发到对应的 Pipeline 去处理。这种模式的好处在于统一输入端口，隔离不同类型的处理配置文件，减少由于配置文件混合在一起带来的维护成本。大家可以想一想如果不用这种Pipeline-to-Pipeline的方式，我们如果轻松做到一个端口处理多个来源的数据呢？
    
    这种模式的参考配置如下所示：
    
    ```yaml
    

    config/pipelines.yml
    

• pipeline.id: beats-server
  config.string: |
  input { beats { port => 5044 } }
  output {
  if [type] == apache {
  pipeline { send_to => weblogs }
  } else if [type] == system {
  pipeline { send_to => syslog }
  } else {
  pipeline { send_to => fallback }
  }
  }
  
• pipeline.id: weblog-processing
  config.string: |
  input { pipeline { address => weblogs } }
  filter {
  

  Weblog filter statements here...
  

  }
  output {
  elasticsearch { hosts => [es_cluster_a_host] }
  }
  

• pipeline.id: syslog-processing
  config.string: |
  input { pipeline { address => syslog } }
  filter {
  

  Syslog filter statements here...
  

  }
  output {
  elasticsearch { hosts => [es_cluster_b_host] }
  }
  

• pipeline.id: fallback-processing
  config.string: |
  input { pipeline { address => fallback } }
  output { elasticsearch { hosts => [es_cluster_b_host] } }
  ```
  
  
  
  

  2.2 Output Isolator Pattern 输出隔离模式
  

  
  虽然 Logstash 的一个 pipeline 可以配置多个 output，但是这多个 output 会相依为命，一旦某一个 output 出问题，会导致另一个 output 也无法接收新数据。而通过这种模式可以完美解决这个问题。其运行方式如下图所示：
  
  
  
  通过输出到两个独立的 pipeline，解除相互之间的影响，比如 http service 出问题的时候，es 依然可以正常接收数据，而且两个 pipeline 可以配置独立的队列来保障数据的完备性，其配置如下所示：
  
  ```yaml
  

  config/pipelines.yml
  

• pipeline.id: intake
  queue.type: persisted
  config.string: |
  input { beats { port => 5044 } }
  output { pipeline { send_to => [es, http] } }
  
• pipeline.id: buffered-es
  queue.type: persisted
  config.string: |
  input { pipeline { address => es } }
  output { elasticsearch { } }
  
• pipeline.id: buffered-http
  queue.type: persisted
  config.string: |
  input { pipeline { address => http } }
  output { http { } }
  ```
  
  

  2.3 Forked Path Pattern 克隆路径模式
  

  
  这个模式类似 Output Isolator Pattern，只是在不同的 output pipeline 中可以配置不同的 filter 来完成各自输出的数据处理需求，这里就不展开讲了，可以参考如下的配置，其中不同 output pipeline 的 filter 是不同的，比如 partner 这个 pipeline 去掉了一些敏感数据：
  
  ```
  

  config/pipelines.yml
  

• pipeline.id: intake
  queue.type: persisted
  config.string: |
  input { beats { port => 5044 } }
  output { pipeline { send_to => ["internal-es", "partner-s3"] } }
  
• pipeline.id: buffered-es
  queue.type: persisted
  config.string: |
  input { pipeline { address => "internal-es" } }
  

  Index the full event
  

  output { elasticsearch { } }
  

• pipeline.id: partner
  queue.type: persisted
  config.string: |
  input { pipeline { address => "partner-s3" } }
  filter {
  

  Remove the sensitive data
  

  mutate { remove_field => 'sensitive-data' }
  }
  output { s3 { } } # Output to partner's bucket
  ```
  
  

  2.4 Collector Pattern 收集者模式
  

  
  从名字可以看出，该模式是将所有 Pipeline 汇集于一处的处理模式，如下图所示：
  
  
  
  其配置参考如下：
  
  ```
  

  config/pipelines.yml
  

• pipeline.id: beats
  config.string: |
  input { beats { port => 5044 } }
  output { pipeline { send_to => [commonOut] } }
  
• pipeline.id: kafka
  config.string: |
  input { kafka { ... } }
  output { pipeline { send_to => [commonOut] } }
  
• pipeline.id: partner
  

  This common pipeline enforces the same logic whether data comes from Kafka or Beats
  

  config.string: |
  input { pipeline { address => commonOut } }
  filter {
  

  Always remove sensitive data from all input sources
  

  mutate { remove_field => 'sensitive-data' }
  }
  output { elasticsearch { } }
  ```
  
  
  
  

  3. 总结
  

  
  本文简单给大家讲解了 Pipeline-to-Pipeline的使用方法及官方推荐的几种模式，希望可以给大家有所帮助。另外这个机制目前还处于 Beta 阶段，尝鲜需谨慎！
  
  

1.图解 Elasticsearch2.2.0 原理
http://t.cn/Eybu8a3
2.ElasticSearch 恢复类型之对等恢复
http://t.cn/EyaCYFV
3.使用带注释的文本插件搜索事物
http://t.cn/EyaCntg

编辑：金桥
归档：https://elasticsearch.cn/article/6175
订阅：https://tinyletter.com/elastic-daily

#10001-log
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/10001/logs/mstore.log
  fields:
    env: uat-10001-log
  include_lines: ['ERROR']
#10001-catalina
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/10001/logs/catalina.out
  fields:
    env: uat-10001-catalina
  include_lines: ['ERROR']
#11001-log
- type: log
  enabled: true
  paths:
    - /mkt/tomcat/8.5.32/11001/logs/delivery.log
  include_lines: ['ERROR']
  fields:
    env: uat-11001-log
#11001-catalina.out
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/11001/logs/catalina.out
  fields:
    env: uat-11001-catalina
  include_lines: ['ERROR']
#12001-log
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/12001/logs/mstore.log
  fields:
    env: uat-12001-log
  include_lines: ['ERROR']
#12001-catalina
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/12001/logs/catalina.out
  fields:
    env: uat-12001-catalina
  include_lines: ['ERROR']
#13001-log
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/13001/logs/pay-web-boss.log
  fields:
    env: uat-13001-log
    include_lines: ['ERROR']
#13001-catalina
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/13001/logs/catalina.out
  fields:
    env: uat-13001-catalina
  include_lines: ['ERROR']
#14001-log
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/14001/logs/pay-web-gateway.log
  fields:
    env: uat-14001-log
  include_lines: ['ERROR']
#14001-catalina
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/14001/logs/catalina.out
  fields:
    env: uat-14001-catalina
  include_lines: ['ERROR']
#15001-log
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/15001/logs/roncoo-pay-web-merchant.log
  fields:
    env: uat-15001-log
  include_lines: ['ERROR']
#15001-catalina
- type: log
  enable: true
  paths:
    - /mkt/tomcat/8.5.32/15001/logs/catalina.out
  fields:
    env: uat-15001-catalina
  include_lines: ['ERROR']      每次创建索引会少uat-11001-log和uat-12001-log,filebeat读取了这个两个日志文件