ELK学习6_Kafka->Logstash->Elasticsearch数据流操作

淩亂°似流年 2023-10-17 13:55 125阅读 0赞

## Logstash配置过程 ##

在Logstash中建立input和output的条件：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ cd /usr/local/ 
2.  \[hadoop@Slave1 local\]$ cd logstash/ 
3.  \[hadoop@Slave1 logstash\]$ ls 
4.  bin CONTRIBUTORS Gemfile.jruby-1.9.lock LICENSE vendor 
5.  CHANGELOG.md Gemfile lib NOTICE.TXT 
6.  \[hadoop@Slave1 logstash\]$ mkdir -p conf 
7.  \[hadoop@Slave1 logstash\]$ ls 
8.  bin conf Gemfile lib NOTICE.TXT 
9.  CHANGELOG.md CONTRIBUTORS Gemfile.jruby-1.9.lock LICENSE vendor 
10. \[hadoop@Slave1 logstash\]$ cd conf 
11. \[hadoop@Slave1 conf\]$ ls 
12. \[hadoop@Slave1 conf\]$ touch kafkaInput\_esOutPut.conf 
13. \[hadoop@Slave1 conf\]$ ls 
14. kafkaInput\_esOutPut.conf 
15. \[hadoop@Slave1 conf\]$ 
16. \[hadoop@Slave1 conf\]$ vim kafkaInput\_esOutPut.conf

对kafkaInput\_esOutPut.conf进行编辑，本机的具体内容如下：

**\[html\]** [view plain][] [copy][view plain]

1.  input \{ 
2.  kafka \{ 
3.  zk\_connect =**>** "192.168.154.158:2181,192.168.154.159:2181,192.168.154.160:2181" 
4.  group\_id =**>** "test-consumer-group" 
5.  topic\_id =**>** "logStash" 
6.  reset\_beginning =**>** false \# boolean (optional)， default: false 
7.  consumer\_threads =**>** 5 \# number (optional)， default: 1 
8.  decorate\_events =**>** true \# boolean (optional)， default: false 
9.   \} 
10. \} 
11. 
12. filter\{ 
13.  mutate\{ 
14.  \#以:号分割message内容，分割后以数据方式显示。 
15.  \#比如abc:efg =**>** message\[0\] = abc message\[1\]=efg 
16. split =**>** \["message",","\] 
17.  \} 
18.  \#第一个数据的内容中ORA-xxxxx这种格式，则这条内容是ora错误。添加二个字段 
19.  mutate\{ 
20. add\_field =**>** \{ 
21.  "source\_Ip" =**>** "%\{\[message\]\[0\]\}" 
22.  "source\_Port" =**>** "%\{\[message\]\[1\]\}" 
23.  "dest\_Ip" =**>** "%\{\[message\]\[2\]\}" 
24.  "dest\_Port" =**>** "%\{\[message\]\[3\]\}" 
25.  \} 
26.  \} 
27. \} 
28. 
29. output \{ 
30.  elasticsearch \{ 
31. 
32. host =**>** "localhost" 
33.  \} 
34. \}

分别ssh链接Slave2和Slave3，将kafkaInput\_esOutPut.conf拷贝到这两台机器上：

创建conf目录过程：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 conf\]$ ssh Slave2 
2.  Last login: Wed Oct 14 10:58:06 2015 from slave1 
3.  \[hadoop@Slave2 ~\]$ cd /usr/local/logstash/ 
4.  \[hadoop@Slave2 logstash\]$ mkdir -p conf 
5.  \[hadoop@Slave2 logstash\]$ ls 
6.  bin conf Gemfile lib NOTICE.TXT 
7.  CHANGELOG.md CONTRIBUTORS Gemfile.jruby-1.9.lock LICENSE vendor 
8.  \[hadoop@Slave2 logstash\]$ exit 
9.  logout 
10. Connection to Slave2 closed. 
11. \[hadoop@Slave1 conf\]$ ssh Slave3 
12. Last login: Wed Oct 14 10:59:01 2015 from slave2 
13. \[hadoop@Slave3 ~\]$ cd /usr/local/logstash/ 
14. \[hadoop@Slave3 logstash\]$ mkdir -p conf 
15. \[hadoop@Slave3 logstash\]$ ls 
16. bin conf Gemfile lib NOTICE.TXT 
17. CHANGELOG.md CONTRIBUTORS Gemfile.jruby-1.9.lock LICENSE vendor 
18. \[hadoop@Slave3 logstash\]$ exit 
19. logout 
20. Connection to Slave3 closed.

