如何监视kafka消息传递？

希望监视kafka消息传递度量-主要是消耗或丢失了多少消息、延迟、最终消耗者偏移量等等。。。
找到了apachespark，但我真的不明白，它是否可以“轻松地插入kafka”，而无需编写任何（java）代码。
我希望只是在一些配置文件中更改kafka连接的配置，在其中存储消息传递度量，并在一些ui中显示和查询它。
有没有比从头开始编写（java）应用程序更“自动化”的方法来使用spark-kafka消息传递监控？
是否有一些已经完成的应用程序，可以自由使用？
如何以最简单和自动化的方式设置kafka-spark-（数据库？）-ui？
spark还可以监视其他指标—集群、代理、主题、insync副本、分区数。。。等等？

经过不同软件的实验，Kafka显示器显示最适合我的情况。
它通过整个集群上的分区创建主题，并提供有关生产和消费的度量，如消息丢失、复制和端到端延迟。
像这样启动Kafka监视器：

cd <path_to_kafka_monitor>
export JMX_PORT=<some_port>
./bin/kafka-monitor-start.sh config/kafka-monitor.properties

使用jmxtrans收集度量。使用的查询：

{
    "servers": [{
        "port": "9999",
        "host": "127.0.0.1",
        "queries": [{
                "obj": "kafka.consumer:type=consumer-coordinator-metrics,client-id=*",
                "attr": ["assigned-partitions","commit-latency-avg","commit-rate","heartbeat-rate","heartbeat-response-time-max",
                        "join-rate","join-time-avg","join-time-max","last-heartbeat-seconds-ago","sync-rate","sync-time-avg","sync-time-max"],
                "resultAlias": "monitor-consumer.consumer-coordinator-metrics;client-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id"]
                }]
            },
            {
                "obj": "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*",
                "attr": ["bytes-consumed-rate","fetch-latency-avg","fetch-latency-max","records-consumed-rate","records-lag-max",
                        "records-per-request-avg"],
                "resultAlias": "monitor-consumer.consumer-fetch-manager-metrics,client-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id"]
                }]
            },
            {
                "obj": "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,topic=*",
                "attr": ["bytes-consumed-rate","fetch-size-avg","fetch-size-max","records-consumed-rate","records-per-request-avg"],
                "resultAlias": "monitor-consumer-fetch-manager-metrics;client-id,topic",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id","topic"]
                }]
            },
            {
                "obj": "kafka.consumer:type=consumer-metrics,client-id=*",
                "attr": [],
                "resultAlias": "monitor-consumer-metrics;client-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id"]
                }]
            },
            {
                "obj": "kafka.consumer:type=consumer-node-metrics,client-id=*,node-id=*",
                "attr": ["incoming-byte-rate","outgoing-byte-rate","request-latency-avg","request-rate","request-size-avg","response-rate"],
                "resultAlias": "monitor-consumer-node-metrics;client-id,node-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id","node-id"]
                }]
            },
            {
                "obj": "kafka.producer:type=producer-metrics,client-id=*",
                "attr": [],
                "resultAlias": "monitor-producer-metrics;client-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id"]
                }]
            },
            {
                "obj": "kafka.producer:type=producer-node-metrics,client-id=*,node-id=*",
                "attr": ["incoming-byte-rate","outgoing-byte-rate","request-latency-avg","request-rate","request-size-avg","response-rate"],
                "resultAlias": "monitor-producer-node-metrics;client-id,node-id",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id","node-id"]
                }]
            },
            {
                "obj": "kafka.producer:type=producer-topic-metrics,client-id=*,topic=*",
                "attr": [],
                "resultAlias": "monitor-producer-topic-metrics;client-id,topic",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor",
                    "typeNames": ["client-id","topic"]
                }]
            },
            {
                "obj": "kmf.services:type=consume-service,name=single-cluster-monitor",
                "attr": [],
                "resultAlias": "monitor-consume-service.single-cluster-monitor",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor"
                }]
            },
            {
                "obj": "kmf.services:type=produce-service,name=single-cluster-monitor",
                "attr": [],
                "resultAlias": "monitor-produce-service.single-cluster-monitor",
                "outputWriters": [{
                    "@class": "com.googlecode.jmxtrans.model.output.InfluxDbWriterFactory",
                    "url": "http://127.0.0.1:8086/",
                    "username": "root",
                    "password": "root",
                    "database": "monitor"
                }]
            }
        ]
    }]
}

如何监视kafka消息传递？

2条答案

相关问题

热门标签

最新问答