正文
logstash1 - kafka - logstash2 - elasticsearch - kibana
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
0.拓扑图
参考：https://www.cnblogs.com/JetpropelledSnake/p/10057545.html
官网： http://kafka.apache.org/documentation.html#introduction kafka原理
https://www.cnblogs.com/jixp/articles/9435699.html Kafka全解析 https://blog.csdn.net/vinfly_li/article/details/79397201
1.logstash的配置
[root@VM_0_4_centos config]# cat wxqyh.yml|egrep -v '^$|^#'
input {
file {
type => "4personal20001"
path => "/mnt/data/logs/personal_20001/log4j.log"
start_position => "beginning"
sincedb_path => "/dev/null"
codec => multiline {
pattern => "^%{TIMESTAMP_ISO8601}"
negate => true
what => "previous"
}
}
file {
type => "4personal20002"
path => "/mnt/data/logs/personal_20002/log4j.log"
start_position => "beginning"
sincedb_path => "/dev/null"
codec => multiline {
pattern => "^%{TIMESTAMP_ISO8601}"
negate => true
what => "previous"
}
}
}
filter { grok {
match => {
"message" => "^%{TIMESTAMP_ISO8601}\[%{WORD:level} %{GREEDYDATA:ajpcon}\| %{GREEDYDATA:data}"
}
match => {
"message" => "^%{TIMESTAMP_ISO8601}\[ %{WORD:level} %{GREEDYDATA:ajpcon}\| %{GREEDYDATA:data}"
}
remove_field => "message"
}
}
output {
if [type] == "4personal20001" {
kafka {
max_request_size => 10485761
bootstrap_servers => "10.0.0.134:9092"
topic_id => "topic4personal1"
compression_type => "snappy"
}
}
if [type] == "4personal20002" {
kafka {
max_request_size => 10485761
bootstrap_servers => "10.0.0.134:9092"
topic_id => "topic4personal2"
compression_type => "snappy"
}
}
}
2.kafka的配置
[root@VM_0_134_centos config]# cat server.properties|egrep -v '^$|^#'
broker.id=
listeners=PLAINTEXT://10.0.0.134:9092
advertised.listeners=PLAINTEXT://10.0.0.134:9092
num.network.threads=
num.io.threads=
socket.send.buffer.bytes=
socket.receive.buffer.bytes=
socket.request.max.bytes=
message.max.bytes=
replica.fetch.max.bytes=
log.dirs=/mnt/data/monitor01/kafka_2.-1.1./data
num.partitions=
num.recovery.threads.per.data.dir=
offsets.topic.replication.factor=
transaction.state.log.replication.factor=
transaction.state.log.min.isr=
log.retention.hours=
log.segment.bytes=
log.retention.check.interval.ms=
zookeeper.connect=localhost:
zookeeper.connection.timeout.ms=
group.initial.rebalance.delay.ms=
3.logstash2的配置
[root@VM_0_134_centos config]# cat haode.yml|egrep -v '^$|^#'
input {
kafka {
type => "topic12wxqyh8"
codec => "plain"
topics => ["topic12wxqyh8"]
client_id => "es1"
group_id => "es1"
bootstrap_servers => "10.0.0.134:9092"
} kafka {
type => "topic12wxqyh9"
codec => "plain"
topics => ["topic12wxqyh9"]
client_id => "es2"
group_id => "es2"
bootstrap_servers => "10.0.0.134:9092"
}
kafka {
type => "topic24wxqyh6"
codec => "plain"
topics => ["topic24wxqyh6"]
client_id => "es3"
group_id => "es3"
bootstrap_servers => "10.0.0.134:9092"
} kafka {
type => "topic24wxqyh7"
codec => "plain"
topics => ["topic24wxqyh7"]
client_id => "es4"
group_id => "es4"
bootstrap_servers => "10.0.0.134:9092"
} kafka {
type => "topic4personal1"
codec => "plain"
topics => ["topic4personal1"]
client_id => "es5"
group_id => "es5"
bootstrap_servers => "10.0.0.134:9092"
}
kafka {
type => "topic4personal2"
codec => "plain"
topics => ["topic4personal2"]
client_id => "es6"
group_id => "es6"
bootstrap_servers => "10.0.0.134:9092"
} } output {
if [type] == "topic12wxqyh8" {
elasticsearch {
index => "topic12wxqyh8"
hosts => ["10.0.0.7:9200"]
}
} if [type] == "topic12wxqyh9" {
elasticsearch {
index => "topic12wxqyh9"
hosts => ["10.0.0.7:9200"]
}
}
if [type] == "topic24wxqyh6" {
elasticsearch {
index => "topic24wxqyh6"
hosts => ["10.0.0.7:9200"]
}
}
if [type] == "topic24wxqyh7" {
elasticsearch {
index => "topic24wxqyh7"
hosts => ["10.0.0.7:9200"]
}
}
if [type] == "topic4personal1" {
elasticsearch {
index => "topic4personal1"
hosts => ["10.0.0.7:9200"]
}
}
if [type] == "topic4personal2" {
elasticsearch {
index => "topic4personal2"
hosts => ["10.0.0.7:9200"]
}
}
}
参考：https://www.cnblogs.com/swordfall/p/8860941.html#auto_id_4
4.kafka的问题：
4.1zookeeper--选主，提供访问入口
4.2 Kafka的原理
生产者将数据发送到Broker代理，Broker代理有多个话题topic，消费者从Broker获取数据。
参考： https://zhuanlan.zhihu.com/p/97030680
5.partion--相当于es的shard