传输文件过程：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 conf\]$ scp kafkaInput\_esOutPut.conf Slave2:/usr/local/logstash/conf/ 
2.  kafkaInput\_esOutPut.conf 100% 1063 1.0KB/s 00:00 
3.  \[hadoop@Slave1 conf\]$ scp kafkaInput\_esOutPut.conf Slave3:/usr/local/logstash/conf/ 
4.  kafkaInput\_esOutPut.conf 100% 1063 1.0KB/s 00:00 
5.  \[hadoop@Slave1 conf\]$ ssh Slave2 
6.  Last login: Tue Oct 27 23:46:19 2015 from slave1 
7.  \[hadoop@Slave2 ~\]$ cd /usr/local/logstash/conf/ 
8.  \[hadoop@Slave2 conf\]$ ls 
9.  kafkaInput\_esOutPut.conf 
10. \[hadoop@Slave2 conf\]$

## Kafka操作过程 ##

在三台机器上启动zookeeper：

关闭防火墙：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 bin\]$ su 
2.  Password: 
3.  \[root@Slave1 bin\]\# service iptables stop 
4.  iptables: Setting chains to policy ACCEPT: filter \[ OK \] 
5.  iptables: Flushing firewall rules: \[ OK \] 
6.  iptables: Unloading modules: \[ OK \] 
7.  \[root@Slave1 bin\]\# exit 
8.  exit 
9.  \[hadoop@Slave1 bin\]

启动：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 bin\]$ ./zkServer.sh start 
2.  JMX enabled by default 
3.  Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg 
4.  Starting zookeeper ... STARTED

在其他三台机器上进行相同操作后，查看结果：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 bin\]$ ./zkServer.sh status 
2.  JMX enabled by default 
3.  Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg 
4.  Mode: leader 
5.  \[hadoop@Slave1 bin\]$

在三台机器上启动Kafka，以Slave1为例：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 bin\]$ cd /usr/local/kafka/ 
2.  \[hadoop@Slave1 kafka\]$ bin/kafka-server-start.sh config/server.properties

新建名为logStash的topic：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ cd /usr/local/kafka/ 
2.  \[hadoop@Slave1 kafka\]$ cd bin 
3.  \[hadoop@Slave1 bin\]$ sh kafka-topics.sh --create --topic logStash --replication-factor 1 --partitions 1 --zookeeper Slave1:2181 
4.  Created topic "logStash". 
5.  \[hadoop@Slave1 bin\]$

启动Logstash

在三台机器上，进行启动：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ cd /usr/local/logstash/ 
2.  \[hadoop@Slave1 logstash\]$ ls 
3.  bin conf Gemfile lib NOTICE.TXT 
4.  CHANGELOG.md CONTRIBUTORS Gemfile.jruby-1.9.lock LICENSE vendor 
5.  \[hadoop@Slave1 logstash\]$ cd bin 
6.  \[hadoop@Slave1 bin\]$ ls 
7.  logstash logstash.lib.sh plugin.bat rspec.bat 
8.  logstash.bat plugin rspec setup.bat

启动过程中，显示的内容如下，会出现一些警告：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave2 bin\]$ ./logstash agent -f ../conf/kafkaInput\_esOutPut.conf 
2.  log4j, \[2015-10-28T21:52:07.116\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-3 for topic logStash 
3.  log4j, \[2015-10-28T21:52:07.118\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-2 for topic logStash 
4.  log4j, \[2015-10-28T21:52:07.119\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-0 for topic logStash 
5.  log4j, \[2015-10-28T21:52:07.119\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-4 for topic logStash 
6.  log4j, \[2015-10-28T21:52:07.120\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-1 for topic logStash 
7.  log4j, \[2015-10-28T21:52:33.934\] WARN: org.elasticsearch.bootstrap: JNA not found. native methods will be disabled. 
8.  log4j, \[2015-10-28T21:53:09.347\] WARN: org.elasticsearch.discovery: \[logstash-Slave2-4244-11624\] waited for 30s and no initial state was set by the discovery 
9.  log4j, \[2015-10-28T21:53:35.632\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-3 for topic logStash 
10. log4j, \[2015-10-28T21:53:35.633\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-2 for topic logStash 
11. log4j, \[2015-10-28T21:53:35.634\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-0 for topic logStash 
12. log4j, \[2015-10-28T21:53:35.634\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-4 for topic logStash 
13. log4j, \[2015-10-28T21:53:35.634\] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread test-consumer-group\_Slave2-1446094310356-56dfbfa7-1 for topic logStash 
14. Failed to install template: waited for \[30s\] \{ :level=**>**:error\} 
15. Logstash startup completed

发送并接收数据

启动刚才建立的topic：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ cd /usr/local/kafka/ 
2.  \[hadoop@Slave1 kafka\]$ ls 
3.  bin config libs LICENSE logs NOTICE 
4.  \[hadoop@Slave1 kafka\]$ bin/kafka-console-producer.sh --broker-list Slave1:9092 --topic logStash

启动ES：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ cd /usr/local/elasticsearch/ 
2.  \[hadoop@Slave1 elasticsearch\]$ bin/elasticsearch -f 
3.  getopt: invalid option -- 'f' 
4.  \[2015-10-29 00:47:27,084\]\[INFO \]\[node \] \[Clown\] version\[1.7.3\], pid\[5208\], build\[05d4530/2015-10-15T09:14:17Z\] 
5.  \[2015-10-29 00:47:27,131\]\[INFO \]\[node \] \[Clown\] initializing ... 
6.  \[2015-10-29 00:47:27,920\]\[INFO \]\[plugins \] \[Clown\] loaded \[\], sites \[\] 
7.  \[2015-10-29 00:47:28,548\]\[INFO \]\[env \] \[Clown\] using \[1\] data paths, mounts \[\[/ (/dev/sda2)\]\], net usable\_space \[9.7gb\], net total\_space \[17.4gb\], types \[ext4\] 
8.  \[2015-10-29 00:47:43,711\]\[INFO \]\[node \] \[Clown\] initialized 
9.  \[2015-10-29 00:47:43,729\]\[INFO \]\[node \] \[Clown\] starting ... 
10. \[2015-10-29 00:47:46,089\]\[INFO \]\[transport \] \[Clown\] bound\_address \{inet\[/0:0:0:0:0:0:0:0:9301\]\}, publish\_address \{inet\[/192.168.154.158:9301\]\} 
11. \[2015-10-29 00:47:46,606\]\[INFO \]\[discovery \] \[Clown\] elasticsearch/v-jkBhkxSheape14hvMAHw 
12. \[2015-10-29 00:47:50,712\]\[INFO \]\[cluster.service \] \[Clown\] new\_master \[Clown\]\[v-jkBhkxSheape14hvMAHw\]\[Slave1\]\[inet\[/192.168.154.158:9301\]\], reason: zen-disco-join (elected\_as\_master) 
13. \[2015-10-29 00:47:50,985\]\[INFO \]\[http \] \[Clown\] bound\_address \{inet\[/0:0:0:0:0:0:0:0:9200\]\}, publish\_address \{inet\[/192.168.154.158:9200\]\} 
14. \[2015-10-29 00:47:50,986\]\[INFO \]\[node \] \[Clown\] started 
15. \[2015-10-29 00:47:51,345\]\[INFO \]\[gateway \] \[Clown\] recovered \[0\] indices into cluster\_state 
16. \[2015-10-29 00:47:51,346\]\[INFO \]\[cluster.service \] \[Clown\] added \{\[logstash-Slave1-4083-11624\]\[loTUXdCXRVC\_WzqzhD3PWg\]\[Slave1\]\[inet\[/192.168.154.158:9300\]\]\{ data=false, client=true\},\}, reason: zen-disco-receive(join from node\[\[logstash-Slave1-4083-11624\]\[loTUXdCXRVC\_WzqzhD3PWg\]\[Slave1\]\[inet\[/192.168.154.158:9300\]\]\{ data=false, client=true\}\]) 
17. \[2015-10-29 00:47:54,185\]\[INFO \]\[cluster.metadata \] \[Clown\] \[logstash-2015.10.29\] creating index, cause \[auto(bulk api)\], templates \[\], shards \[5\]/\[1\], mappings \[logs\] 
18. \[2015-10-29 00:47:56,201\]\[INFO \]\[cluster.metadata \] \[Clown\] \[logstash-2015.10.29\] update\_mapping \[logs\] (dynamic) 
19. \[2015-10-29 00:47:57,166\]\[INFO \]\[cluster.metadata \] \[Clown\] \[logstash-2015.10.29\] update\_mapping \[logs\] (dynamic)

检查ES是否启动成功：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ curl -X GET http://localhost:9200 
2.  \{ 
3.   "status" : 200, 
4.   "name" : "Clown", 
5.   "cluster\_name" : "elasticsearch", 
6.   "version" : \{ 
7.   "number" : "1.7.3", 
8.   "build\_hash" : "05d4530971ef0ea46d0f4fa6ee64dbc8df659682", 
9.   "build\_timestamp" : "2015-10-15T09:14:17Z", 
10.  "build\_snapshot" : false, 
11.  "lucene\_version" : "4.10.4" 
12.  \}, 
13.  "tagline" : "You Know, for Search" 
14. \} 
15. \[hadoop@Slave1 ~\]$

在刚才启动的topic里发送数据：

（数据的格式是源IP，源端口，目的IP，目的端口；为了简便，发送1,1,1,1）

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 kafka\]$ bin/kafka-console-producer.sh --broker-list Slave1:9092 --topic logStash 
2.  \[2015-10-29 00:39:33,085\] WARN Property topic is not valid (kafka.utils.VerifiableProperties) 
3.  1,1,1,1

查看接收的数据：

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ curl -XGET 'localhost:9200/logstash-2015.10.27/\_search' 
2.  \{"error":"IndexMissingException\[\[logstash-2015.10.27\] missing\]","status":404\}\[hadoop@Slave1 ~\]$ curl -XGET 'localhost:9200/logstash-2015.10.29/\_search' 
3.  \{"took":260,"timed\_out":false,"\_shards":\{"total":5,"successful":5,"failed":0\},"hits":\{"total":2,"max\_score":1.0,"hits":\[\{"\_index":"logstash-2015.10.29","\_type":"logs","\_id":"AVCykUgg6gAQTB\_SuF\_V","\_score":1.0,"\_source":\{"message":\["1","1","1","1"\],"tags":\["\_jsonparsefailure"\],"@version":"1","@timestamp":"2015-10-29T07:39:50.871Z","kafka":\{"msg\_size":7,"topic":"logStash","consumer\_group":"test-consumer-group","partition":0,"key":null\},"source\_Ip":"1","source\_Port":"1","dest\_Ip":"1","dest\_Port":"1"\}\},\{"\_index":"logstash-2015.10.29","\_type":"logs","\_id":"AVCykUGv6gAQTB\_SuF\_U","\_score":1.0,"\_source":\{"message":\[\],"tags":\["\_jsonparsefailure"\],"@version":"1","@timestamp":"2015-10-29T07:39:46.345Z","kafka":\{"msg\_size":0,"topic":"logStash","consumer\_group":"test-consumer-group","partition":0,"key":null\},"source\_Ip":"%\{\[message\]\[0\]\}","source\_Port":"%\{\[message\]\[1\]\}","dest\_Ip":"%\{\[message\]\[2\]\}","dest\_Port":"%\{\[message\]\[3\]\}"\}\}\]\}\}\[hadoop@Slave1 ~\]$

**\[html\]** [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 ~\]$ curl -XGET 'localhost:9200/logstash-2015.10.29/\_search?pretty' 
2.  \{ 
3.   "took" : 26, 
4.   "timed\_out" : false, 
5.   "\_shards" : \{ 
6.   "total" : 5, 
7.   "successful" : 5, 
8.   "failed" : 0 
9.   \}, 
10.  "hits" : \{ 
11.  "total" : 2, 
12.  "max\_score" : 1.0, 
13.  "hits" : \[ \{ 
14.  "\_index" : "logstash-2015.10.29", 
15.  "\_type" : "logs", 
16.  "\_id" : "AVCykUgg6gAQTB\_SuF\_V", 
17.  "\_score" : 1.0, 
18.  "\_source":\{"message":\["1","1","1","1"\],"tags":\["\_jsonparsefailure"\],"@version":"1","@timestamp":"2015-10-29T07:39:50.871Z","kafka":\{"msg\_size":7,"topic":"logStash","consumer\_group":"test-consumer-group","partition":0,"key":null\},"source\_Ip":"1","source\_Port":"1","dest\_Ip":"1","dest\_Port":"1"\} 
19.  \}, \{ 
20.  "\_index" : "logstash-2015.10.29", 
21.  "\_type" : "logs", 
22.  "\_id" : "AVCykUGv6gAQTB\_SuF\_U", 
23.  "\_score" : 1.0, 
24.  "\_source":\{"message":\[\],"tags":\["\_jsonparsefailure"\],"@version":"1","@timestamp":"2015-10-29T07:39:46.345Z","kafka":\{"msg\_size":0,"topic":"logStash","consumer\_group":"test-consumer-group","partition":0,"key":null\},"source\_Ip":"%\{\[message\]\[0\]\}","source\_Port":"%\{\[message\]\[1\]\}","dest\_Ip":"%\{\[message\]\[2\]\}","dest\_Port":"%\{\[message\]\[3\]\}"\} 
25.  \} \] 
26.  \} 
27. \} 
28. \[hadoop@Slave1 ~\]$

参考资料：

[http://blog.csdn.net/xuguokun1986/article/details/49452101][http_blog.csdn.net_xuguokun1986_article_details_49452101]

对这篇博客的内容进行了扩展。

来源:http://blog.csdn.net/wang\_zhenwei/article/details/49493131

[view plain]: http://blog.csdn.net/wang_zhenwei/article/details/49493131#
[http_blog.csdn.net_xuguokun1986_article_details_49452101]: http://blog.csdn.net/xuguokun1986/article/details/49452101